[ONOS-7942] Implement k8s ingress service and watcher with unit tests
Change-Id: I8089336916e882bd31b40facf73c46c5ee541718
diff --git a/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressAdminService.java b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressAdminService.java
new file mode 100644
index 0000000..5e2b95b
--- /dev/null
+++ b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressAdminService.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.api;
+
+import io.fabric8.kubernetes.api.model.extensions.Ingress;
+
+/**
+ * Service for administering the inventory of kubernetes ingress.
+ */
+public interface K8sIngressAdminService extends K8sIngressService {
+
+ /**
+ * Creates kubernetes ingress with the given information.
+ *
+ * @param ingress the new kubernetes ingress
+ */
+ void createIngress(Ingress ingress);
+
+ /**
+ * Updates the kubernetes ingress with the given information.
+ *
+ * @param ingress the updated kubernetes ingress
+ */
+ void updateIngress(Ingress ingress);
+
+ /**
+ * Removes the kubernetes ingress.
+ *
+ * @param uid kubernetes ingress UID
+ */
+ void removeIngress(String uid);
+
+ /**
+ * Clears the existing ingresses.
+ */
+ void clear();
+}
diff --git a/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressEvent.java b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressEvent.java
new file mode 100644
index 0000000..c6e92fc
--- /dev/null
+++ b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressEvent.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.api;
+
+import io.fabric8.kubernetes.api.model.extensions.Ingress;
+import org.onosproject.event.AbstractEvent;
+
+/**
+ * Describes kubernetes ingress event.
+ */
+public class K8sIngressEvent extends AbstractEvent<K8sIngressEvent.Type, Ingress> {
+
+ /**
+ * Creates an event of a given type for the specified kubernetes ingress.
+ *
+ * @param type kubernetes ingress event type
+ * @param subject kubernetes ingress
+ */
+ public K8sIngressEvent(Type type, Ingress subject) {
+ super(type, subject);
+ }
+
+ /**
+ * Kubernetes ingress events.
+ */
+ public enum Type {
+
+ /**
+ * Signifies that a new kubernetes ingress is created.
+ */
+ K8S_INGRESS_CREATED,
+
+ /**
+ * Signifies that the kubernetes ingress is updated.
+ */
+ K8S_INGRESS_UPDATED,
+
+ /**
+ * Signifies that the kubernetes ingress is removed.
+ */
+ K8S_INGRESS_REMOVED,
+ }
+}
diff --git a/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressListener.java b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressListener.java
new file mode 100644
index 0000000..0429ad3
--- /dev/null
+++ b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressListener.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.api;
+
+import org.onosproject.event.EventListener;
+
+/**
+ * Listener for kubernetes ingress event.
+ */
+public interface K8sIngressListener extends EventListener<K8sIngressEvent> {
+}
diff --git a/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressService.java b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressService.java
new file mode 100644
index 0000000..370dd79
--- /dev/null
+++ b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressService.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.api;
+
+import io.fabric8.kubernetes.api.model.extensions.Ingress;
+import org.onosproject.event.ListenerService;
+
+import java.util.Set;
+
+/**
+ * Service for interacting with the inventory of kubernetes ingress.
+ */
+public interface K8sIngressService
+ extends ListenerService<K8sIngressEvent, K8sIngressListener> {
+
+ /**
+ * Returns the kubernetes ingress with the supplied ingress UID.
+ *
+ * @param uid ingress UID
+ * @return kubernetes ingress
+ */
+ Ingress ingress(String uid);
+
+ /**
+ * Returns all kubernetes ingresses registered.
+ *
+ * @return set of kubernetes ingresses
+ */
+ Set<Ingress> ingresses();
+}
diff --git a/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressStore.java b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressStore.java
new file mode 100644
index 0000000..1025202
--- /dev/null
+++ b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressStore.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.api;
+
+import io.fabric8.kubernetes.api.model.extensions.Ingress;
+import org.onosproject.store.Store;
+
+import java.util.Set;
+
+/**
+ * Manages inventory of kubernetes ingresses; not intended for direct use.
+ */
+public interface K8sIngressStore
+ extends Store<K8sIngressEvent, K8sIngressStoreDelegate> {
+
+ /**
+ * Creates the new kubernetes ingress.
+ *
+ * @param ingress kubernetes ingress
+ */
+ void createIngress(Ingress ingress);
+
+ /**
+ * Updates the kubernetes ingress.
+ *
+ * @param ingress kubernetes ingress
+ */
+ void updateIngress(Ingress ingress);
+
+ /**
+ * Removes the kubernetes ingress with the given ingress UID.
+ *
+ * @param uid kubernetes ingress UID.
+ * @return removed kubernetes ingress; null if failed
+ */
+ Ingress removeIngress(String uid);
+
+ /**
+ * Returns the kubernetes ingress with the given ingress UID.
+ *
+ * @param uid kubernetes ingress UID
+ * @return kubernetes ingress; null if not found
+ */
+ Ingress ingress(String uid);
+
+ /**
+ * Returns all kubernetes ingresses.
+ *
+ * @return set of kubernetes ingresses
+ */
+ Set<Ingress> ingresses();
+
+ /**
+ * Removes all kubernetes ingresses.
+ */
+ void clear();
+}
diff --git a/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressStoreDelegate.java b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressStoreDelegate.java
new file mode 100644
index 0000000..4bfa16b
--- /dev/null
+++ b/apps/k8s-networking/api/src/main/java/org/onosproject/k8snetworking/api/K8sIngressStoreDelegate.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.api;
+
+import org.onosproject.store.StoreDelegate;
+
+/**
+ * Kubernetes ingress store delegate abstraction.
+ */
+public interface K8sIngressStoreDelegate extends StoreDelegate<K8sIngressEvent> {
+}
diff --git a/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/cli/K8sIngressListCommand.java b/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/cli/K8sIngressListCommand.java
new file mode 100644
index 0000000..8003e28
--- /dev/null
+++ b/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/cli/K8sIngressListCommand.java
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.cli;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Lists;
+import io.fabric8.kubernetes.api.model.extensions.Ingress;
+import io.fabric8.kubernetes.client.utils.Serialization;
+import org.apache.karaf.shell.api.action.Command;
+import org.apache.karaf.shell.api.action.lifecycle.Service;
+import org.onosproject.cli.AbstractShellCommand;
+import org.onosproject.k8snetworking.api.K8sIngressService;
+
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.List;
+
+import static org.onosproject.k8snetworking.util.K8sNetworkingUtil.prettyJson;
+
+/**
+ * Lists kubernetes ingresses.
+ */
+@Service
+@Command(scope = "onos", name = "k8s-ingresses",
+ description = "Lists all kubernetes ingresses")
+public class K8sIngressListCommand extends AbstractShellCommand {
+
+ private static final String FORMAT = "%-50s%-15s%-30s";
+
+ @Override
+ protected void doExecute() {
+ K8sIngressService service = get(K8sIngressService.class);
+ List<Ingress> ingresses = Lists.newArrayList(service.ingresses());
+ ingresses.sort(Comparator.comparing(p -> p.getMetadata().getName()));
+
+ if (outputJson()) {
+ print("%s", json(ingresses));
+ } else {
+ print(FORMAT, "Name", "Namespace", "LB Addresses");
+
+ for (Ingress ingress : ingresses) {
+
+ List<String> lbIps = Lists.newArrayList();
+
+ ingress.getStatus().getLoadBalancer()
+ .getIngress().forEach(i -> lbIps.add(i.getIp()));
+
+ print(FORMAT,
+ ingress.getMetadata().getName(),
+ ingress.getMetadata().getNamespace(),
+ lbIps.isEmpty() ? "" : lbIps);
+ }
+ }
+ }
+
+ private String json(List<Ingress> ingresses) {
+ ObjectMapper mapper = new ObjectMapper();
+ ArrayNode result = mapper.createArrayNode();
+ try {
+ for (Ingress ingress : ingresses) {
+ ObjectNode json = (ObjectNode) new ObjectMapper()
+ .readTree(Serialization.asJson(ingress));
+ result.add(json);
+ }
+ return prettyJson(mapper, result.toString());
+ } catch (IOException e) {
+ log.warn("Failed to parse ingress's JSON string.");
+ return "";
+ }
+ }
+}
diff --git a/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/cli/K8sSyncStateCommand.java b/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/cli/K8sSyncStateCommand.java
index 170425c..9b16795 100644
--- a/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/cli/K8sSyncStateCommand.java
+++ b/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/cli/K8sSyncStateCommand.java
@@ -18,6 +18,7 @@
import com.google.common.collect.Lists;
import io.fabric8.kubernetes.api.model.Endpoints;
import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.extensions.Ingress;
import io.fabric8.kubernetes.client.KubernetesClient;
import org.apache.karaf.shell.api.action.Command;
import org.apache.karaf.shell.api.action.lifecycle.Service;
@@ -26,6 +27,7 @@
import org.onosproject.cli.AbstractShellCommand;
import org.onosproject.k8snetworking.api.DefaultK8sPort;
import org.onosproject.k8snetworking.api.K8sEndpointsAdminService;
+import org.onosproject.k8snetworking.api.K8sIngressAdminService;
import org.onosproject.k8snetworking.api.K8sNetworkAdminService;
import org.onosproject.k8snetworking.api.K8sPodAdminService;
import org.onosproject.k8snetworking.api.K8sPort;
@@ -52,6 +54,7 @@
private static final String POD_FORMAT = "%-50s%-15s%-15s%-30s";
private static final String SERVICE_FORMAT = "%-50s%-30s%-30s";
private static final String ENDPOINTS_FORMAT = "%-50s%-50s%-20s";
+ private static final String INGRESS_FORMAT = "%-50s%-15s%-30s";
private static final String PORT_ID = "portId";
private static final String DEVICE_ID = "deviceId";
@@ -66,6 +69,8 @@
K8sPodAdminService podAdminService = get(K8sPodAdminService.class);
K8sServiceAdminService serviceAdminService =
get(K8sServiceAdminService.class);
+ K8sIngressAdminService ingressAdminService =
+ get(K8sIngressAdminService.class);
K8sEndpointsAdminService endpointsAdminService =
get(K8sEndpointsAdminService.class);
K8sNetworkAdminService networkAdminService =
@@ -120,6 +125,31 @@
printPod(pod);
});
+
+ print("\nSynchronizing kubernetes ingresses");
+ print(INGRESS_FORMAT, "Name", "Namespace", "LB Addresses");
+ client.extensions().ingresses().list().getItems().forEach(ingress -> {
+ if (ingressAdminService.ingress(ingress.getMetadata().getUid()) != null) {
+ ingressAdminService.updateIngress(ingress);
+ } else {
+ ingressAdminService.createIngress(ingress);
+ }
+ printIngresses(ingress);
+ });
+
+ }
+
+ private void printIngresses(Ingress ingress) {
+
+ List<String> lbIps = Lists.newArrayList();
+
+ ingress.getStatus().getLoadBalancer()
+ .getIngress().forEach(i -> lbIps.add(i.getIp()));
+
+ print(INGRESS_FORMAT,
+ ingress.getMetadata().getName(),
+ ingress.getMetadata().getNamespace(),
+ lbIps.isEmpty() ? "" : lbIps);
}
private void printEndpoints(Endpoints endpoints) {
diff --git a/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/impl/DistributedK8sIngressStore.java b/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/impl/DistributedK8sIngressStore.java
new file mode 100644
index 0000000..c6a0f49
--- /dev/null
+++ b/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/impl/DistributedK8sIngressStore.java
@@ -0,0 +1,193 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.impl;
+
+import com.google.common.collect.ImmutableSet;
+import io.fabric8.kubernetes.api.model.LoadBalancerIngress;
+import io.fabric8.kubernetes.api.model.LoadBalancerStatus;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.extensions.Ingress;
+import io.fabric8.kubernetes.api.model.extensions.IngressBackend;
+import io.fabric8.kubernetes.api.model.extensions.IngressRule;
+import io.fabric8.kubernetes.api.model.extensions.IngressSpec;
+import io.fabric8.kubernetes.api.model.extensions.IngressStatus;
+import io.fabric8.kubernetes.api.model.extensions.IngressTLS;
+import org.onlab.util.KryoNamespace;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.k8snetworking.api.K8sIngressEvent;
+import org.onosproject.k8snetworking.api.K8sIngressStore;
+import org.onosproject.k8snetworking.api.K8sIngressStoreDelegate;
+import org.onosproject.store.AbstractStore;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Versioned;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.slf4j.Logger;
+
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static java.util.concurrent.Executors.newSingleThreadExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.onosproject.k8snetworking.api.K8sIngressEvent.Type.K8S_INGRESS_CREATED;
+import static org.onosproject.k8snetworking.api.K8sIngressEvent.Type.K8S_INGRESS_REMOVED;
+import static org.onosproject.k8snetworking.api.K8sIngressEvent.Type.K8S_INGRESS_UPDATED;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Implementation of kubernetes ingress store using consistent map.
+ */
+@Component(immediate = true, service = K8sIngressStore.class)
+public class DistributedK8sIngressStore
+ extends AbstractStore<K8sIngressEvent, K8sIngressStoreDelegate>
+ implements K8sIngressStore {
+
+ private final Logger log = getLogger(getClass());
+
+ private static final String ERR_NOT_FOUND = " does not exist";
+ private static final String ERR_DUPLICATE = " already exists";
+ private static final String APP_ID = "org.onosproject.k8snetwork";
+
+ private static final KryoNamespace
+ SERIALIZER_K8S_INGRESS = KryoNamespace.newBuilder()
+ .register(KryoNamespaces.API)
+ .register(Ingress.class)
+ .register(ObjectMeta.class)
+ .register(IngressSpec.class)
+ .register(IngressStatus.class)
+ .register(IngressBackend.class)
+ .register(IngressRule.class)
+ .register(IngressTLS.class)
+ .register(LoadBalancerStatus.class)
+ .register(LoadBalancerIngress.class)
+ .build();
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected CoreService coreService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected StorageService storageService;
+
+ private final ExecutorService eventExecutor = newSingleThreadExecutor(
+ groupedThreads(this.getClass().getSimpleName(), "event-handler", log));
+
+ private final MapEventListener<String, Ingress> ingressMapListener = new K8sIngressMapListener();
+ private ConsistentMap<String, Ingress> ingressStore;
+
+ @Activate
+ protected void activate() {
+ ApplicationId appId = coreService.registerApplication(APP_ID);
+ ingressStore = storageService.<String, Ingress>consistentMapBuilder()
+ .withSerializer(Serializer.using(SERIALIZER_K8S_INGRESS))
+ .withName("k8s-ingress-store")
+ .withApplicationId(appId)
+ .build();
+
+ ingressStore.addListener(ingressMapListener);
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ ingressStore.removeListener(ingressMapListener);
+ eventExecutor.shutdown();
+ log.info("Stopped");
+ }
+
+ @Override
+ public void createIngress(Ingress ingress) {
+ ingressStore.compute(ingress.getMetadata().getUid(), (uid, existing) -> {
+ final String error = ingress.getMetadata().getUid() + ERR_DUPLICATE;
+ checkArgument(existing == null, error);
+ return ingress;
+ });
+ }
+
+ @Override
+ public void updateIngress(Ingress ingress) {
+ ingressStore.compute(ingress.getMetadata().getUid(), (uid, existing) -> {
+ final String error = ingress.getMetadata().getUid() + ERR_NOT_FOUND;
+ checkArgument(existing != null, error);
+ return ingress;
+ });
+ }
+
+ @Override
+ public Ingress removeIngress(String uid) {
+ Versioned<Ingress> ingress = ingressStore.remove(uid);
+ if (ingress == null) {
+ final String error = uid + ERR_NOT_FOUND;
+ throw new IllegalArgumentException(error);
+ }
+ return ingress.value();
+ }
+
+ @Override
+ public Ingress ingress(String uid) {
+ return ingressStore.asJavaMap().get(uid);
+ }
+
+ @Override
+ public Set<Ingress> ingresses() {
+ return ImmutableSet.copyOf(ingressStore.asJavaMap().values());
+ }
+
+ @Override
+ public void clear() {
+ ingressStore.clear();
+ }
+
+ private class K8sIngressMapListener implements MapEventListener<String, Ingress> {
+
+ @Override
+ public void event(MapEvent<String, Ingress> event) {
+
+ switch (event.type()) {
+ case INSERT:
+ log.debug("Kubernetes ingress created {}", event.newValue());
+ eventExecutor.execute(() ->
+ notifyDelegate(new K8sIngressEvent(
+ K8S_INGRESS_CREATED, event.newValue().value())));
+ break;
+ case UPDATE:
+ log.debug("Kubernetes ingress updated {}", event.newValue());
+ eventExecutor.execute(() ->
+ notifyDelegate(new K8sIngressEvent(
+ K8S_INGRESS_UPDATED, event.newValue().value())));
+ break;
+ case REMOVE:
+ log.debug("Kubernetes ingress removed {}", event.oldValue());
+ eventExecutor.execute(() ->
+ notifyDelegate(new K8sIngressEvent(
+ K8S_INGRESS_REMOVED, event.oldValue().value())));
+ break;
+ default:
+ // do nothing
+ break;
+ }
+ }
+ }
+}
diff --git a/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/impl/K8sIngressManager.java b/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/impl/K8sIngressManager.java
new file mode 100644
index 0000000..037ff0c
--- /dev/null
+++ b/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/impl/K8sIngressManager.java
@@ -0,0 +1,162 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.impl;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableSet;
+import io.fabric8.kubernetes.api.model.extensions.Ingress;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.event.ListenerRegistry;
+import org.onosproject.k8snetworking.api.K8sIngressAdminService;
+import org.onosproject.k8snetworking.api.K8sIngressEvent;
+import org.onosproject.k8snetworking.api.K8sIngressListener;
+import org.onosproject.k8snetworking.api.K8sIngressService;
+import org.onosproject.k8snetworking.api.K8sIngressStore;
+import org.onosproject.k8snetworking.api.K8sIngressStoreDelegate;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.slf4j.Logger;
+
+import java.util.Set;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onosproject.k8snetworking.api.Constants.K8S_NETWORKING_APP_ID;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Provides implementation of administering and interfacing kubernetes ingress.
+ */
+@Component(
+ immediate = true,
+ service = {K8sIngressAdminService.class, K8sIngressService.class}
+)
+public class K8sIngressManager
+ extends ListenerRegistry<K8sIngressEvent, K8sIngressListener>
+ implements K8sIngressAdminService, K8sIngressService {
+
+ protected final Logger log = getLogger(getClass());
+
+ private static final String MSG_INGRESS = "Kubernetes ingress %s %s";
+ private static final String MSG_CREATED = "created";
+ private static final String MSG_UPDATED = "updated";
+ private static final String MSG_REMOVED = "removed";
+
+ private static final String ERR_NULL_INGRESS = "Kubernetes ingress cannot be null";
+ private static final String ERR_NULL_INGRESS_UID = "Kubernetes ingress UID cannot be null";
+
+ private static final String ERR_IN_USE = " still in use";
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected CoreService coreService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected K8sIngressStore k8sIngressStore;
+
+ private final K8sIngressStoreDelegate delegate = new InternalIngressStorageDelegate();
+
+ private ApplicationId appId;
+
+ @Activate
+ protected void activate() {
+ appId = coreService.registerApplication(K8S_NETWORKING_APP_ID);
+
+ k8sIngressStore.setDelegate(delegate);
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ k8sIngressStore.unsetDelegate(delegate);
+ log.info("Stopped");
+ }
+
+ @Override
+ public void createIngress(Ingress ingress) {
+ checkNotNull(ingress, ERR_NULL_INGRESS);
+ checkArgument(!Strings.isNullOrEmpty(ingress.getMetadata().getUid()),
+ ERR_NULL_INGRESS_UID);
+
+ k8sIngressStore.createIngress(ingress);
+
+ log.info(String.format(MSG_INGRESS, ingress.getMetadata().getName(), MSG_CREATED));
+ }
+
+ @Override
+ public void updateIngress(Ingress ingress) {
+ checkNotNull(ingress, ERR_NULL_INGRESS);
+ checkArgument(!Strings.isNullOrEmpty(ingress.getMetadata().getUid()),
+ ERR_NULL_INGRESS_UID);
+
+ k8sIngressStore.updateIngress(ingress);
+
+ log.info(String.format(MSG_INGRESS, ingress.getMetadata().getName(), MSG_UPDATED));
+ }
+
+ @Override
+ public void removeIngress(String uid) {
+ checkArgument(!Strings.isNullOrEmpty(uid), ERR_NULL_INGRESS_UID);
+
+ synchronized (this) {
+ if (isIngressInUse(uid)) {
+ final String error = String.format(MSG_INGRESS, uid, ERR_IN_USE);
+ throw new IllegalStateException(error);
+ }
+
+ Ingress ingress = k8sIngressStore.removeIngress(uid);
+
+ if (ingress != null) {
+ log.info(String.format(MSG_INGRESS,
+ ingress.getMetadata().getName(), MSG_REMOVED));
+ }
+ }
+ }
+
+ @Override
+ public void clear() {
+ k8sIngressStore.clear();
+ }
+
+ @Override
+ public Ingress ingress(String uid) {
+ checkArgument(!Strings.isNullOrEmpty(uid), ERR_NULL_INGRESS_UID);
+ return k8sIngressStore.ingress(uid);
+ }
+
+ @Override
+ public Set<Ingress> ingresses() {
+ return ImmutableSet.copyOf(k8sIngressStore.ingresses());
+ }
+
+ private boolean isIngressInUse(String uid) {
+ return false;
+ }
+
+ private class InternalIngressStorageDelegate implements K8sIngressStoreDelegate {
+
+ @Override
+ public void notify(K8sIngressEvent event) {
+ if (event != null) {
+ log.trace("send kubernetes ingress event {}", event);
+ process(event);
+ }
+ }
+ }
+}
diff --git a/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/impl/K8sIngressWatcher.java b/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/impl/K8sIngressWatcher.java
new file mode 100644
index 0000000..e640fcc
--- /dev/null
+++ b/apps/k8s-networking/app/src/main/java/org/onosproject/k8snetworking/impl/K8sIngressWatcher.java
@@ -0,0 +1,203 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.impl;
+
+import io.fabric8.kubernetes.api.model.extensions.Ingress;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.Watcher;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.LeadershipService;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.k8snetworking.api.K8sIngressAdminService;
+import org.onosproject.k8snode.api.K8sApiConfigEvent;
+import org.onosproject.k8snode.api.K8sApiConfigListener;
+import org.onosproject.k8snode.api.K8sApiConfigService;
+import org.onosproject.mastership.MastershipService;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.component.annotations.ReferenceCardinality;
+import org.slf4j.Logger;
+
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+
+import static java.util.concurrent.Executors.newSingleThreadExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.onosproject.k8snetworking.api.Constants.K8S_NETWORKING_APP_ID;
+import static org.onosproject.k8snetworking.util.K8sNetworkingUtil.k8sClient;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Kubernetes ingress watcher used for feeding ingress information.
+ */
+@Component(immediate = true)
+public class K8sIngressWatcher {
+
+ private final Logger log = getLogger(getClass());
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected CoreService coreService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected MastershipService mastershipService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected ClusterService clusterService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected LeadershipService leadershipService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected K8sIngressAdminService k8sIngressAdminService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected K8sApiConfigService k8sApiConfigService;
+
+ private final ExecutorService eventExecutor = newSingleThreadExecutor(
+ groupedThreads(this.getClass().getSimpleName(), "event-handler"));
+
+ private final Watcher<Ingress> internalK8sIngressWatcher =
+ new InternalK8sIngressWatcher();
+ private final InternalK8sApiConfigListener internalK8sApiConfigListener =
+ new InternalK8sApiConfigListener();
+
+ private ApplicationId appId;
+ private NodeId localNodeId;
+
+ @Activate
+ protected void activate() {
+ appId = coreService.registerApplication(K8S_NETWORKING_APP_ID);
+ localNodeId = clusterService.getLocalNode().id();
+ leadershipService.runForLeadership(appId.name());
+ k8sApiConfigService.addListener(internalK8sApiConfigListener);
+
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ k8sApiConfigService.removeListener(internalK8sApiConfigListener);
+ leadershipService.withdraw(appId.name());
+ eventExecutor.shutdown();
+
+ log.info("Stopped");
+ }
+
+ private class InternalK8sApiConfigListener implements K8sApiConfigListener {
+
+ private boolean isRelevantHelper() {
+ return Objects.equals(localNodeId, leadershipService.getLeader(appId.name()));
+ }
+
+ @Override
+ public void event(K8sApiConfigEvent event) {
+
+ switch (event.type()) {
+ case K8S_API_CONFIG_UPDATED:
+ eventExecutor.execute(this::processConfigUpdating);
+ break;
+ case K8S_API_CONFIG_CREATED:
+ case K8S_API_CONFIG_REMOVED:
+ default:
+ // do nothing
+ break;
+ }
+ }
+
+ private void processConfigUpdating() {
+ if (!isRelevantHelper()) {
+ return;
+ }
+
+ KubernetesClient client = k8sClient(k8sApiConfigService);
+
+ if (client != null) {
+ client.extensions().ingresses().watch(internalK8sIngressWatcher);
+ }
+ }
+ }
+
+ private class InternalK8sIngressWatcher implements Watcher<Ingress> {
+
+ @Override
+ public void eventReceived(Action action, Ingress ingress) {
+ switch (action) {
+ case ADDED:
+ eventExecutor.execute(() -> processAddition(ingress));
+ break;
+ case MODIFIED:
+ eventExecutor.execute(() -> processModification(ingress));
+ break;
+ case DELETED:
+ eventExecutor.execute(() -> processDeletion(ingress));
+ break;
+ case ERROR:
+ log.warn("Failures processing ingress manipulation.");
+ default:
+ break;
+ }
+ }
+
+ @Override
+ public void onClose(KubernetesClientException e) {
+ log.info("Ingress watcher OnClose: {}" + e);
+ }
+
+ private void processAddition(Ingress ingress) {
+ if (!isMaster()) {
+ return;
+ }
+
+ log.trace("Process ingress {} creating event from API server.",
+ ingress.getMetadata().getName());
+
+ k8sIngressAdminService.createIngress(ingress);
+ }
+
+ private void processModification(Ingress ingress) {
+ if (!isMaster()) {
+ return;
+ }
+
+ log.trace("Process ingress {} updating event from API server.",
+ ingress.getMetadata().getName());
+
+ if (k8sIngressAdminService.ingress(ingress.getMetadata().getUid()) != null) {
+ k8sIngressAdminService.updateIngress(ingress);
+ }
+ }
+
+ private void processDeletion(Ingress ingress) {
+ if (!isMaster()) {
+ return;
+ }
+
+ log.trace("Process ingress {} removal event from API server.",
+ ingress.getMetadata().getName());
+
+ k8sIngressAdminService.removeIngress(ingress.getMetadata().getUid());
+ }
+
+ private boolean isMaster() {
+ return Objects.equals(localNodeId, leadershipService.getLeader(appId.name()));
+ }
+ }
+}
diff --git a/apps/k8s-networking/app/src/test/java/org/onosproject/k8snetworking/impl/K8sIngressManagerTest.java b/apps/k8s-networking/app/src/test/java/org/onosproject/k8snetworking/impl/K8sIngressManagerTest.java
new file mode 100644
index 0000000..f401207
--- /dev/null
+++ b/apps/k8s-networking/app/src/test/java/org/onosproject/k8snetworking/impl/K8sIngressManagerTest.java
@@ -0,0 +1,218 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.k8snetworking.impl;
+
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.MoreExecutors;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.extensions.Ingress;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.onlab.junit.TestUtils;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreServiceAdapter;
+import org.onosproject.core.DefaultApplicationId;
+import org.onosproject.event.Event;
+import org.onosproject.k8snetworking.api.K8sIngressEvent;
+import org.onosproject.k8snetworking.api.K8sIngressListener;
+import org.onosproject.store.service.TestStorageService;
+
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.onosproject.k8snetworking.api.K8sIngressEvent.Type.K8S_INGRESS_CREATED;
+import static org.onosproject.k8snetworking.api.K8sIngressEvent.Type.K8S_INGRESS_REMOVED;
+import static org.onosproject.k8snetworking.api.K8sIngressEvent.Type.K8S_INGRESS_UPDATED;
+
+/**
+ * Unit tests for kubernetes ingress manager.
+ */
+public class K8sIngressManagerTest {
+
+ private static final ApplicationId TEST_APP_ID = new DefaultApplicationId(1, "test");
+
+ private static final String UNKNOWN_UID = "unknown_uid";
+ private static final String UPDATED_UID = "updated_uid";
+ private static final String UPDATED_NAME = "updated_name";
+
+ private static final String INGRESS_UID = "ingress_uid";
+ private static final String INGRESS_NAME = "ingress_name";
+
+ private static final Ingress INGRESS = createK8sIngress(INGRESS_UID, INGRESS_NAME);
+ private static final Ingress INGRESS_UPDATED = createK8sIngress(INGRESS_UID, UPDATED_NAME);
+
+ private final TestK8sIngressListener testListener = new TestK8sIngressListener();
+
+ private K8sIngressManager target;
+ private DistributedK8sIngressStore k8sIngressStore;
+
+ @Before
+ public void setUp() throws Exception {
+ k8sIngressStore = new DistributedK8sIngressStore();
+ TestUtils.setField(k8sIngressStore, "coreService", new TestCoreService());
+ TestUtils.setField(k8sIngressStore, "storageService", new TestStorageService());
+ TestUtils.setField(k8sIngressStore, "eventExecutor", MoreExecutors.newDirectExecutorService());
+ k8sIngressStore.activate();
+
+ target = new K8sIngressManager();
+ TestUtils.setField(target, "coreService", new TestCoreService());
+ target.k8sIngressStore = k8sIngressStore;
+ target.addListener(testListener);
+ target.activate();
+ }
+
+ @After
+ public void tearDown() {
+ target.removeListener(testListener);
+ k8sIngressStore.deactivate();
+ target.deactivate();
+ k8sIngressStore = null;
+ target = null;
+ }
+
+ /**
+ * Tests if getting all ingresses return correct set of ingresses.
+ */
+ @Test
+ public void testGetIngresses() {
+ createBasicIngresses();
+ assertEquals("Number of ingresses did not match", 1, target.ingresses().size());
+ }
+
+ /**
+ * Tests if getting an ingress with UID returns the correct ingress.
+ */
+ @Test
+ public void testGetIngressByUid() {
+ createBasicIngresses();
+ assertNotNull("Ingress did not match", target.ingress(INGRESS_UID));
+ assertNull("Ingress did not match", target.ingress(UNKNOWN_UID));
+ }
+
+ /**
+ * Tests creating and removing a ingress, and checks if it triggers proper events.
+ */
+ @Test
+ public void testCreateAndRemoveIngress() {
+ target.createIngress(INGRESS);
+ assertEquals("Number of ingresses did not match", 1, target.ingresses().size());
+ assertNotNull("Ingress was not created", target.ingress(INGRESS_UID));
+
+ target.removeIngress(INGRESS_UID);
+ assertEquals("Number of ingresses did not match", 0, target.ingresses().size());
+ assertNull("Ingress was not removed", target.ingress(INGRESS_UID));
+
+ validateEvents(K8S_INGRESS_CREATED, K8S_INGRESS_REMOVED);
+ }
+
+ /**
+ * Tests updating a ingress, and checks if it triggers proper events.
+ */
+ @Test
+ public void testCreateAndUpdateIngress() {
+ target.createIngress(INGRESS);
+ assertEquals("Number of ingresses did not match", 1, target.ingresses().size());
+ assertEquals("Ingress did not match", INGRESS_NAME,
+ target.ingress(INGRESS_UID).getMetadata().getName());
+
+ target.updateIngress(INGRESS_UPDATED);
+
+ assertEquals("Number of ingresses did not match", 1, target.ingresses().size());
+ assertEquals("Ingress did not match", UPDATED_NAME,
+ target.ingress(INGRESS_UID).getMetadata().getName());
+ validateEvents(K8S_INGRESS_CREATED, K8S_INGRESS_UPDATED);
+ }
+
+ /**
+ * Tests if creating a null ingress fails with an exception.
+ */
+ @Test(expected = NullPointerException.class)
+ public void testCreateNullIngress() {
+ target.createIngress(null);
+ }
+
+ /**
+ * Tests if creating a duplicate ingress fails with an exception.
+ */
+ @Test(expected = IllegalArgumentException.class)
+ public void testCreateDuplicateIngress() {
+ target.createIngress(INGRESS);
+ target.createIngress(INGRESS);
+ }
+
+ /**
+ * Tests if removing ingress with null ID fails with an exception.
+ */
+ @Test(expected = IllegalArgumentException.class)
+ public void testRemoveIngressWithNull() {
+ target.removeIngress(null);
+ }
+
+ /**
+ * Tests if updating an unregistered ingress fails with an exception.
+ */
+ @Test(expected = IllegalArgumentException.class)
+ public void testUpdateUnregisteredIngress() {
+ target.updateIngress(INGRESS);
+ }
+
+ private void createBasicIngresses() {
+ target.createIngress(INGRESS);
+ }
+
+ private static Ingress createK8sIngress(String uid, String name) {
+ ObjectMeta meta = new ObjectMeta();
+ meta.setUid(uid);
+ meta.setName(name);
+
+ Ingress ingress = new Ingress();
+ ingress.setApiVersion("v1");
+ ingress.setKind("Ingress");
+ ingress.setMetadata(meta);
+
+ return ingress;
+ }
+
+ private static class TestCoreService extends CoreServiceAdapter {
+
+ @Override
+ public ApplicationId registerApplication(String name) {
+ return TEST_APP_ID;
+ }
+ }
+
+ private static class TestK8sIngressListener implements K8sIngressListener {
+ private List<K8sIngressEvent> events = Lists.newArrayList();
+
+ @Override
+ public void event(K8sIngressEvent event) {
+ events.add(event);
+ }
+ }
+
+ private void validateEvents(Enum... types) {
+ int i = 0;
+ assertEquals("Number of events did not match", types.length, testListener.events.size());
+ for (Event event : testListener.events) {
+ assertEquals("Incorrect event received", types[i], event.type());
+ i++;
+ }
+ testListener.events.clear();
+ }
+}