Implement security group manager, codec and watcher with unit tests
Change-Id: Ib2201d140b9dcb2eff453f13447113bdba66babd
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtListSecurityGroupCommand.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtListSecurityGroupCommand.java
new file mode 100644
index 0000000..3f1b78b
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtListSecurityGroupCommand.java
@@ -0,0 +1,82 @@
+/*
+ * 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.cli;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.StringUtils;
+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.kubevirtnetworking.api.KubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupService;
+
+import java.util.Comparator;
+import java.util.List;
+
+import static org.onosproject.kubevirtnetworking.api.Constants.CLI_ID_LENGTH;
+import static org.onosproject.kubevirtnetworking.api.Constants.CLI_MARGIN_LENGTH;
+import static org.onosproject.kubevirtnetworking.api.Constants.CLI_NAME_LENGTH;
+import static org.onosproject.kubevirtnetworking.api.Constants.CLI_NUMBER_LENGTH;
+import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.genFormatString;
+import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.prettyJson;
+
+/**
+ * Lists kubevirt security groups.
+ */
+@Service
+@Command(scope = "onos", name = "kubevirt-security-groups",
+ description = "Lists all kubevirt security groups")
+public class KubevirtListSecurityGroupCommand extends AbstractShellCommand {
+ @Override
+ protected void doExecute() throws Exception {
+ KubevirtSecurityGroupService service = get(KubevirtSecurityGroupService.class);
+ List<KubevirtSecurityGroup> sgs = Lists.newArrayList(service.securityGroups());
+ sgs.sort(Comparator.comparing(KubevirtSecurityGroup::name));
+
+ String format = genFormatString(ImmutableList.of(CLI_ID_LENGTH,
+ CLI_NAME_LENGTH, CLI_NUMBER_LENGTH));
+
+ if (outputJson()) {
+ print("%s", json(sgs));
+ } else {
+ print(format, "ID", "Name", "# of Rules");
+
+ for (KubevirtSecurityGroup sg : sgs) {
+ print(format, StringUtils.substring(sg.id(), 0,
+ CLI_ID_LENGTH - CLI_MARGIN_LENGTH),
+ StringUtils.substring(sg.name(), 0,
+ CLI_NAME_LENGTH - CLI_MARGIN_LENGTH),
+ StringUtils.substring(String.valueOf(sg.rules().size()), 0,
+ CLI_NUMBER_LENGTH - CLI_MARGIN_LENGTH)
+ );
+ }
+ }
+ }
+
+ private String json(List<KubevirtSecurityGroup> sgs) {
+ ObjectMapper mapper = new ObjectMapper();
+ ArrayNode result = mapper.createArrayNode();
+
+ for (KubevirtSecurityGroup sg : sgs) {
+ result.add(jsonForEntity(sg, KubevirtSecurityGroup.class));
+ }
+
+ return prettyJson(mapper, result.toString());
+ }
+}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtSecurityGroupCompleter.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtSecurityGroupCompleter.java
new file mode 100644
index 0000000..893605f
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtSecurityGroupCompleter.java
@@ -0,0 +1,52 @@
+/*
+ * 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.cli;
+
+import org.apache.karaf.shell.api.action.lifecycle.Service;
+import org.apache.karaf.shell.api.console.CommandLine;
+import org.apache.karaf.shell.api.console.Completer;
+import org.apache.karaf.shell.api.console.Session;
+import org.apache.karaf.shell.support.completers.StringsCompleter;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupService;
+
+import java.util.List;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.stream.Collectors;
+
+import static org.onosproject.cli.AbstractShellCommand.get;
+
+/**
+ * Kubevirt security group name completer.
+ */
+@Service
+public class KubevirtSecurityGroupCompleter implements Completer {
+ @Override
+ public int complete(Session session, CommandLine commandLine, List<String> candidates) {
+ StringsCompleter delegate = new StringsCompleter();
+ KubevirtSecurityGroupService service = get(KubevirtSecurityGroupService.class);
+
+ Set<String> sgNames = service.securityGroups().stream()
+ .map(KubevirtSecurityGroup::name).collect(Collectors.toSet());
+
+ SortedSet<String> strings = delegate.getStrings();
+
+ strings.addAll(sgNames);
+
+ return delegate.complete(session, commandLine, candidates);
+ }
+}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtShowPodCommand.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtShowPodCommand.java
index 514e651..c8bf0a9 100644
--- a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtShowPodCommand.java
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtShowPodCommand.java
@@ -85,6 +85,7 @@
print(" Pull Policy: %s", container.getImagePullPolicy());
print(" Commands: %s", container.getCommand());
print(" Args: %s", container.getArgs());
+ counter++;
}
}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtShowSecurityGroupCommand.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtShowSecurityGroupCommand.java
new file mode 100644
index 0000000..4ac752d
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/cli/KubevirtShowSecurityGroupCommand.java
@@ -0,0 +1,95 @@
+/*
+ * 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.cli;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.karaf.shell.api.action.Command;
+import org.apache.karaf.shell.api.action.Completion;
+import org.apache.karaf.shell.api.action.Option;
+import org.apache.karaf.shell.api.action.lifecycle.Service;
+import org.onosproject.cli.AbstractShellCommand;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupRule;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupService;
+
+import java.util.List;
+
+import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.prettyJson;
+
+/**
+ * Show a detailed security group info.
+ */
+@Service
+@Command(scope = "onos", name = "kubevirt-security-group",
+ description = "Displays a security group details")
+public class KubevirtShowSecurityGroupCommand extends AbstractShellCommand {
+
+ @Option(name = "--name",
+ description = "Filter security group by specific name", multiValued = true)
+ @Completion(KubevirtSecurityGroupCompleter.class)
+ private List<String> names;
+
+ @Override
+ protected void doExecute() throws Exception {
+ KubevirtSecurityGroupService service = get(KubevirtSecurityGroupService.class);
+
+ if (names == null || names.size() == 0) {
+ print("Need to specify at least one security group name using --name option.");
+ return;
+ }
+
+ for (String name : names) {
+ KubevirtSecurityGroup sg = service.securityGroups().stream()
+ .filter(s -> s.name().equals(name))
+ .findAny().orElse(null);
+ if (sg == null) {
+ print("Unable to find %s", name);
+ continue;
+ }
+
+ if (outputJson()) {
+ print("%s", json(sg));
+ } else {
+ printSecurityGroup(sg);
+ }
+ }
+ }
+
+ private void printSecurityGroup(KubevirtSecurityGroup sg) {
+ print("Name: %s", sg.name());
+ print(" ID: %s", sg.id());
+ print(" Description: %s", sg.description());
+
+ int counter = 1;
+ for (KubevirtSecurityGroupRule rule : sg.rules()) {
+ print(" Rule #%d:", counter);
+ print(" ID: %s", rule.id());
+ print(" Direction: %s", rule.direction());
+ print(" EtherType: %s", rule.etherType());
+ print(" Protocol: %s", rule.protocol());
+ print(" PortRangeMax: %s", rule.portRangeMax());
+ print(" PortRangeMin: %s", rule.portRangeMin());
+ print(" RemoteIpPrefix: %s", rule.remoteIpPrefix());
+ print(" RemoteGroupID: %s", rule.remoteGroupId());
+ counter++;
+ }
+ }
+
+ private String json(KubevirtSecurityGroup sg) {
+ return prettyJson(new ObjectMapper(),
+ jsonForEntity(sg, KubevirtSecurityGroup.class).toString());
+ }
+}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/codec/KubevirtSecurityGroupCodec.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/codec/KubevirtSecurityGroupCodec.java
new file mode 100644
index 0000000..fe290f9
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/codec/KubevirtSecurityGroupCodec.java
@@ -0,0 +1,108 @@
+/*
+ * 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.codec;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.onosproject.codec.CodecContext;
+import org.onosproject.codec.JsonCodec;
+import org.onosproject.kubevirtnetworking.api.DefaultKubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupRule;
+import org.slf4j.Logger;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onlab.util.Tools.nullIsIllegal;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Kubevirt security group codec used for serializing and de-serializing JSON string.
+ */
+public final class KubevirtSecurityGroupCodec extends JsonCodec<KubevirtSecurityGroup> {
+
+ private final Logger log = getLogger(getClass());
+
+ private static final String ID = "id";
+ private static final String NAME = "name";
+ private static final String DESCRIPTION = "description";
+ private static final String RULES = "rules";
+
+ private static final String MISSING_MESSAGE = " is required in KubevirtSecurityGroup";
+
+ @Override
+ public ObjectNode encode(KubevirtSecurityGroup sg, CodecContext context) {
+ checkNotNull(sg, "Kubevirt Security Group cannot be null");
+
+ ObjectNode result = context.mapper().createObjectNode()
+ .put(ID, sg.id())
+ .put(NAME, sg.name());
+
+ if (sg.description() != null) {
+ result.put(DESCRIPTION, sg.description());
+ }
+
+ if (sg.rules() != null && !sg.rules().isEmpty()) {
+ ArrayNode rules = context.mapper().createArrayNode();
+ sg.rules().forEach(rule -> {
+ ObjectNode ruleJson = context.codec(
+ KubevirtSecurityGroupRule.class).encode(rule, context);
+ rules.add(ruleJson);
+ });
+ result.set(RULES, rules);
+ }
+
+ return result;
+ }
+
+ @Override
+ public KubevirtSecurityGroup decode(ObjectNode json, CodecContext context) {
+ if (json == null || !json.isObject()) {
+ return null;
+ }
+
+ String id = nullIsIllegal(json.get(ID).asText(), ID + MISSING_MESSAGE);
+ String name = nullIsIllegal(json.get(NAME).asText(), NAME + MISSING_MESSAGE);
+
+ KubevirtSecurityGroup.Builder builder = DefaultKubevirtSecurityGroup.builder()
+ .id(id)
+ .name(name);
+
+ JsonNode description = json.get(DESCRIPTION);
+ if (description != null) {
+ builder.description(description.asText());
+ }
+
+ JsonNode rulesJson = json.get(RULES);
+ if (rulesJson != null) {
+ Set<KubevirtSecurityGroupRule> rules = new HashSet<>();
+ IntStream.range(0, rulesJson.size())
+ .forEach(i -> {
+ ObjectNode ruleJson = get(rulesJson, i);
+ KubevirtSecurityGroupRule rule = context.codec(
+ KubevirtSecurityGroupRule.class).decode(ruleJson, context);
+ rules.add(rule);
+ });
+ builder.rules(rules);
+ }
+
+ return builder.build();
+ }
+}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/codec/KubevirtSecurityGroupRuleCodec.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/codec/KubevirtSecurityGroupRuleCodec.java
new file mode 100644
index 0000000..b63b555
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/codec/KubevirtSecurityGroupRuleCodec.java
@@ -0,0 +1,135 @@
+/*
+ * 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.codec;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.onlab.packet.IpPrefix;
+import org.onosproject.codec.CodecContext;
+import org.onosproject.codec.JsonCodec;
+import org.onosproject.kubevirtnetworking.api.DefaultKubevirtSecurityGroupRule;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupRule;
+import org.slf4j.Logger;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onlab.util.Tools.nullIsIllegal;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Kubevirt security group rule codec used for serializing and de-serializing JSON string.
+ */
+public final class KubevirtSecurityGroupRuleCodec extends JsonCodec<KubevirtSecurityGroupRule> {
+
+ private final Logger log = getLogger(getClass());
+
+ private static final String ID = "id";
+ private static final String SECURITY_GROUP_ID = "securityGroupId";
+ private static final String DIRECTION = "direction";
+ private static final String ETHER_TYPE = "etherType";
+ private static final String PORT_RANGE_MAX = "portRangeMax";
+ private static final String PORT_RANGE_MIN = "portRangeMin";
+ private static final String PROTOCOL = "protocol";
+ private static final String REMOTE_IP_PREFIX = "remoteIpPrefix";
+ private static final String REMOTE_GROUP_ID = "remoteGroupId";
+
+ private static final String MISSING_MESSAGE = " is required in KubevirtSecurityGroupRule";
+
+ @Override
+ public ObjectNode encode(KubevirtSecurityGroupRule sgRule, CodecContext context) {
+ checkNotNull(sgRule, "Kubevirt security group rule cannot be null");
+
+ ObjectNode result = context.mapper().createObjectNode()
+ .put(ID, sgRule.id())
+ .put(SECURITY_GROUP_ID, sgRule.securityGroupId())
+ .put(DIRECTION, sgRule.direction());
+
+ if (sgRule.etherType() != null) {
+ result.put(ETHER_TYPE, sgRule.etherType());
+ }
+
+ if (sgRule.portRangeMax() != null) {
+ result.put(PORT_RANGE_MAX, sgRule.portRangeMax());
+ }
+
+ if (sgRule.portRangeMin() != null) {
+ result.put(PORT_RANGE_MIN, sgRule.portRangeMin());
+ }
+
+ if (sgRule.protocol() != null) {
+ result.put(PROTOCOL, sgRule.protocol());
+ }
+
+ if (sgRule.remoteIpPrefix() != null) {
+ result.put(REMOTE_IP_PREFIX, sgRule.remoteIpPrefix().toString());
+ }
+
+ if (sgRule.remoteGroupId() != null) {
+ result.put(REMOTE_GROUP_ID, sgRule.remoteGroupId());
+ }
+
+ return result;
+ }
+
+ @Override
+ public KubevirtSecurityGroupRule decode(ObjectNode json, CodecContext context) {
+ if (json == null || !json.isObject()) {
+ return null;
+ }
+
+ String id = nullIsIllegal(json.get(ID).asText(), ID + MISSING_MESSAGE);
+ String securityGroupId = nullIsIllegal(json.get(SECURITY_GROUP_ID).asText(),
+ SECURITY_GROUP_ID + MISSING_MESSAGE);
+ String direction = nullIsIllegal(json.get(DIRECTION).asText(),
+ DIRECTION + MISSING_MESSAGE);
+
+ KubevirtSecurityGroupRule.Builder builder = DefaultKubevirtSecurityGroupRule.builder()
+ .id(id)
+ .securityGroupId(securityGroupId)
+ .direction(direction);
+
+ JsonNode etherType = json.get(ETHER_TYPE);
+ if (etherType != null) {
+ builder.etherType(etherType.asText());
+ }
+
+ JsonNode portRangeMax = json.get(PORT_RANGE_MAX);
+ if (portRangeMax != null) {
+ builder.portRangeMax(portRangeMax.asInt());
+ }
+
+ JsonNode portRangeMin = json.get(PORT_RANGE_MIN);
+ if (portRangeMin != null) {
+ builder.portRangeMin(portRangeMin.asInt());
+ }
+
+ JsonNode protocol = json.get(PROTOCOL);
+ if (protocol != null) {
+ builder.protocol(protocol.asText());
+ }
+
+ JsonNode remoteIpPrefix = json.get(REMOTE_IP_PREFIX);
+ if (remoteIpPrefix != null) {
+ builder.remoteIpPrefix(IpPrefix.valueOf(remoteIpPrefix.asText()));
+ }
+
+ JsonNode remoteGroupId = json.get(REMOTE_GROUP_ID);
+ if (remoteGroupId != null) {
+ builder.remoteGroupId(remoteGroupId.asText());
+ }
+
+ return builder.build();
+ }
+}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/DistributedKubevirtSecurityGroupStore.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/DistributedKubevirtSecurityGroupStore.java
new file mode 100644
index 0000000..96bbb16
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/DistributedKubevirtSecurityGroupStore.java
@@ -0,0 +1,203 @@
+/*
+ * 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.google.common.collect.ImmutableSet;
+import org.onlab.util.KryoNamespace;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.kubevirtnetworking.api.DefaultKubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.DefaultKubevirtSecurityGroupRule;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupEvent;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupRule;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupStore;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupStoreDelegate;
+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.Collection;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+
+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.kubevirtnetworking.api.KubevirtSecurityGroupEvent.Type.KUBEVIRT_SECURITY_GROUP_CREATED;
+import static org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupEvent.Type.KUBEVIRT_SECURITY_GROUP_REMOVED;
+import static org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupEvent.Type.KUBEVIRT_SECURITY_GROUP_RULE_CREATED;
+import static org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupEvent.Type.KUBEVIRT_SECURITY_GROUP_RULE_REMOVED;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Implementation of kubevirt security group store using consistent map.
+ */
+@Component(immediate = true, service = KubevirtSecurityGroupStore.class)
+public class DistributedKubevirtSecurityGroupStore
+ extends AbstractStore<KubevirtSecurityGroupEvent, KubevirtSecurityGroupStoreDelegate>
+ implements KubevirtSecurityGroupStore {
+
+ 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.kubevirtnetwork";
+
+
+ private static final KryoNamespace SERIALIZER_KUBEVIRT_SG = KryoNamespace.newBuilder()
+ .register(KryoNamespaces.API)
+ .register(KubevirtSecurityGroup.class)
+ .register(KubevirtSecurityGroupRule.class)
+ .register(DefaultKubevirtSecurityGroup.class)
+ .register(DefaultKubevirtSecurityGroupRule.class)
+ .register(Collection.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, KubevirtSecurityGroup> securityGroupListener =
+ new KubevirtSecurityGroupMapListener();
+
+ private ConsistentMap<String, KubevirtSecurityGroup> sgStore;
+
+ @Activate
+ protected void activate() {
+ ApplicationId appId = coreService.registerApplication(APP_ID);
+ sgStore = storageService.<String, KubevirtSecurityGroup>consistentMapBuilder()
+ .withSerializer(Serializer.using(SERIALIZER_KUBEVIRT_SG))
+ .withName("kubevirt-securitygroupstore")
+ .withApplicationId(appId)
+ .build();
+ sgStore.addListener(securityGroupListener);
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ sgStore.removeListener(securityGroupListener);
+ eventExecutor.shutdown();
+ log.info("Stopped");
+ }
+
+ @Override
+ public void createSecurityGroup(KubevirtSecurityGroup sg) {
+ sgStore.compute(sg.id(), (id, existing) -> {
+ final String error = sg.id() + ERR_DUPLICATE;
+ checkArgument(existing == null, error);
+ return sg;
+ });
+ }
+
+ @Override
+ public void updateSecurityGroup(KubevirtSecurityGroup sg) {
+ sgStore.compute(sg.id(), (id, existing) -> {
+ final String error = sg.id() + ERR_NOT_FOUND;
+ checkArgument(existing != null, error);
+ return sg;
+ });
+ }
+
+ @Override
+ public KubevirtSecurityGroup removeSecurityGroup(String sgId) {
+ Versioned<KubevirtSecurityGroup> sg = sgStore.remove(sgId);
+ if (sg == null) {
+ final String error = sgId + ERR_NOT_FOUND;
+ throw new IllegalArgumentException(error);
+ }
+ return sg.value();
+ }
+
+ @Override
+ public KubevirtSecurityGroup securityGroup(String sgId) {
+ return sgStore.asJavaMap().get(sgId);
+ }
+
+ @Override
+ public Set<KubevirtSecurityGroup> securityGroups() {
+ return ImmutableSet.copyOf(sgStore.asJavaMap().values());
+ }
+
+ @Override
+ public void clear() {
+ sgStore.clear();
+ }
+
+ private class KubevirtSecurityGroupMapListener
+ implements MapEventListener<String, KubevirtSecurityGroup> {
+
+ @Override
+ public void event(MapEvent<String, KubevirtSecurityGroup> event) {
+
+ switch (event.type()) {
+ case INSERT:
+ log.debug("Kubevirt security group created {}", event.newValue());
+ eventExecutor.execute(() ->
+ notifyDelegate(new KubevirtSecurityGroupEvent(
+ KUBEVIRT_SECURITY_GROUP_CREATED, event.newValue().value())));
+ break;
+ case UPDATE:
+ log.debug("Kubevirt security group updated {}", event.newValue());
+ eventExecutor.execute(() -> processUpdate(
+ event.oldValue().value(),
+ event.newValue().value()));
+ break;
+ case REMOVE:
+ log.debug("Kubevirt security group removed {}", event.oldValue());
+ eventExecutor.execute(() ->
+ notifyDelegate(new KubevirtSecurityGroupEvent(
+ KUBEVIRT_SECURITY_GROUP_REMOVED, event.oldValue().value())));
+ break;
+ default:
+ // do nothing
+ break;
+ }
+ }
+
+ private void processUpdate(KubevirtSecurityGroup oldSg, KubevirtSecurityGroup newSg) {
+ Set<String> oldSgRuleIds = oldSg.rules().stream()
+ .map(KubevirtSecurityGroupRule::id).collect(Collectors.toSet());
+ Set<String> newSgRuleIds = newSg.rules().stream()
+ .map(KubevirtSecurityGroupRule::id).collect(Collectors.toSet());
+
+ oldSg.rules().stream().filter(sgRule -> !newSgRuleIds.contains(sgRule.id()))
+ .forEach(sgRule -> notifyDelegate(new KubevirtSecurityGroupEvent(
+ KUBEVIRT_SECURITY_GROUP_RULE_REMOVED, newSg, sgRule)));
+ newSg.rules().stream().filter(sgRule -> !oldSgRuleIds.contains(sgRule.id()))
+ .forEach(sgRule -> notifyDelegate(new KubevirtSecurityGroupEvent(
+ KUBEVIRT_SECURITY_GROUP_RULE_CREATED, newSg, sgRule)));
+ }
+ }
+}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSecurityGroupManager.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSecurityGroupManager.java
new file mode 100644
index 0000000..b92db05
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSecurityGroupManager.java
@@ -0,0 +1,232 @@
+/*
+ * 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.google.common.base.Strings;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.event.ListenerRegistry;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupAdminService;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupEvent;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupListener;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupRule;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupService;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupStore;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupStoreDelegate;
+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.HashSet;
+import java.util.Objects;
+import java.util.Set;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onosproject.kubevirtnetworking.api.Constants.KUBEVIRT_NETWORKING_APP_ID;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Provides implementation of administering and interfacing kubevirt security groups.
+ */
+@Component(
+ immediate = true,
+ service = {KubevirtSecurityGroupAdminService.class, KubevirtSecurityGroupService.class }
+)
+public class KubevirtSecurityGroupManager
+ extends ListenerRegistry<KubevirtSecurityGroupEvent, KubevirtSecurityGroupListener>
+ implements KubevirtSecurityGroupAdminService, KubevirtSecurityGroupService {
+
+ protected final Logger log = getLogger(getClass());
+
+ private static final String MSG_SG = "Kubevirt security group %s %s";
+ private static final String MSG_SG_RULE = "Kubevirt security group rule %s %s";
+
+ private static final String MSG_CREATED = "created";
+ private static final String MSG_REMOVED = "removed";
+
+ private static final String ERR_NULL_SG =
+ "Kubevirt security group cannot be null";
+ private static final String ERR_NULL_SG_ID =
+ "Kubevirt security group ID cannot be null";
+ private static final String ERR_NULL_SG_RULE =
+ "Kubevirt security group rule cannot be null";
+ private static final String ERR_NULL_SG_RULE_ID =
+ "Kubevirt security group rule ID cannot be null";
+ private static final String ERR_NOT_FOUND = "not found";
+ private static final String ERR_DUPLICATE = "already exist";
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected CoreService coreService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected KubevirtSecurityGroupStore sgStore;
+
+ private final KubevirtSecurityGroupStoreDelegate
+ delegate = new InternalSecurityGroupStoreDelegate();
+
+ private ApplicationId appId;
+ private boolean useSecurityGroup = false;
+
+ @Activate
+ protected void activate() {
+ appId = coreService.registerApplication(KUBEVIRT_NETWORKING_APP_ID);
+
+ sgStore.setDelegate(delegate);
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ sgStore.unsetDelegate(delegate);
+ log.info("Stopped");
+ }
+
+ @Override
+ public void createSecurityGroup(KubevirtSecurityGroup sg) {
+ checkNotNull(sg, ERR_NULL_SG);
+ checkArgument(!Strings.isNullOrEmpty(sg.id()), ERR_NULL_SG_ID);
+
+ sgStore.createSecurityGroup(sg);
+ log.info(String.format(MSG_SG, sg.id(), MSG_CREATED));
+ }
+
+ @Override
+ public void updateSecurityGroup(KubevirtSecurityGroup sg) {
+ checkNotNull(sg, ERR_NULL_SG);
+ checkArgument(!Strings.isNullOrEmpty(sg.id()), ERR_NULL_SG_ID);
+
+ sgStore.updateSecurityGroup(sg);
+ }
+
+ @Override
+ public void removeSecurityGroup(String sgId) {
+ checkArgument(!Strings.isNullOrEmpty(sgId), ERR_NULL_SG_ID);
+
+ sgStore.removeSecurityGroup(sgId);
+ log.info(String.format(MSG_SG, sgId, MSG_REMOVED));
+ }
+
+ @Override
+ public void createSecurityGroupRule(KubevirtSecurityGroupRule sgRule) {
+ checkNotNull(sgRule, ERR_NULL_SG_RULE);
+ checkArgument(!Strings.isNullOrEmpty(sgRule.id()), ERR_NULL_SG_RULE_ID);
+ checkArgument(!Strings.isNullOrEmpty(sgRule.securityGroupId()), ERR_NULL_SG_ID);
+
+ synchronized (this) {
+ KubevirtSecurityGroup sg = securityGroup(sgRule.securityGroupId());
+ if (sg == null) {
+ final String error = String.format(MSG_SG,
+ sgRule.securityGroupId(), ERR_NOT_FOUND);
+ throw new IllegalStateException(error);
+ }
+
+ if (sg.rules().stream().anyMatch(rule -> Objects.equals(rule.id(), sgRule.id()))) {
+ final String error = String.format(MSG_SG_RULE, sgRule.securityGroupId(), ERR_DUPLICATE);
+ throw new IllegalStateException(error);
+ }
+
+ // FIXME we cannot add element to extend list
+ Set<KubevirtSecurityGroupRule> updatedSgRules = new HashSet<>(sg.rules());
+ updatedSgRules.add(sgRule);
+ sgStore.updateSecurityGroup(sg.updateRules(updatedSgRules));
+ }
+
+ log.info(String.format(MSG_SG_RULE, sgRule.id(), MSG_CREATED));
+ }
+
+ @Override
+ public void removeSecurityGroupRule(String sgRuleId) {
+ checkArgument(!Strings.isNullOrEmpty(sgRuleId), ERR_NULL_SG_RULE_ID);
+
+ synchronized (this) {
+ KubevirtSecurityGroupRule sgRule = securityGroupRule(sgRuleId);
+ if (sgRule == null) {
+ final String error = String.format(MSG_SG_RULE, sgRuleId, ERR_NOT_FOUND);
+ throw new IllegalStateException(error);
+ }
+
+ KubevirtSecurityGroup sg = securityGroup(sgRule.securityGroupId());
+ if (sg == null) {
+ final String error = String.format(MSG_SG,
+ sgRule.securityGroupId(), ERR_NOT_FOUND);
+ throw new IllegalStateException(error);
+ }
+
+ if (sg.rules().stream().noneMatch(rule -> Objects.equals(rule.id(), sgRule.id()))) {
+ final String error = String.format(MSG_SG_RULE,
+ sgRule.securityGroupId(), ERR_NOT_FOUND);
+ throw new IllegalStateException(error);
+ }
+
+ Set<KubevirtSecurityGroupRule> updatedSgRules = new HashSet<>(sg.rules());
+ updatedSgRules.removeIf(r -> r.id().equals(sgRuleId));
+ sgStore.updateSecurityGroup(sg.updateRules(updatedSgRules));
+ }
+
+ log.info(String.format(MSG_SG_RULE, sgRuleId, MSG_REMOVED));
+ }
+
+ @Override
+ public void clear() {
+ sgStore.clear();
+ }
+
+ @Override
+ public Set<KubevirtSecurityGroup> securityGroups() {
+ return sgStore.securityGroups();
+ }
+
+ @Override
+ public KubevirtSecurityGroup securityGroup(String sgId) {
+ checkArgument(!Strings.isNullOrEmpty(sgId), ERR_NULL_SG_ID);
+ return sgStore.securityGroup(sgId);
+ }
+
+ @Override
+ public boolean isSecurityGroupEnabled() {
+ return useSecurityGroup;
+ }
+
+ @Override
+ public void setSecurityGroupEnabled(boolean option) {
+ useSecurityGroup = option;
+ }
+
+ @Override
+ public KubevirtSecurityGroupRule securityGroupRule(String sgRuleId) {
+ return sgStore.securityGroups().stream()
+ .flatMap(sg -> sg.rules().stream())
+ .filter(sgRule -> Objects.equals(sgRule.id(), sgRuleId))
+ .findAny().orElse(null);
+ }
+
+ private class InternalSecurityGroupStoreDelegate
+ implements KubevirtSecurityGroupStoreDelegate {
+
+ @Override
+ public void notify(KubevirtSecurityGroupEvent event) {
+ if (event != null) {
+ log.trace("send kubevirt security group event {}", event);
+ process(event);
+ }
+ }
+ }
+}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSecurityGroupWatcher.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSecurityGroupWatcher.java
new file mode 100644
index 0000000..2fb53a9
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/impl/KubevirtSecurityGroupWatcher.java
@@ -0,0 +1,364 @@
+/*
+ * 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.ObjectNode;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.Watcher;
+import io.fabric8.kubernetes.client.WatcherException;
+import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext;
+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.AbstractWatcher;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupAdminService;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupRule;
+import org.onosproject.kubevirtnode.api.KubevirtApiConfigEvent;
+import org.onosproject.kubevirtnode.api.KubevirtApiConfigListener;
+import org.onosproject.kubevirtnode.api.KubevirtApiConfigService;
+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.io.IOException;
+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.kubevirtnetworking.api.Constants.KUBEVIRT_NETWORKING_APP_ID;
+import static org.onosproject.kubevirtnetworking.util.KubevirtNetworkingUtil.k8sClient;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Kubevirt security group watcher used for feeding kubevirt security group information.
+ */
+@Component(immediate = true)
+public class KubevirtSecurityGroupWatcher extends AbstractWatcher {
+
+ 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 KubevirtSecurityGroupAdminService adminService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY)
+ protected KubevirtApiConfigService configService;
+
+ private final ExecutorService eventExecutor = newSingleThreadExecutor(
+ groupedThreads(this.getClass().getSimpleName(), "event-handler"));
+
+ private final InternalSecurityGroupWatcher
+ sgWatcher = new InternalSecurityGroupWatcher();
+ private final InternalSecurityGroupRuleWatcher
+ sgrWatcher = new InternalSecurityGroupRuleWatcher();
+ private final InternalKubevirtApiConfigListener
+ configListener = new InternalKubevirtApiConfigListener();
+
+ CustomResourceDefinitionContext securityGroupCrdCxt = new CustomResourceDefinitionContext
+ .Builder()
+ .withGroup("kubevirt.io")
+ .withScope("Cluster")
+ .withVersion("v1")
+ .withPlural("securitygroups")
+ .build();
+
+ CustomResourceDefinitionContext securityGroupRuleCrdCxt = new CustomResourceDefinitionContext
+ .Builder()
+ .withGroup("kubevirt.io")
+ .withScope("Cluster")
+ .withVersion("v1")
+ .withPlural("securitygrouprules")
+ .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 instantiateSgWatcher() {
+ KubernetesClient client = k8sClient(configService);
+
+ if (client != null) {
+ try {
+ client.customResource(securityGroupCrdCxt).watch(sgWatcher);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ private void instantiateSgrWatcher() {
+ KubernetesClient client = k8sClient(configService);
+
+ if (client != null) {
+ try {
+ client.customResource(securityGroupRuleCrdCxt).watch(sgrWatcher);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ private KubevirtSecurityGroup parseSecurityGroup(String resource) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode json = mapper.readTree(resource);
+ ObjectNode spec = (ObjectNode) json.get("spec");
+ return codec(KubevirtSecurityGroup.class).decode(spec, this);
+ } catch (IOException e) {
+ log.error("Failed to parse kubevirt security group object");
+ }
+
+ return null;
+ }
+
+ private KubevirtSecurityGroupRule parseSecurityGroupRule(String resource) {
+ try {
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode json = mapper.readTree(resource);
+ ObjectNode spec = (ObjectNode) json.get("spec");
+ return codec(KubevirtSecurityGroupRule.class).decode(spec, this);
+ } catch (IOException e) {
+ log.error("Failed to parse kubevirt security group rule object");
+ }
+
+ return null;
+ }
+
+ 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;
+ }
+
+ instantiateSgWatcher();
+ instantiateSgrWatcher();
+ }
+ }
+
+ private class InternalSecurityGroupWatcher implements Watcher<String> {
+
+ @Override
+ public void eventReceived(Action action, String resource) {
+ switch (action) {
+ case ADDED:
+ eventExecutor.execute(() -> processAddition(resource));
+ break;
+ case MODIFIED:
+ eventExecutor.execute(() -> processModification(resource));
+ break;
+ case DELETED:
+ eventExecutor.execute(() -> processDeletion(resource));
+ break;
+ default:
+ // do nothing
+ break;
+ }
+ }
+
+ @Override
+ public void onClose(WatcherException e) {
+ // due to the bugs in fabric8, the watcher might be closed,
+ // we will re-instantiate the watcher in this case
+ // FIXME: https://github.com/fabric8io/kubernetes-client/issues/2135
+ log.warn("Security Group watcher OnClose, re-instantiate the watcher...");
+
+ instantiateSgWatcher();
+ }
+
+ private void processAddition(String resource) {
+ if (!isMaster()) {
+ return;
+ }
+
+ KubevirtSecurityGroup sg = parseSecurityGroup(resource);
+
+ if (sg != null) {
+ log.trace("Process Security Group {} creating event from API server.", sg.name());
+
+ if (adminService.securityGroup(sg.id()) == null) {
+ adminService.createSecurityGroup(sg);
+ }
+ }
+ }
+
+ private void processModification(String resource) {
+ if (!isMaster()) {
+ return;
+ }
+
+ KubevirtSecurityGroup sg = parseSecurityGroup(resource);
+
+ if (sg != null) {
+ log.trace("Process Security Group {} updating event from API server.", sg.name());
+
+ // since Security Group CRD does not contains any rules information,
+ // we need to manually add all rules from original to the updated one
+ KubevirtSecurityGroup orig = adminService.securityGroup(sg.id());
+ if (orig != null) {
+ KubevirtSecurityGroup updated = sg.updateRules(orig.rules());
+ adminService.updateSecurityGroup(updated);
+ }
+ }
+ }
+
+ private void processDeletion(String resource) {
+ if (!isMaster()) {
+ return;
+ }
+
+ KubevirtSecurityGroup sg = parseSecurityGroup(resource);
+
+ if (sg != null) {
+ log.trace("Process Security Group {} removal event from API server.", sg.name());
+
+ adminService.removeSecurityGroup(sg.id());
+ }
+ }
+ }
+
+ private class InternalSecurityGroupRuleWatcher implements Watcher<String> {
+
+ @Override
+ public void eventReceived(Action action, String resource) {
+ switch (action) {
+ case ADDED:
+ eventExecutor.execute(() -> processAddition(resource));
+ break;
+ case MODIFIED:
+ eventExecutor.execute(() -> processModification(resource));
+ break;
+ case DELETED:
+ eventExecutor.execute(() -> processDeletion(resource));
+ break;
+ default:
+ // do nothing
+ break;
+ }
+ }
+
+ @Override
+ public void onClose(WatcherException e) {
+ // due to the bugs in fabric8, the watcher might be closed,
+ // we will re-instantiate the watcher in this case
+ // FIXME: https://github.com/fabric8io/kubernetes-client/issues/2135
+ log.warn("Security Group Rule watcher OnClose, re-instantiate the watcher...");
+
+ instantiateSgrWatcher();
+ }
+
+ private void processAddition(String resource) {
+ if (!isMaster()) {
+ return;
+ }
+
+ KubevirtSecurityGroupRule sgr = parseSecurityGroupRule(resource);
+
+ if (sgr != null) {
+ log.trace("Process Security Group Rule {} creating event from API server.", sgr.id());
+
+ if (adminService.securityGroupRule(sgr.id()) == null) {
+ adminService.createSecurityGroupRule(sgr);
+ }
+ }
+ }
+
+ private void processModification(String resource) {
+ if (!isMaster()) {
+ return;
+ }
+
+ // we do not handle the update case, as we assume the security group rule
+ // object is immutable
+ }
+
+ private void processDeletion(String resource) {
+ if (!isMaster()) {
+ return;
+ }
+
+ KubevirtSecurityGroupRule sgr = parseSecurityGroupRule(resource);
+
+ if (sgr != null) {
+ log.trace("Process Security Group Rule {} removal event from API server.", sgr.id());
+
+ adminService.removeSecurityGroupRule(sgr.id());
+ }
+ }
+ }
+
+ private boolean isMaster() {
+ return Objects.equals(localNodeId, leadershipService.getLeader(appId.name()));
+ }
+}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtFloatingIpsWebResource.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtFloatingIpsWebResource.java
index 6314a31..ec41622 100644
--- a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtFloatingIpsWebResource.java
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtFloatingIpsWebResource.java
@@ -30,11 +30,13 @@
/**
* Handles REST API call for kubevirt floating IPs.
*/
-@Path("floatingips")
+@Path("floating-ip")
public class KubevirtFloatingIpsWebResource extends AbstractWebResource {
protected final Logger log = LoggerFactory.getLogger(getClass());
+ private static final String FLOATING_IPS = "floating-ips";
+
/**
* Returns set of all floating IPs.
*
@@ -46,6 +48,6 @@
public Response getFloatingIps() {
KubevirtRouterService service = get(KubevirtRouterService.class);
final Iterable<KubevirtFloatingIp> fips = service.floatingIps();
- return ok(encodeArray(KubevirtFloatingIp.class, "floatingips", fips)).build();
+ return ok(encodeArray(KubevirtFloatingIp.class, FLOATING_IPS, fips)).build();
}
}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtNetworkingCodecRegister.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtNetworkingCodecRegister.java
index 2754e7c..b8f27e3 100644
--- a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtNetworkingCodecRegister.java
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtNetworkingCodecRegister.java
@@ -22,12 +22,16 @@
import org.onosproject.kubevirtnetworking.api.KubevirtNetwork;
import org.onosproject.kubevirtnetworking.api.KubevirtPort;
import org.onosproject.kubevirtnetworking.api.KubevirtRouter;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupRule;
import org.onosproject.kubevirtnetworking.codec.KubevirtFloatingIpCodec;
import org.onosproject.kubevirtnetworking.codec.KubevirtHostRouteCodec;
import org.onosproject.kubevirtnetworking.codec.KubevirtIpPoolCodec;
import org.onosproject.kubevirtnetworking.codec.KubevirtNetworkCodec;
import org.onosproject.kubevirtnetworking.codec.KubevirtPortCodec;
import org.onosproject.kubevirtnetworking.codec.KubevirtRouterCodec;
+import org.onosproject.kubevirtnetworking.codec.KubevirtSecurityGroupCodec;
+import org.onosproject.kubevirtnetworking.codec.KubevirtSecurityGroupRuleCodec;
import org.osgi.service.component.annotations.Activate;
import org.osgi.service.component.annotations.Component;
import org.osgi.service.component.annotations.Deactivate;
@@ -57,6 +61,8 @@
codecService.registerCodec(KubevirtPort.class, new KubevirtPortCodec());
codecService.registerCodec(KubevirtRouter.class, new KubevirtRouterCodec());
codecService.registerCodec(KubevirtFloatingIp.class, new KubevirtFloatingIpCodec());
+ codecService.registerCodec(KubevirtSecurityGroup.class, new KubevirtSecurityGroupCodec());
+ codecService.registerCodec(KubevirtSecurityGroupRule.class, new KubevirtSecurityGroupRuleCodec());
log.info("Started");
}
@@ -70,6 +76,8 @@
codecService.unregisterCodec(KubevirtPort.class);
codecService.unregisterCodec(KubevirtRouter.class);
codecService.unregisterCodec(KubevirtFloatingIp.class);
+ codecService.unregisterCodec(KubevirtSecurityGroup.class);
+ codecService.unregisterCodec(KubevirtSecurityGroupRule.class);
log.info("Stopped");
}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtNetworkingWebApplication.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtNetworkingWebApplication.java
index d2a45af..139b92d 100644
--- a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtNetworkingWebApplication.java
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtNetworkingWebApplication.java
@@ -28,7 +28,9 @@
return getClasses(
KubevirtNetworkWebResource.class,
KubevirtManagementWebResource.class,
- KubevirtRouterWebResource.class
+ KubevirtRouterWebResource.class,
+ KubevirtFloatingIpsWebResource.class,
+ KubevirtSecurityGroupWebResource.class
);
}
}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtRouterWebResource.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtRouterWebResource.java
index 731b0d4..8a40f51 100644
--- a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtRouterWebResource.java
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtRouterWebResource.java
@@ -34,6 +34,7 @@
public class KubevirtRouterWebResource extends AbstractWebResource {
protected final Logger log = LoggerFactory.getLogger(getClass());
+ private static final String ROUTERS = "routers";
/**
* Returns set of all routers.
@@ -46,6 +47,6 @@
public Response getRouters() {
KubevirtRouterService service = get(KubevirtRouterService.class);
final Iterable<KubevirtRouter> routers = service.routers();
- return ok(encodeArray(KubevirtRouter.class, "routers", routers)).build();
+ return ok(encodeArray(KubevirtRouter.class, ROUTERS, routers)).build();
}
}
diff --git a/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtSecurityGroupWebResource.java b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtSecurityGroupWebResource.java
new file mode 100644
index 0000000..02acd87
--- /dev/null
+++ b/apps/kubevirt-networking/app/src/main/java/org/onosproject/kubevirtnetworking/web/KubevirtSecurityGroupWebResource.java
@@ -0,0 +1,52 @@
+/*
+ * 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.web;
+
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroup;
+import org.onosproject.kubevirtnetworking.api.KubevirtSecurityGroupService;
+import org.onosproject.rest.AbstractWebResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+
+/**
+ * Handles REST API call for kubevirt security group.
+ */
+@Path("security-group")
+public class KubevirtSecurityGroupWebResource extends AbstractWebResource {
+
+ protected final Logger log = LoggerFactory.getLogger(getClass());
+ private static final String SECURITY_GROUPS = "security-groups";
+
+ /**
+ * Returns set of all security groups.
+ *
+ * @return 200 OK with set of all security groups
+ * @onos.rsModel KubevirtSecurityGroups
+ */
+ @GET
+ @Produces(MediaType.APPLICATION_JSON)
+ public Response getSecurityGroups() {
+ KubevirtSecurityGroupService service = get(KubevirtSecurityGroupService.class);
+ final Iterable<KubevirtSecurityGroup> sgs = service.securityGroups();
+ return ok(encodeArray(KubevirtSecurityGroup.class, SECURITY_GROUPS, sgs)).build();
+ }
+}