Sync kubevirt port LCM with VirtualMachine resource LCM
Change-Id: I0f19817dad60e3f836cb43847f567fece4a4e6d6
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/DistributedKubevirtPodStore.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/DistributedKubevirtPodStore.java
index 1b3fd0b..1ef99f6 100644
--- a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/DistributedKubevirtPodStore.java
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/DistributedKubevirtPodStore.java
@@ -28,6 +28,8 @@
import io.fabric8.kubernetes.api.model.ContainerStateTerminated;
import io.fabric8.kubernetes.api.model.ContainerStateWaiting;
import io.fabric8.kubernetes.api.model.ContainerStatus;
+import io.fabric8.kubernetes.api.model.DownwardAPIVolumeFile;
+import io.fabric8.kubernetes.api.model.DownwardAPIVolumeSource;
import io.fabric8.kubernetes.api.model.EmptyDirVolumeSource;
import io.fabric8.kubernetes.api.model.EnvFromSource;
import io.fabric8.kubernetes.api.model.EnvVar;
@@ -212,6 +214,9 @@
.register(LocalObjectReference.class)
.register(PodAntiAffinity.class)
.register(ManagedFieldsEntry.class)
+ .register(DownwardAPIVolumeSource.class)
+ .register(DownwardAPIVolumeFile.class)
+ .register(ObjectFieldSelector.class)
.register(FieldsV1.class)
.register(LinkedHashMap.class)
.register(Collection.class)
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtPodManager.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtPodManager.java
index 3bb2241..b8d9c42 100644
--- a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtPodManager.java
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtPodManager.java
@@ -96,7 +96,7 @@
kubevirtPodStore.createPod(pod);
- log.info(String.format(MSG_POD, pod.getMetadata().getName(), MSG_CREATED));
+ log.debug(String.format(MSG_POD, pod.getMetadata().getName(), MSG_CREATED));
}
@Override
@@ -107,7 +107,7 @@
kubevirtPodStore.updatePod(pod);
- log.info(String.format(MSG_POD, pod.getMetadata().getName(), MSG_UPDATED));
+ log.debug(String.format(MSG_POD, pod.getMetadata().getName(), MSG_UPDATED));
}
@Override
@@ -123,7 +123,7 @@
Pod pod = kubevirtPodStore.removePod(uid);
if (pod != null) {
- log.info(String.format(MSG_POD,
+ log.debug(String.format(MSG_POD,
pod.getMetadata().getName(), MSG_REMOVED));
}
}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtPodPortMapper.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtPodPortMapper.java
index 8529b5c..7b73f69 100644
--- a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtPodPortMapper.java
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtPodPortMapper.java
@@ -16,17 +16,11 @@
package org.onosproject.kubevirtnetworking.impl;
import io.fabric8.kubernetes.api.model.Pod;
-import io.fabric8.kubernetes.api.model.PodBuilder;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import org.json.JSONArray;
-import org.json.JSONObject;
-import org.onlab.packet.IpAddress;
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.kubevirtnetworking.api.KubevirtNetwork;
import org.onosproject.kubevirtnetworking.api.KubevirtNetworkAdminService;
import org.onosproject.kubevirtnetworking.api.KubevirtPodAdminService;
import org.onosproject.kubevirtnetworking.api.KubevirtPodEvent;
@@ -34,6 +28,7 @@
import org.onosproject.kubevirtnetworking.api.KubevirtPort;
import org.onosproject.kubevirtnetworking.api.KubevirtPortAdminService;
import org.onosproject.kubevirtnode.api.KubevirtApiConfigService;
+import org.onosproject.kubevirtnode.api.KubevirtNode;
import org.onosproject.kubevirtnode.api.KubevirtNodeService;
import org.onosproject.mastership.MastershipService;
import org.onosproject.net.device.DeviceService;
@@ -50,11 +45,11 @@
import java.util.Set;
import java.util.concurrent.ExecutorService;
+import static java.lang.Thread.sleep;
import static java.util.concurrent.Executors.newSingleThreadExecutor;
import static org.onlab.util.Tools.groupedThreads;
import static org.onosproject.kubevirtnetworking.api.Constants.KUBEVIRT_NETWORKING_APP_ID;
import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.getPorts;
-import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.k8sClient;
import static org.slf4j.LoggerFactory.getLogger;
/**
@@ -69,6 +64,7 @@
private static final String NAME = "name";
private static final String IPS = "ips";
private static final String NETWORK_PREFIX = "default/";
+ private static final long SLEEP_MS = 2000; // we wait 2s
@Reference(cardinality = ReferenceCardinality.MANDATORY)
protected CoreService coreService;
@@ -143,19 +139,15 @@
case KUBEVIRT_POD_UPDATED:
eventExecutor.execute(() -> processPodUpdate(event.subject()));
break;
- case KUBEVIRT_POD_REMOVED:
- eventExecutor.execute(() -> processPodDeletion(event.subject()));
- break;
case KUBEVIRT_POD_CREATED:
- eventExecutor.execute(() -> processPodCreation(event.subject()));
- break;
+ case KUBEVIRT_POD_REMOVED:
default:
// do nothing
break;
}
}
- private void processPodCreation(Pod pod) {
+ private void processPodUpdate(Pod pod) {
if (!isRelevantHelper()) {
return;
}
@@ -169,113 +161,33 @@
return;
}
- try {
- String networkStatusStr = pod.getMetadata().getAnnotations().get(NETWORK_STATUS_KEY);
- JSONArray networkStatus = new JSONArray(networkStatusStr);
- for (int i = 0; i < networkStatus.length(); i++) {
- JSONObject object = networkStatus.getJSONObject(i);
- String name = object.getString(NAME);
- KubevirtNetwork jsonNetwork = kubevirtNetworkAdminService.networks().stream()
- .filter(n -> (NETWORK_PREFIX + n.name()).equals(name))
- .findAny().orElse(null);
- if (jsonNetwork != null) {
- JSONArray ips = object.getJSONArray(IPS);
- if (ips != null && ips.length() > 0) {
- IpAddress ip = IpAddress.valueOf(ips.getString(0));
- kubevirtNetworkAdminService.reserveIp(jsonNetwork.networkId(), ip);
- }
- }
+ KubevirtNode node = kubevirtNodeService.node(pod.getSpec().getNodeName());
+
+ if (node == null) {
+ try {
+ // we wait until all k8s nodes are available
+ sleep(SLEEP_MS);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
}
- } catch (Exception e) {
- log.error("Failed to reserve IP address", e);
}
- Set<KubevirtPort> ports = getPorts(kubevirtNodeService, kubevirtNetworkAdminService.networks(), pod);
+ Set<KubevirtPort> ports =
+ getPorts(kubevirtNodeService, kubevirtNetworkAdminService.networks(), pod);
if (ports.size() == 0) {
return;
}
ports.forEach(port -> {
- if (kubevirtPortAdminService.port(port.macAddress()) == null) {
- kubevirtPortAdminService.createPort(port);
- }
- });
- }
+ KubevirtPort existing = kubevirtPortAdminService.port(port.macAddress());
- private void processPodUpdate(Pod pod) {
- if (!isRelevantHelper()) {
- return;
- }
-
- Set<KubevirtPort> ports = getPorts(kubevirtNodeService, kubevirtNetworkAdminService.networks(), pod);
- if (ports.size() == 0) {
- return;
- }
-
- for (KubevirtPort port : ports) {
- if (kubevirtPortAdminService.port(port.macAddress()) != null) {
- continue;
- }
-
- if (port.ipAddress() == null) {
- try {
- IpAddress ip = kubevirtNetworkAdminService.allocateIp(port.networkId());
- log.info("IP address {} is allocated from network {}", ip, port.networkId());
- port = port.updateIpAddress(ip);
-
- // update the POD annotation to inject the allocated IP address
- String networkStatusStr = pod.getMetadata().getAnnotations().get(NETWORK_STATUS_KEY);
- JSONArray networkStatus = new JSONArray(networkStatusStr);
- for (int i = 0; i < networkStatus.length(); i++) {
- JSONObject object = networkStatus.getJSONObject(i);
- String name = object.getString(NAME);
-
- if (name.equals(NETWORK_PREFIX + port.networkId())) {
- JSONArray ipsJson = new JSONArray();
- ipsJson.put(ip.toString());
- object.put(IPS, ipsJson);
- }
- }
- Map<String, String> annots = pod.getMetadata().getAnnotations();
- annots.put(NETWORK_STATUS_KEY, networkStatus.toString(4));
-
- KubernetesClient client = k8sClient(kubevirtApiConfigService);
-
- if (client == null) {
- return;
- }
-
- client.pods().inNamespace(pod.getMetadata().getNamespace())
- .withName(pod.getMetadata().getName())
- .edit(r -> new PodBuilder(r)
- .editMetadata()
- .addToAnnotations(annots)
- .endMetadata().build()
- );
- } catch (Exception e) {
- log.error("Failed to allocate IP address", e);
+ if (existing != null) {
+ if (port.deviceId() != null && existing.deviceId() == null) {
+ KubevirtPort updated = existing.updateDeviceId(port.deviceId());
+ // internal we update device ID of kubevirt port
+ kubevirtPortAdminService.updatePort(updated);
}
}
- kubevirtPortAdminService.createPort(port);
- }
- }
-
- private void processPodDeletion(Pod pod) {
- if (!isRelevantHelper()) {
- return;
- }
-
- Set<KubevirtPort> ports = getPorts(kubevirtNodeService, kubevirtNetworkAdminService.networks(), pod);
- if (ports.size() == 0) {
- return;
- }
-
- ports.forEach(port -> {
- if (port.ipAddress() != null) {
- kubevirtNetworkAdminService.releaseIp(port.networkId(), port.ipAddress());
- }
-
- kubevirtPortAdminService.removePort(port.macAddress());
});
}
}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSwitchingTenantHandler.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSwitchingTenantHandler.java
index 5eb7732..066ce49 100644
--- a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSwitchingTenantHandler.java
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSwitchingTenantHandler.java
@@ -15,7 +15,6 @@
*/
package org.onosproject.kubevirtnetworking.impl;
-import io.fabric8.kubernetes.api.model.Pod;
import org.onlab.packet.Ethernet;
import org.onlab.packet.Ip4Address;
import org.onlab.packet.IpPrefix;
@@ -27,10 +26,10 @@
import org.onosproject.kubevirtnetworking.api.KubevirtFlowRuleService;
import org.onosproject.kubevirtnetworking.api.KubevirtNetwork;
import org.onosproject.kubevirtnetworking.api.KubevirtNetworkService;
-import org.onosproject.kubevirtnetworking.api.KubevirtPodEvent;
-import org.onosproject.kubevirtnetworking.api.KubevirtPodListener;
import org.onosproject.kubevirtnetworking.api.KubevirtPodService;
import org.onosproject.kubevirtnetworking.api.KubevirtPort;
+import org.onosproject.kubevirtnetworking.api.KubevirtPortEvent;
+import org.onosproject.kubevirtnetworking.api.KubevirtPortListener;
import org.onosproject.kubevirtnetworking.api.KubevirtPortService;
import org.onosproject.kubevirtnode.api.KubevirtNode;
import org.onosproject.kubevirtnode.api.KubevirtNodeEvent;
@@ -52,7 +51,6 @@
import org.slf4j.Logger;
import java.util.Objects;
-import java.util.Set;
import java.util.concurrent.ExecutorService;
import static java.util.concurrent.Executors.newSingleThreadExecutor;
@@ -62,7 +60,6 @@
import static org.onosproject.kubevirtnetworking.api.Constants.TUNNEL_DEFAULT_TABLE;
import static org.onosproject.kubevirtnetworking.api.KubevirtNetwork.Type.FLAT;
import static org.onosproject.kubevirtnetworking.api.KubevirtNetwork.Type.VLAN;
-import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.getPorts;
import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.tunnelPort;
import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.tunnelToTenantPort;
import static org.onosproject.kubevirtnetworking.util.RulePopulatorUtil.buildExtension;
@@ -103,8 +100,8 @@
private final ExecutorService eventExecutor = newSingleThreadExecutor(
groupedThreads(this.getClass().getSimpleName(), "event-handler"));
- private final InternalKubevirtPodListener kubevirtPodListener =
- new InternalKubevirtPodListener();
+ private final InternalKubevirtPortListener kubevirtPortListener =
+ new InternalKubevirtPortListener();
private final InternalKubevirtNodeListener kubevirtNodeListener =
new InternalKubevirtNodeListener();
@@ -116,7 +113,7 @@
appId = coreService.registerApplication(KUBEVIRT_NETWORKING_APP_ID);
localNodeId = clusterService.getLocalNode().id();
leadershipService.runForLeadership(appId.name());
- kubevirtPodService.addListener(kubevirtPodListener);
+ kubevirtPortService.addListener(kubevirtPortListener);
kubevirtNodeService.addListener(kubevirtNodeListener);
log.info("Started");
@@ -124,7 +121,7 @@
@Deactivate
protected void deactivate() {
- kubevirtPodService.removeListener(kubevirtPodListener);
+ kubevirtPortService.removeListener(kubevirtPortListener);
kubevirtNodeService.removeListener(kubevirtNodeListener);
leadershipService.withdraw(appId.name());
eventExecutor.shutdown();
@@ -132,158 +129,134 @@
log.info("Stopped");
}
- private Set<KubevirtPort> getPortByPod(Pod pod) {
- return getPorts(kubevirtNodeService, kubevirtNetworkService.networks(), pod);
- }
-
- private void setIngressRules(Pod pod, boolean install) {
- Set<KubevirtPort> ports = getPortByPod(pod);
-
- if (ports.size() == 0) {
+ private void setIngressRules(KubevirtPort port, boolean install) {
+ if (port.ipAddress() == null) {
return;
}
- for (KubevirtPort port : ports) {
- if (port.ipAddress() == null) {
- return;
+ KubevirtNetwork network = kubevirtNetworkService.network(port.networkId());
+
+ if (network == null) {
+ return;
+ }
+
+ if (network.type() == FLAT || network.type() == VLAN) {
+ return;
+ }
+
+ if (network.segmentId() == null) {
+ return;
+ }
+
+ KubevirtNode localNode = kubevirtNodeService.node(port.deviceId());
+ if (localNode == null || localNode.type() == MASTER) {
+ return;
+ }
+
+ PortNumber patchPortNumber = tunnelToTenantPort(localNode, network);
+ if (patchPortNumber == null) {
+ return;
+ }
+
+ TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder()
+ .matchTunnelId(Long.parseLong(network.segmentId()));
+
+ TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder()
+ .setOutput(patchPortNumber);
+
+ flowRuleService.setRule(
+ appId,
+ localNode.tunBridge(),
+ sBuilder.build(),
+ tBuilder.build(),
+ PRIORITY_TUNNEL_RULE,
+ TUNNEL_DEFAULT_TABLE,
+ install);
+
+ log.debug("Install ingress rules for instance {}, segment ID {}",
+ port.ipAddress(), network.segmentId());
+ }
+
+ private void setEgressRules(KubevirtPort port, boolean install) {
+ if (port.ipAddress() == null) {
+ return;
+ }
+
+ KubevirtNetwork network = kubevirtNetworkService.network(port.networkId());
+
+ if (network == null) {
+ return;
+ }
+
+ if (network.type() == FLAT || network.type() == VLAN) {
+ return;
+ }
+
+ if (network.segmentId() == null) {
+ return;
+ }
+
+ KubevirtNode localNode = kubevirtNodeService.node(port.deviceId());
+
+ if (localNode == null || localNode.type() == MASTER) {
+ return;
+ }
+
+ for (KubevirtNode remoteNode : kubevirtNodeService.completeNodes(WORKER)) {
+ if (remoteNode.hostname().equals(localNode.hostname())) {
+ continue;
}
- KubevirtNetwork network = kubevirtNetworkService.network(port.networkId());
-
- if (network == null) {
- return;
- }
-
- if (network.type() == FLAT || network.type() == VLAN) {
- return;
- }
-
- if (network.segmentId() == null) {
- return;
- }
-
- KubevirtNode localNode = kubevirtNodeService.node(pod.getSpec().getNodeName());
- if (localNode == null || localNode.type() == MASTER) {
- return;
- }
-
- PortNumber patchPortNumber = tunnelToTenantPort(localNode, network);
+ PortNumber patchPortNumber = tunnelToTenantPort(remoteNode, network);
if (patchPortNumber == null) {
return;
}
- TrafficSelector.Builder sBuilder = DefaultTrafficSelector.builder()
- .matchTunnelId(Long.parseLong(network.segmentId()));
+ PortNumber tunnelPortNumber = tunnelPort(remoteNode, network);
+ if (tunnelPortNumber == null) {
+ return;
+ }
+
+ TrafficSelector.Builder sIpBuilder = DefaultTrafficSelector.builder()
+ .matchInPort(patchPortNumber)
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPDst(IpPrefix.valueOf(port.ipAddress(), 32));
+
+ TrafficSelector.Builder sArpBuilder = DefaultTrafficSelector.builder()
+ .matchInPort(patchPortNumber)
+ .matchEthType(Ethernet.TYPE_ARP)
+ .matchArpTpa(Ip4Address.valueOf(port.ipAddress().toString()));
TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder()
- .setOutput(patchPortNumber);
+ .setTunnelId(Long.parseLong(network.segmentId()))
+ .extension(buildExtension(
+ deviceService,
+ remoteNode.tunBridge(),
+ localNode.dataIp().getIp4Address()),
+ remoteNode.tunBridge())
+ .setOutput(tunnelPortNumber);
flowRuleService.setRule(
appId,
- localNode.tunBridge(),
- sBuilder.build(),
+ remoteNode.tunBridge(),
+ sIpBuilder.build(),
tBuilder.build(),
PRIORITY_TUNNEL_RULE,
TUNNEL_DEFAULT_TABLE,
install);
- log.debug("Install ingress rules for instance {}, segment ID {}",
- port.ipAddress(), network.segmentId());
- }
- }
-
- private void setEgressRules(Pod pod, boolean install) {
- KubevirtNode localNode = kubevirtNodeService.node(pod.getSpec().getNodeName());
-
- if (localNode == null) {
- return;
+ flowRuleService.setRule(
+ appId,
+ remoteNode.tunBridge(),
+ sArpBuilder.build(),
+ tBuilder.build(),
+ PRIORITY_TUNNEL_RULE,
+ TUNNEL_DEFAULT_TABLE,
+ install);
}
- if (localNode.type() == MASTER) {
- return;
- }
-
- Set<KubevirtPort> ports = getPortByPod(pod);
-
- if (ports.size() == 0) {
- return;
- }
-
- for (KubevirtPort port : ports) {
- if (port.ipAddress() == null) {
- return;
- }
-
- KubevirtNetwork network = kubevirtNetworkService.network(port.networkId());
-
- if (network == null) {
- return;
- }
-
- if (network.type() == FLAT || network.type() == VLAN) {
- return;
- }
-
- if (network.segmentId() == null) {
- return;
- }
-
- for (KubevirtNode remoteNode : kubevirtNodeService.completeNodes(WORKER)) {
- if (remoteNode.hostname().equals(localNode.hostname())) {
- continue;
- }
-
- PortNumber patchPortNumber = tunnelToTenantPort(remoteNode, network);
- if (patchPortNumber == null) {
- return;
- }
-
- PortNumber tunnelPortNumber = tunnelPort(remoteNode, network);
- if (tunnelPortNumber == null) {
- return;
- }
-
- TrafficSelector.Builder sIpBuilder = DefaultTrafficSelector.builder()
- .matchInPort(patchPortNumber)
- .matchEthType(Ethernet.TYPE_IPV4)
- .matchIPDst(IpPrefix.valueOf(port.ipAddress(), 32));
-
- TrafficSelector.Builder sArpBuilder = DefaultTrafficSelector.builder()
- .matchInPort(patchPortNumber)
- .matchEthType(Ethernet.TYPE_ARP)
- .matchArpTpa(Ip4Address.valueOf(port.ipAddress().toString()));
-
- TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder()
- .setTunnelId(Long.parseLong(network.segmentId()))
- .extension(buildExtension(
- deviceService,
- remoteNode.tunBridge(),
- localNode.dataIp().getIp4Address()),
- remoteNode.tunBridge())
- .setOutput(tunnelPortNumber);
-
- flowRuleService.setRule(
- appId,
- remoteNode.tunBridge(),
- sIpBuilder.build(),
- tBuilder.build(),
- PRIORITY_TUNNEL_RULE,
- TUNNEL_DEFAULT_TABLE,
- install);
-
- flowRuleService.setRule(
- appId,
- remoteNode.tunBridge(),
- sArpBuilder.build(),
- tBuilder.build(),
- PRIORITY_TUNNEL_RULE,
- TUNNEL_DEFAULT_TABLE,
- install);
- }
-
- log.debug("Install egress rules for instance {}, segment ID {}",
- port.ipAddress(), network.segmentId());
- }
+ log.debug("Install egress rules for instance {}, segment ID {}",
+ port.ipAddress(), network.segmentId());
}
private class InternalKubevirtNodeListener implements KubevirtNodeListener {
@@ -311,30 +284,35 @@
return;
}
- kubevirtPodService.pods().stream()
- .filter(pod -> node.hostname().equals(pod.getSpec().getNodeName()))
- .forEach(pod -> {
- setIngressRules(pod, true);
- setEgressRules(pod, true);
+ kubevirtPortService.ports().stream()
+ .filter(port -> node.equals(kubevirtNodeService.node(port.deviceId())))
+ .forEach(port -> {
+ setIngressRules(port, true);
+ setEgressRules(port, true);
});
}
}
- private class InternalKubevirtPodListener implements KubevirtPodListener {
+ private class InternalKubevirtPortListener implements KubevirtPortListener {
+
+ @Override
+ public boolean isRelevant(KubevirtPortEvent event) {
+ return event.subject().deviceId() != null;
+ }
private boolean isRelevantHelper() {
return Objects.equals(localNodeId, leadershipService.getLeader(appId.name()));
}
@Override
- public void event(KubevirtPodEvent event) {
+ public void event(KubevirtPortEvent event) {
switch (event.type()) {
- case KUBEVIRT_POD_UPDATED:
- eventExecutor.execute(() -> processPodUpdate(event.subject()));
+ case KUBEVIRT_PORT_UPDATED:
+ eventExecutor.execute(() -> processPortUpdate(event.subject()));
break;
- case KUBEVIRT_POD_REMOVED:
- eventExecutor.execute(() -> processPodRemoval(event.subject()));
+ case KUBEVIRT_PORT_REMOVED:
+ eventExecutor.execute(() -> processPortRemoval(event.subject()));
break;
default:
// do nothing
@@ -342,22 +320,22 @@
}
}
- private void processPodUpdate(Pod pod) {
+ private void processPortUpdate(KubevirtPort port) {
if (!isRelevantHelper()) {
return;
}
- setIngressRules(pod, true);
- setEgressRules(pod, true);
+ setIngressRules(port, true);
+ setEgressRules(port, true);
}
- private void processPodRemoval(Pod pod) {
+ private void processPortRemoval(KubevirtPort port) {
if (!isRelevantHelper()) {
return;
}
- setIngressRules(pod, false);
- setEgressRules(pod, false);
+ setIngressRules(port, false);
+ setEgressRules(port, false);
}
}
}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtVmWatcher.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtVmWatcher.java
new file mode 100644
index 0000000..61b9be0
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtVmWatcher.java
@@ -0,0 +1,440 @@
+/*
+ * Copyright 2021-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.kubevirtnetworking.impl;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.Watcher;
+import io.fabric8.kubernetes.client.WatcherException;
+import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext;
+import org.apache.commons.lang3.StringUtils;
+import org.onlab.packet.IpAddress;
+import org.onlab.packet.MacAddress;
+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.kubevirtnetworking.api.DefaultKubevirtPort;
+import org.onosproject.kubevirtnetworking.api.KubevirtNetwork;
+import org.onosproject.kubevirtnetworking.api.KubevirtNetworkAdminService;
+import org.onosproject.kubevirtnetworking.api.KubevirtPodService;
+import org.onosproject.kubevirtnetworking.api.KubevirtPort;
+import org.onosproject.kubevirtnetworking.api.KubevirtPortAdminService;
+import org.onosproject.kubevirtnode.api.KubevirtApiConfigEvent;
+import org.onosproject.kubevirtnode.api.KubevirtApiConfigListener;
+import org.onosproject.kubevirtnode.api.KubevirtApiConfigService;
+import org.onosproject.kubevirtnode.api.KubevirtNodeService;
+import org.onosproject.mastership.MastershipService;
+import org.onosproject.net.DeviceId;
+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.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+
+import static java.lang.Thread.sleep;
+import static java.util.concurrent.Executors.newSingleThreadExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.onosproject.kubevirtnetworking.api.Constants.KUBEVIRT_NETWORKING_APP_ID;
+import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.getPorts;
+import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.k8sClient;
+import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.parseResourceName;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Kubernetes VM watcher used for feeding VM information.
+ */
+@Component(immediate = true)
+public class KubevirtVmWatcher {
+
+ private final Logger log = getLogger(getClass());
+
+ private static final long SLEEP_MS = 3000; // we wait 3s
+
+ private static final String SPEC = "spec";
+ private static final String TEMPLATE = "template";
+ private static final String METADATA = "metadata";
+ private static final String ANNOTATIONS = "annotations";
+ private static final String DOMAIN = "domain";
+ private static final String DEVICES = "devices";
+ private static final String INTERFACES = "interfaces";
+ private static final String NAME = "name";
+ private static final String NETWORK = "network";
+ private static final String MAC = "macAddress";
+ private static final String IP = "ipAddress";
+ private static final String DEFAULT = "default";
+ private static final String NETWORK_SUFFIX = "-net";
+
+ @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 KubevirtNodeService nodeService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected KubevirtNetworkAdminService networkAdminService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected KubevirtPortAdminService portAdminService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected KubevirtPodService podService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected KubevirtApiConfigService configService;
+
+ private final ExecutorService eventExecutor = newSingleThreadExecutor(
+ groupedThreads(this.getClass().getSimpleName(), "event-handler"));
+
+ private final InternalKubevirtVmWatcher watcher = new InternalKubevirtVmWatcher();
+ private final InternalKubevirtApiConfigListener
+ configListener = new InternalKubevirtApiConfigListener();
+
+ CustomResourceDefinitionContext vmCrdCxt = new CustomResourceDefinitionContext
+ .Builder()
+ .withGroup("kubevirt.io")
+ .withScope("Namespaced")
+ .withVersion("v1")
+ .withPlural("virtualmachines")
+ .build();
+
+ private ApplicationId appId;
+ private NodeId localNodeId;
+
+ @Activate
+ protected void activate() {
+ appId = coreService.registerApplication(KUBEVIRT_NETWORKING_APP_ID);
+ localNodeId = clusterService.getLocalNode().id();
+ leadershipService.runForLeadership(appId.name());
+ configService.addListener(configListener);
+
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ configService.removeListener(configListener);
+ leadershipService.withdraw(appId.name());
+ eventExecutor.shutdown();
+
+ log.info("Stopped");
+ }
+
+ private void instantiateWatcher() {
+ KubernetesClient client = k8sClient(configService);
+
+ if (client != null) {
+ try {
+ client.customResource(vmCrdCxt).watch(watcher);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ private class InternalKubevirtApiConfigListener implements KubevirtApiConfigListener {
+
+ private boolean isRelevantHelper() {
+ return Objects.equals(localNodeId, leadershipService.getLeader(appId.name()));
+ }
+
+ @Override
+ public void event(KubevirtApiConfigEvent event) {
+
+ switch (event.type()) {
+ case KUBEVIRT_API_CONFIG_UPDATED:
+ eventExecutor.execute(this::processConfigUpdate);
+ break;
+ case KUBEVIRT_API_CONFIG_CREATED:
+ case KUBEVIRT_API_CONFIG_REMOVED:
+ default:
+ // do nothing
+ break;
+ }
+ }
+
+ private void processConfigUpdate() {
+ if (!isRelevantHelper()) {
+ return;
+ }
+
+ instantiateWatcher();
+ }
+ }
+
+ private class InternalKubevirtVmWatcher implements Watcher<String> {
+
+ @Override
+ public void eventReceived(Action action, String resource) {
+ switch (action) {
+ case ADDED:
+ eventExecutor.execute(() -> processAddition(resource));
+ break;
+ case DELETED:
+ eventExecutor.execute(() -> processDeletion(resource));
+ break;
+ case ERROR:
+ log.warn("Failures processing VM manipulation.");
+ break;
+ default:
+ break;
+ }
+ }
+
+ @Override
+ public void onClose(WatcherException e) {
+ log.warn("VM watcher OnClose, re-instantiate the VM watcher...");
+ instantiateWatcher();
+ }
+
+ private void processAddition(String resource) {
+ if (!isMaster()) {
+ return;
+ }
+
+ parseMacAddresses(resource).forEach((mac, net) -> {
+ KubevirtPort port = DefaultKubevirtPort.builder()
+ .macAddress(mac)
+ .networkId(net)
+ .build();
+
+ String name = parseResourceName(resource);
+
+ Map<String, IpAddress> ips = parseIpAddresses(resource);
+ IpAddress ip;
+ IpAddress existingIp = ips.get(port.networkId());
+
+ KubevirtNetwork network = networkAdminService.network(port.networkId());
+ if (network == null) {
+ try {
+ sleep(SLEEP_MS);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+
+ KubevirtPort existingPort = portAdminService.port(port.macAddress());
+
+ if (existingIp == null) {
+
+ KubernetesClient client = k8sClient(configService);
+
+ if (client == null) {
+ return;
+ }
+
+ ip = networkAdminService.allocateIp(port.networkId());
+ log.info("IP address {} is allocated from network {}", ip, port.networkId());
+
+ try {
+ // we wait a while to avoid potentially referring to old version resource
+ // FIXME: we may need to find a better solution to avoid this
+ sleep(SLEEP_MS);
+ ObjectMapper mapper = new ObjectMapper();
+ Map<String, Object> newResource = client.customResource(vmCrdCxt).get(DEFAULT, name);
+ String newResourceStr = mapper.writeValueAsString(newResource);
+ String updatedResource = updateIpAddress(newResourceStr, port.networkId(), ip);
+ client.customResource(vmCrdCxt).edit(DEFAULT, name, updatedResource);
+ } catch (IOException | InterruptedException e) {
+ log.error("Failed to annotate IP addresses", e);
+ } catch (KubernetesClientException kce) {
+ log.error("Failed to update VM resource", kce);
+ }
+
+ } else {
+ if (existingPort != null) {
+ return;
+ }
+
+ ip = existingIp;
+ networkAdminService.reserveIp(port.networkId(), ip);
+ log.info("IP address {} is reserved from network {}", ip, port.networkId());
+ }
+
+ if (existingPort == null) {
+ KubevirtPort updated = port.updateIpAddress(ip);
+
+ DeviceId deviceId = getDeviceId(podService.pods(), port);
+
+ if (deviceId != null) {
+ updated = updated.updateDeviceId(deviceId);
+ }
+
+ portAdminService.createPort(updated);
+ }
+ });
+ }
+
+ private void processDeletion(String resource) {
+ if (!isMaster()) {
+ return;
+ }
+
+ parseMacAddresses(resource).forEach((mac, net) -> {
+ KubevirtPort port = portAdminService.port(mac);
+ if (port != null) {
+ networkAdminService.releaseIp(port.networkId(), port.ipAddress());
+ log.info("IP address {} is released from network {}",
+ port.ipAddress(), port.networkId());
+
+ portAdminService.removePort(mac);
+ }
+ });
+ }
+
+ private boolean isMaster() {
+ return Objects.equals(localNodeId, leadershipService.getLeader(appId.name()));
+ }
+
+ // FIXME: to obtains the device ID, we have to search through
+ // existing POD inventory, need to find a better wat to obtain device ID
+ private DeviceId getDeviceId(Set<Pod> pods, KubevirtPort port) {
+ Set<Pod> defaultPods = pods.stream()
+ .filter(pod -> pod.getMetadata().getNamespace().equals(DEFAULT))
+ .collect(Collectors.toSet());
+
+ Set<KubevirtPort> allPorts = new HashSet<>();
+ for (Pod pod : defaultPods) {
+ allPorts.addAll(getPorts(nodeService, networkAdminService.networks(), pod));
+ }
+
+ return allPorts.stream().filter(p -> p.macAddress().equals(port.macAddress()))
+ .map(KubevirtPort::deviceId).findFirst().orElse(null);
+ }
+
+ private Map<String, IpAddress> parseIpAddresses(String resource) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode json = mapper.readTree(resource);
+ JsonNode metadata = json.get(SPEC).get(TEMPLATE).get(METADATA);
+
+ JsonNode annots = metadata.get(ANNOTATIONS);
+ if (annots == null) {
+ return new HashMap<>();
+ }
+
+ JsonNode interfacesJson = annots.get(INTERFACES);
+ if (interfacesJson == null) {
+ return new HashMap<>();
+ }
+
+ Map<String, IpAddress> result = new HashMap<>();
+
+ String interfacesString = interfacesJson.asText();
+ ArrayNode interfaces = (ArrayNode) mapper.readTree(interfacesString);
+ for (JsonNode intf : interfaces) {
+ String network = intf.get(NETWORK).asText();
+ String ip = intf.get(IP).asText();
+ result.put(network, IpAddress.valueOf(ip));
+ }
+
+ return result;
+ } catch (IOException e) {
+ log.error("Failed to parse kubevirt VM IP addresses");
+ }
+
+ return new HashMap<>();
+ }
+
+ private String updateIpAddress(String resource, String network, IpAddress ip) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ ObjectNode json = (ObjectNode) mapper.readTree(resource);
+ ObjectNode spec = (ObjectNode) json.get(SPEC);
+ ObjectNode template = (ObjectNode) spec.get(TEMPLATE);
+ ObjectNode metadata = (ObjectNode) template.get(METADATA);
+ ObjectNode annots = (ObjectNode) metadata.get(ANNOTATIONS);
+
+ if (!annots.has(INTERFACES)) {
+ annots.put(INTERFACES, "[]");
+ }
+
+ String intfs = annots.get(INTERFACES).asText();
+ ArrayNode intfsJson = (ArrayNode) mapper.readTree(intfs);
+
+ ObjectNode intf = mapper.createObjectNode();
+ intf.put(NETWORK, network);
+ intf.put(IP, ip.toString());
+
+ intfsJson.add(intf);
+
+ annots.put(INTERFACES, intfsJson.toString());
+ metadata.set(ANNOTATIONS, annots);
+ template.set(METADATA, metadata);
+ spec.set(TEMPLATE, template);
+ json.set(SPEC, spec);
+
+ return json.toString();
+
+ } catch (IOException e) {
+ log.error("Failed to update kubevirt VM IP addresses");
+ }
+ return null;
+ }
+
+ private Map<MacAddress, String> parseMacAddresses(String resource) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode json = mapper.readTree(resource);
+ JsonNode spec = json.get(SPEC).get(TEMPLATE).get(SPEC);
+ ArrayNode interfaces = (ArrayNode) spec.get(DOMAIN).get(DEVICES).get(INTERFACES);
+
+ Map<MacAddress, String> result = new HashMap<>();
+ for (JsonNode intf : interfaces) {
+ String network = intf.get(NAME).asText();
+ JsonNode macJson = intf.get(MAC);
+
+ if (!DEFAULT.equals(network) && macJson != null) {
+ String compact = StringUtils.substringBeforeLast(network, NETWORK_SUFFIX);
+ MacAddress mac = MacAddress.valueOf(macJson.asText());
+ result.put(mac, compact);
+ }
+ }
+
+ return result;
+ } catch (IOException e) {
+ log.error("Failed to parse kubevirt VM MAC addresses");
+ }
+
+ return new HashMap<>();
+ }
+ }
+}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/util/KubevirtNetworkingUtil.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/util/KubevirtNetworkingUtil.java
index 92104a0..4ee8fa2 100644
--- a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/util/KubevirtNetworkingUtil.java
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/util/KubevirtNetworkingUtil.java
@@ -42,6 +42,7 @@
import org.onosproject.kubevirtnode.api.KubevirtApiConfigService;
import org.onosproject.kubevirtnode.api.KubevirtNode;
import org.onosproject.kubevirtnode.api.KubevirtNodeService;
+import org.onosproject.kubevirtnode.api.KubevirtPhyInterface;
import org.onosproject.net.DeviceId;
import org.onosproject.net.Port;
import org.onosproject.net.PortNumber;
@@ -323,7 +324,8 @@
* @param pod kubevirt POD
* @return kubevirt ports attached to the POD
*/
- public static Set<KubevirtPort> getPorts(KubevirtNodeService nodeService, Set<KubevirtNetwork> networks, Pod pod) {
+ public static Set<KubevirtPort> getPorts(KubevirtNodeService nodeService,
+ Set<KubevirtNetwork> networks, Pod pod) {
try {
Map<String, String> annots = pod.getMetadata().getAnnotations();
if (annots == null) {
@@ -339,9 +341,11 @@
if (networkStatusStr == null) {
return ImmutableSet.of();
}
+
KubevirtPort.Builder builder = DefaultKubevirtPort.builder();
KubevirtNode node = nodeService.node(pod.getSpec().getNodeName());
+
if (node != null) {
builder.deviceId(node.intgBridge());
}
@@ -361,12 +365,6 @@
builder.macAddress(MacAddress.valueOf(mac))
.networkId(network.networkId());
- if (object.has(IPS)) {
- JSONArray ips = object.getJSONArray(IPS);
- String ip = (String) ips.get(0);
- builder.ipAddress(IpAddress.valueOf(ip));
- }
-
ports.add(builder.build());
}
}
@@ -549,13 +547,13 @@
* @return external patch port number
*/
public static PortNumber externalPatchPortNum(DeviceService deviceService, KubevirtNode gatewayNode) {
- String gatewayBridgeName = gatewayNode.gatewayBridgeName();
- if (gatewayBridgeName == null) {
+ KubevirtPhyInterface intf = gatewayNode.phyIntfs().stream().findFirst().orElse(null);
+ if (intf == null) {
log.warn("No external interface is attached to gateway {}", gatewayNode.hostname());
return null;
}
- String patchPortName = "int-to-" + gatewayBridgeName;
+ String patchPortName = "int-to-" + intf.network();
Port port = deviceService.getPorts(gatewayNode.intgBridge()).stream()
.filter(p -> p.isEnabled() &&
Objects.equals(p.annotations().value(PORT_NAME), patchPortName))