Add Alcor security group for Arion gateway

schema/proto3/goalstateprovisioner.proto

@@ -23,6 +23,8 @@ option java_package = "com.futurewei.alcor.schema";
 import "common.proto";
 import "goalstate.proto";
 import "neighbor.proto";
+import "securitygroup.proto";
+import "port.proto";
 
 service GoalStateProvisioner {
 
@@ -50,7 +52,10 @@ service GoalStateProvisioner {
     rpc RequestGoalStates (HostRequest) returns (HostRequestReply) {
     }
 
-    rpc PushGoalstates (NeighborRulesRequest) returns (GoalStateOperationReply) {
+    rpc PushGoalstates (ArionGoalStateRequest) returns (GoalStateOperationReply) {
+    }
+
+    rpc PushSecurityGroupGoalState (SecurityGroupState) returns (GoalStateOperationReply) {
     }
 }
 
@@ -86,11 +91,12 @@ message HostRequestReply {
     uint32 total_operation_time = 3;
 }
 
-message NeighborRulesRequest {
+message ArionGoalStateRequest {
     uint32 format_version = 1;
     string request_id = 2;
 
     repeated NeighborState neigborstates = 3;
+    repeated PortState portstates = 4;
 }
 
 message GoalStateOperationReply {

services/data_plane_manager/src/main/java/com/futurewei/alcor/dataplane/cache/PortBindingSecurityGroupRepository.java

@@ -0,0 +1,82 @@
+/*
+MIT License
+Copyright(c) 2020 Futurewei Cloud
+
+    Permission is hereby granted,
+    free of charge, to any person obtaining a copy of this software and associated documentation files(the "Software"), to deal in the Software without restriction,
+    including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and / or sell copies of the Software, and to permit persons
+    to whom the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+    WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+package com.futurewei.alcor.dataplane.cache;
+
+import com.futurewei.alcor.common.db.CacheException;
+import com.futurewei.alcor.common.db.CacheFactory;
+import com.futurewei.alcor.common.db.ICache;
+import com.futurewei.alcor.common.stats.DurationStatistics;
+import com.futurewei.alcor.web.entity.securitygroup.PortBindingSecurityGroup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.stereotype.Repository;
+import javax.annotation.PostConstruct;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Repository
+@ComponentScan(value="com.futurewei.alcor.common.db")
+public class PortBindingSecurityGroupRepository {
+    private static final Logger LOG = LoggerFactory.getLogger(PortBindingSecurityGroupRepository.class);
+
+    private ICache<String, PortBindingSecurityGroup> bindingCache;
+
+    @Autowired
+    public PortBindingSecurityGroupRepository(CacheFactory cacheFactory) {
+        bindingCache = cacheFactory.getCache(PortBindingSecurityGroup.class);
+    }
+
+    @PostConstruct
+    private void init() {
+        LOG.info("PortBindingSecurityGroupRepository init done");
+    }
+
+    @DurationStatistics
+    public Collection<PortBindingSecurityGroup> getPortBindingSecurityGroupBySecurityGroupId(String securityGroupId) throws CacheException {
+        Map<String, Object[]> filterParams = new HashMap<>();
+        filterParams.put("securityGroupId", new String[] {securityGroupId});
+
+        Map<String, PortBindingSecurityGroup> portBindingSecurityGroupMap = bindingCache.getAll(filterParams);
+        return portBindingSecurityGroupMap.values();
+    }
+
+    @DurationStatistics
+    public Collection<PortBindingSecurityGroup> getPortBindingSecurityGroupByPortId(String portId) throws CacheException {
+        Map<String, Object[]> filterParams = new HashMap<>();
+        filterParams.put("portId", new String[] {portId});
+
+        Map<String, PortBindingSecurityGroup> portBindingSecurityGroupMap = bindingCache.getAll(filterParams);
+        return portBindingSecurityGroupMap.values();
+    }
+
+    @DurationStatistics
+    public synchronized void addPortBindingSecurityGroup(List<PortBindingSecurityGroup> portBindingSecurityGroups) throws Exception {
+        Map<String, PortBindingSecurityGroup> portBindingSecurityGroupMap = portBindingSecurityGroups
+                .stream()
+                .collect(Collectors.toMap(PortBindingSecurityGroup::getId, Function.identity()));
+        bindingCache.putAll(portBindingSecurityGroupMap);
+    }
+
+    @DurationStatistics
+    public synchronized void deleteSecurityGroupBinding(List<PortBindingSecurityGroup> portBindingSecurityGroups) throws Exception {
+        for (PortBindingSecurityGroup portBindingSecurityGroup: portBindingSecurityGroups) {
+            bindingCache.remove(portBindingSecurityGroup.getId());
+        }
+    }
+}

services/data_plane_manager/src/main/java/com/futurewei/alcor/dataplane/cache/SecurityGroupRepository.java

@@ -0,0 +1,94 @@
+/*
+MIT License
+Copyright(c) 2020 Futurewei Cloud
+
+    Permission is hereby granted,
+    free of charge, to any person obtaining a copy of this software and associated documentation files(the "Software"), to deal in the Software without restriction,
+    including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and / or sell copies of the Software, and to permit persons
+    to whom the Software is furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+    WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+package com.futurewei.alcor.dataplane.cache;
+
+import com.futurewei.alcor.common.db.CacheException;
+import com.futurewei.alcor.common.db.CacheFactory;
+import com.futurewei.alcor.common.db.ICache;
+import com.futurewei.alcor.common.stats.DurationStatistics;
+import com.futurewei.alcor.web.entity.securitygroup.SecurityGroupRule;
+import com.futurewei.alcor.web.entity.securitygroup.SecurityGroupRuleBulkJson;
+import com.futurewei.alcor.web.entity.securitygroup.SecurityGroupRuleJson;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Repository;
+import javax.annotation.PostConstruct;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Repository
+public class SecurityGroupRepository {
+    private static final Logger LOG = LoggerFactory.getLogger(SecurityGroupRepository.class);
+
+    private ICache<String, SecurityGroupRule> securityGroupRuleCache;
+
+    @Autowired
+    public SecurityGroupRepository(CacheFactory cacheFactory) {
+        securityGroupRuleCache = cacheFactory.getCache(SecurityGroupRule.class);
+    }
+
+    @PostConstruct
+    private void init() {
+        LOG.info("SecurityGroupRepository init done");
+    }
+
+    @DurationStatistics
+    public synchronized void addSecurityGroupRule(SecurityGroupRuleJson securityGroupRuleJson) throws Exception {
+        securityGroupRuleCache.put(securityGroupRuleJson.getSecurityGroupRule().getId(), securityGroupRuleJson.getSecurityGroupRule());
+    }
+    @DurationStatistics
+    public synchronized void addSecurityGroupRules(SecurityGroupRuleBulkJson securityGroupRuleBulkJson) throws Exception {
+        var securityGroupRules = securityGroupRuleBulkJson.getSecurityGroupRules().stream().collect(Collectors.toMap(SecurityGroupRule::getId, Function.identity()));
+        securityGroupRuleCache.putAll(securityGroupRules);
+    }
+
+    @DurationStatistics
+    public synchronized void deleteSecurityGroupRules(List<String> securityGroupIds) throws Exception {
+        securityGroupIds.forEach(securityGroupId -> {
+            try {
+                securityGroupRuleCache.remove(securityGroupId);
+            } catch (CacheException e) {
+                e.printStackTrace();
+            }
+        });
+    }
+
+    @DurationStatistics
+    public SecurityGroupRule getSecurityGroupRule(String securityGroupRuleId) throws CacheException {
+        return securityGroupRuleCache.get(securityGroupRuleId);
+    }
+
+    @DurationStatistics
+    public List<SecurityGroupRule> getSecurityGroupRules(List<String> securityGroupRuleIds) throws CacheException {
+        List<SecurityGroupRule> securityGroupRules = new ArrayList<>();
+        for (String securityGroupId : securityGroupRuleIds) {
+            SecurityGroupRule securityGroupRule = securityGroupRuleCache.get(securityGroupId);
+            securityGroupRules.add(securityGroupRule);
+        }
+        return securityGroupRules;
+    }
+
+    @DurationStatistics
+    public Collection<SecurityGroupRule> getSecurityGroupRules(String scurityGroupId) throws CacheException {
+        Map<String, Object[]> queryParams =  new HashMap<>();
+        Object[] value = new Object[1];
+        value[0] = scurityGroupId;
+        queryParams.put("securityGroupId", value);
+        return securityGroupRuleCache.getAll(queryParams).values();
+    }
+}

services/data_plane_manager/src/main/java/com/futurewei/alcor/dataplane/client/DataPlaneClient.java

@@ -21,6 +21,7 @@ free of charge, to any person obtaining a copy of this software and associated d
 import com.futurewei.alcor.dataplane.entity.UnicastGoalStateV2;
 import com.futurewei.alcor.schema.Goalstate.GoalState;
 import com.futurewei.alcor.schema.Goalstateprovisioner.GoalStateOperationReply.GoalStateOperationStatus;
+import com.futurewei.alcor.schema.SecurityGroup;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
@@ -31,4 +32,5 @@ public interface DataPlaneClient <U, T> {
     List<String> sendGoalStates(List<U> unicastGoalStates) throws Exception;
     List<String> sendGoalStates(List<U> unicastGoalStates,
                                 T multicastGoalState) throws Exception;
+    List<String> sendGoalStates(SecurityGroup.SecurityGroupState securityGroupState) throws Exception;
 }

services/data_plane_manager/src/main/java/com/futurewei/alcor/dataplane/client/grpc/DataPlaneClientImpl.java

@@ -23,6 +23,7 @@ free of charge, to any person obtaining a copy of this software and associated d
 import com.futurewei.alcor.schema.Goalstate;
 import com.futurewei.alcor.schema.Goalstateprovisioner.GoalStateOperationReply.GoalStateOperationStatus;
 import com.futurewei.alcor.schema.Goalstateprovisioner.GoalStateOperationReply;
+import com.futurewei.alcor.schema.SecurityGroup;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.stub.StreamObserver;
@@ -89,6 +90,11 @@ public List<String> sendGoalStates(
         return null;
     }
 
+    @Override
+    public List<String> sendGoalStates(SecurityGroup.SecurityGroupState securityGroupState) throws Exception {
+        return null;
+    }
+
     private List<String> doSendGoalStates(List<UnicastGoalState> unicastGoalStates) {
         List<Future<String>>
                 futures = new ArrayList<>(unicastGoalStates.size());

services/data_plane_manager/src/main/java/com/futurewei/alcor/dataplane/client/grpc/DataPlaneClientImplV2.java

@@ -164,6 +164,12 @@ public List<String> sendGoalStates(List<UnicastGoalStateV2> unicastGoalStates, M
         return null;
     }
 
+    @Override
+    public List<String> sendGoalStates(SecurityGroup.SecurityGroupState securityGroupState) throws Exception {
+        doSendGoalStateToArionMaster(securityGroupState);
+        return new ArrayList<>();
+    }
+
     private GrpcChannelStub createGrpcChannelStub(String hostIp, int port) {
         ManagedChannel channel = ManagedChannelBuilder.forAddress(hostIp, port)
                 .usePlaintext()
@@ -251,9 +257,33 @@ public void onCompleted() {
     private String doSendGoalStateToArionMaster (Goalstate.GoalStateV2.Builder goalStateV2) {
         GrpcChannelStub channelStub = getOrCreateGrpcChannel(arionMasterServer, arionMasterPort);
         GoalStateProvisionerGrpc.GoalStateProvisionerStub asyncStub = channelStub.stub;
-        var neighborStateRequestBuilder = Goalstateprovisioner.NeighborRulesRequest.newBuilder();
-        neighborStateRequestBuilder.addAllNeigborstates(goalStateV2.getNeighborStatesMap().values());
-        asyncStub.pushGoalstates(neighborStateRequestBuilder.build(), new StreamObserver<Goalstateprovisioner.GoalStateOperationReply>() {
+        var arionGoalStateBuilder = Goalstateprovisioner.ArionGoalStateRequest.newBuilder();
+        arionGoalStateBuilder.addAllNeigborstates(goalStateV2.getNeighborStatesMap().values());
+        arionGoalStateBuilder.addAllPortstates(goalStateV2.getPortStatesMap().values());
+
+        asyncStub.pushGoalstates(arionGoalStateBuilder.build(), new StreamObserver<Goalstateprovisioner.GoalStateOperationReply>() {
+            @Override
+            public void onNext(Goalstateprovisioner.GoalStateOperationReply goalStateOperationReply) {
+                LOG.info("Get response: " + goalStateOperationReply.toString());
+            }
+
+            @Override
+            public void onError(Throwable throwable) {
+                LOG.info(throwable.getMessage());
+            }
+
+            @Override
+            public void onCompleted() {
+
+            }
+        });
+        return null;
+    }
+
+    private String doSendGoalStateToArionMaster (SecurityGroup.SecurityGroupState securityGroupState) {
+        GrpcChannelStub channelStub = getOrCreateGrpcChannel(arionMasterServer, arionMasterPort);
+        GoalStateProvisionerGrpc.GoalStateProvisionerStub asyncStub = channelStub.stub;
+        asyncStub.pushSecurityGroupGoalState(securityGroupState, new StreamObserver<Goalstateprovisioner.GoalStateOperationReply>() {
             @Override
             public void onNext(Goalstateprovisioner.GoalStateOperationReply goalStateOperationReply) {
                 LOG.info("Get response: " + goalStateOperationReply.toString());

services/data_plane_manager/src/main/java/com/futurewei/alcor/dataplane/client/kafka/DataPlaneClientImpl.java

@@ -22,6 +22,7 @@ free of charge, to any person obtaining a copy of this software and associated d
 import com.futurewei.alcor.dataplane.entity.UnicastGoalStateV2;
 import com.futurewei.alcor.schema.Goalstate;
 import com.futurewei.alcor.schema.Goalstateprovisioner;
+import com.futurewei.alcor.schema.SecurityGroup;
 
 import java.util.List;
 import java.util.Map;
@@ -37,4 +38,9 @@ public List<String> sendGoalStates(List<UnicastGoalState> unicastGoalStates) thr
     public List<String> sendGoalStates(List<UnicastGoalState> unicastGoalStates, MulticastGoalState multicastGoalState) throws Exception {
         return null;
     }
+
+    @Override
+    public List<String> sendGoalStates(SecurityGroup.SecurityGroupState securityGroupState) throws Exception {
+        return null;
+    }
 }

services/data_plane_manager/src/main/java/com/futurewei/alcor/dataplane/client/kafka/DataPlaneClientImplV2.java

@@ -21,6 +21,7 @@
 import com.futurewei.alcor.dataplane.client.DataPlaneClient;
 import com.futurewei.alcor.dataplane.entity.MulticastGoalStateV2;
 import com.futurewei.alcor.dataplane.entity.UnicastGoalStateV2;
+import com.futurewei.alcor.schema.SecurityGroup;
 
 import java.util.List;
 
@@ -34,4 +35,9 @@ public List<String> sendGoalStates(List<UnicastGoalStateV2> unicastGoalStates) t
     public List<String> sendGoalStates(List<UnicastGoalStateV2> unicastGoalStates, MulticastGoalStateV2 multicastGoalState) throws Exception {
         return null;
     }
+
+    @Override
+    public List<String> sendGoalStates(SecurityGroup.SecurityGroupState securityGroupState) throws Exception {
+        return null;
+    }
 }

services/data_plane_manager/src/main/java/com/futurewei/alcor/dataplane/client/pulsar/group_node_mode/DataPlaneClientImpl.java

@@ -21,6 +21,7 @@ free of charge, to any person obtaining a copy of this software and associated d
 import com.futurewei.alcor.dataplane.entity.UnicastGoalState;
 import com.futurewei.alcor.dataplane.exception.GroupTopicNotFound;
 import com.futurewei.alcor.dataplane.exception.MulticastTopicNotFound;
+import com.futurewei.alcor.schema.SecurityGroup;
 import com.futurewei.alcor.web.entity.dataplane.MulticastGoalStateByte;
 import com.futurewei.alcor.web.entity.dataplane.UnicastGoalStateByte;
 import org.apache.pulsar.client.api.Producer;
@@ -152,4 +153,9 @@ public List<String> sendGoalStates(List<UnicastGoalState> unicastGoalStates, Mul
 
         return failedHosts;
     }
+
+    @Override
+    public List<String> sendGoalStates(SecurityGroup.SecurityGroupState securityGroupState) throws Exception {
+        return null;
+    }
 }

services/data_plane_manager/src/main/java/com/futurewei/alcor/dataplane/client/pulsar/group_node_mode/DataPlaneClientImplV2.java

@@ -20,6 +20,7 @@ free of charge, to any person obtaining a copy of this software and associated d
 import com.futurewei.alcor.dataplane.entity.*;
 import com.futurewei.alcor.dataplane.exception.GroupTopicNotFound;
 import com.futurewei.alcor.dataplane.exception.MulticastTopicNotFound;
+import com.futurewei.alcor.schema.SecurityGroup;
 import com.futurewei.alcor.web.entity.dataplane.MulticastGoalStateByte;
 import com.futurewei.alcor.web.entity.dataplane.UnicastGoalStateByte;
 import org.apache.pulsar.client.api.Producer;
@@ -159,4 +160,9 @@ public List<String> sendGoalStates(List<UnicastGoalStateV2> unicastGoalStates, M
 
         return failedHosts;
     }
+
+    @Override
+    public List<String> sendGoalStates(SecurityGroup.SecurityGroupState securityGroupState) throws Exception {
+        return null;
+    }
 }

services/data_plane_manager/src/main/java/com/futurewei/alcor/dataplane/client/pulsar/vpc_mode/DataPlaneClientImpl.java

@@ -19,6 +19,7 @@ free of charge, to any person obtaining a copy of this software and associated d
 import com.futurewei.alcor.dataplane.client.DataPlaneClient;
 import com.futurewei.alcor.dataplane.entity.MulticastGoalState;
 import com.futurewei.alcor.dataplane.entity.UnicastGoalState;
+import com.futurewei.alcor.schema.SecurityGroup;
 import com.futurewei.alcor.web.entity.dataplane.MulticastGoalStateByte;
 import com.futurewei.alcor.web.entity.dataplane.UnicastGoalStateByte;
 import com.futurewei.alcor.web.entity.topic.VpcTopicInfo;
@@ -129,4 +130,9 @@ public List<String> sendGoalStates(List<UnicastGoalState> unicastGoalStates, Mul
 
         return failedHosts;
     }
+
+    @Override
+    public List<String> sendGoalStates(SecurityGroup.SecurityGroupState securityGroupState) throws Exception {
+        return null;
+    }
 }