Add Arion gateway (#748)

kubernetes/services/dpm_manager.yaml

@@ -61,6 +61,11 @@ data:
     host.ip.to.group.topic.map=group-topic1:192.168.131.131,10.10.10.11 group-topic2:192.168.131.131,11.11.11.12
     group.topic.to.multicast.topic.map=multicast-topic1:group-topic1,group-topic3 multicast-topic2:group-topic2,group-topic4
 
+
+    arionGateway.enabled = false
+    arionMaster.server = 127.0.0.1
+    arionMaster.port = 9090
+
     zetaGateway.enabled=false
     zetaGateway.node.mac=e0:97:96:02:45:53
     microservices.node.service.url=http://nodemanager-service.default.svc.cluster.local:9007/nodes

lib/pom.xml

@@ -81,22 +81,22 @@ Copyright(c) 2020 Futurewei Cloud
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-netty-shaded</artifactId>
-            <version>1.23.0</version>
+            <version>1.42.2</version>
         </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-protobuf</artifactId>
-            <version>1.23.0</version>
+            <version>1.42.2</version>
         </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-stub</artifactId>
-            <version>1.23.0</version>
+            <version>1.42.2</version>
         </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>protoc-gen-grpc-java</artifactId>
-            <version>1.23.0</version>
+            <version>1.42.2</version>
             <type>pom</type>
         </dependency>
 

schema/pom.xml

@@ -36,22 +36,22 @@
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-netty-shaded</artifactId>
-            <version>1.23.0</version>
+            <version>1.42.2</version>
         </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-protobuf</artifactId>
-            <version>1.23.0</version>
+            <version>1.42.2</version>
         </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-stub</artifactId>
-            <version>1.23.0</version>
+            <version>1.42.2</version>
         </dependency>
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>protoc-gen-grpc-java</artifactId>
-            <version>1.23.0</version>
+            <version>1.42.2</version>
             <type>pom</type>
         </dependency>
         <dependency>
@@ -62,7 +62,7 @@
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-testing</artifactId>
-            <version>1.23.0</version>
+            <version>1.42.2</version>
             <scope>test</scope>
         </dependency>
     </dependencies>

schema/proto3/gateway.proto

@@ -29,6 +29,7 @@ enum GatewayType {
     TGW = 2;    // Transit Gateway
     IGW = 3;    // Internet Gateway
     NGW = 4;    // NAT Gateway
+    ARION = 5;
 }
 
 message GatewayConfiguration {
@@ -50,8 +51,17 @@ message GatewayConfiguration {
         uint32 port_inband_operation = 1;
     }
 
+    message arion {
+            string vpc_id = 1;
+            uint32 vni = 2;
+            string subnet_id = 3;
+            // port for in-band (same NIC channel) operation
+            uint32 port_inband_operation = 4;
+    }
+
     oneof extra_info {
         zeta zeta_info = 6;
+        arion arion_info = 7;
     }
 }
 

schema/proto3/goalstateprovisioner.proto

@@ -22,6 +22,7 @@ option java_package = "com.futurewei.alcor.schema";
 
 import "common.proto";
 import "goalstate.proto";
+import "neighbor.proto";
 
 service GoalStateProvisioner {
 
@@ -49,6 +50,8 @@ service GoalStateProvisioner {
     rpc RequestGoalStates (HostRequest) returns (HostRequestReply) {
     }
 
+    rpc PushGoalstates (NeighborRulesRequest) returns (GoalStateOperationReply) {
+    }
 }
 
 message HostRequest {
@@ -83,6 +86,13 @@ message HostRequestReply {
     uint32 total_operation_time = 3;
 }
 
+message NeighborRulesRequest {
+    uint32 format_version = 1;
+    string request_id = 2;
+
+    repeated NeighborState neigborstates = 3;
+}
+
 message GoalStateOperationReply {
     uint32 format_version = 1;
 

schema/proto3/neighbor.proto

@@ -43,6 +43,9 @@ message NeighborConfiguration {
         NeighborType neighbor_type = 1;
         string subnet_id = 2;
         string ip_address = 3;
+        string arion_group = 4;
+        uint32 tunnel_id = 5;
+        string mac_address = 6;
     }
 
     message AllowAddressPair {

services/data_plane_manager/pom.xml

@@ -86,7 +86,11 @@ Copyright(c) 2020 Futurewei Cloud
             <artifactId>junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
-
+        <dependency>
+            <groupId>com.github.ishugaliy</groupId>
+            <artifactId>allgood-consistent-hash</artifactId>
+            <version>1.0.0</version>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>

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

@@ -0,0 +1,101 @@
+/*
+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.dataplane.entity.ArionGroup;
+import com.futurewei.alcor.dataplane.entity.ArionWing;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.stereotype.Repository;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+
+@Repository
+@ComponentScan(value="com.futurewei.alcor.common.db")
+public class ArionWingCache {
+
+    // arionWingCache store Arion wing meta data. key is Arion Wing hash code and value is Arion wing meta data.
+    private ICache<String, ArionWing> arionWingCache;
+
+    // arionWingGroupCache store Arion group meta data. key is Arion wing group name, value is Arion group meta data.
+    private ICache<String, ArionGroup> arionWingGroupCache;
+    private CacheFactory cacheFactory;
+
+    @Autowired
+    public ArionWingCache(CacheFactory cacheFactory) {
+        this.cacheFactory = cacheFactory;
+        arionWingCache = cacheFactory.getCache(ArionWing.class);
+        arionWingGroupCache = cacheFactory.getCache(ArionGroup.class);
+    }
+
+    @DurationStatistics
+    public ArionWing getArionWing (String resourceId) throws CacheException {
+        return arionWingCache.get(resourceId);
+    }
+
+    @DurationStatistics
+    public Collection<ArionWing> getArionWings () throws CacheException {
+        return arionWingCache.getAll().values();
+    }
+
+    @DurationStatistics
+    public Map<String, ArionWing> getAllSubnetPorts(Map<String, Object[]> queryParams) throws CacheException {
+        return arionWingCache.getAll(queryParams);
+    }
+
+    @DurationStatistics
+    public Collection<ArionWing> getArionWings (Set<String> keys) throws CacheException {
+        return arionWingCache.getAll(keys).values();
+    }
+
+    @DurationStatistics
+    public void insertArionWing (ArionWing arionWing) throws CacheException {
+        arionWingCache.put(String.valueOf(arionWing.hashCode()), arionWing);
+    }
+
+    @DurationStatistics
+    public void deleteArionWing (String resourceId) throws CacheException {
+        arionWingCache.remove(resourceId);
+    }
+
+    @DurationStatistics
+    public Object getArionGroup (String resourceId) throws CacheException {
+        return arionWingGroupCache.get(resourceId);
+    }
+
+    @DurationStatistics
+    public void insertArionGroup (String resourceId) throws CacheException {
+        System.out.println("Insert arion group: " + resourceId);
+        arionWingGroupCache.put(resourceId, new ArionGroup(resourceId));
+    }
+
+    @DurationStatistics
+    public void deleteArionGroup (String resourceId) throws CacheException {
+        arionWingGroupCache.remove(resourceId);
+    }
+
+    @DurationStatistics
+        public Map<String, ArionGroup> getAllArionGroup () throws CacheException {
+        return arionWingGroupCache.getAll();
+    }
+
+}

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

@@ -59,6 +59,15 @@ public class DataPlaneClientImplV2 implements DataPlaneClient<UnicastGoalStateV2
     // prints out UUID and time, when sending a GoalState to any of the monitorHosts
     private ArrayList<String> monitorHosts;
 
+    @Value("${arionGateway.enabled:false}")
+    private boolean arionGatwayEnabled;
+
+    @Value("${arionGateway.server:127.0.0.1}")
+    private String arionMasterServer;
+
+    @Value("${arionGateway.port:9090}")
+    private int arionMasterPort;
+
     @Value("${microservices.connectTimeout:300}")
     private String connectTimeout;
 
@@ -72,6 +81,9 @@ public List<String> sendGoalStates(List<UnicastGoalStateV2> unicastGoalStates) t
         for (UnicastGoalStateV2 unicastGoalState : unicastGoalStates) {
             goalStateBuilder = getGoalState(goalStateBuilder, unicastGoalState);
         }
+        if (arionGatwayEnabled) {
+            doSendGoalStateToArionMaster(goalStateBuilder);
+        }
         doSendGoalState(goalStateBuilder.build(), finishLatch, results);
 
         if (!finishLatch.await(Integer.parseInt(connectTimeout), TimeUnit.SECONDS)) {
@@ -149,8 +161,8 @@ public List<String> sendGoalStates(List<UnicastGoalStateV2> unicastGoalStates, M
         return null;
     }
 
-    private GrpcChannelStub createGrpcChannelStub(String hostIp) {
-        ManagedChannel channel = ManagedChannelBuilder.forAddress(hostIp, this.hostAgentPort)
+    private GrpcChannelStub createGrpcChannelStub(String hostIp, int port) {
+        ManagedChannel channel = ManagedChannelBuilder.forAddress(hostIp, port)
                 .usePlaintext()
                 .keepAliveWithoutCalls(true)
                 .keepAliveTime(Long.MAX_VALUE, TimeUnit.SECONDS)
@@ -160,30 +172,31 @@ private GrpcChannelStub createGrpcChannelStub(String hostIp) {
         return new GrpcChannelStub(channel, asyncStub);
     }
 
-    private GrpcChannelStub getOrCreateGrpcChannel(String hostIp) {
-        if (!this.hostIpGrpcChannelStubMap.containsKey(hostIp)) {
-            this.hostIpGrpcChannelStubMap.put(hostIp, createGrpcChannelStubArrayList(hostIp));
+    private GrpcChannelStub getOrCreateGrpcChannel(String hostIp, Integer port) {
+        if (!this.hostIpGrpcChannelStubMap.containsKey(hostIp + port)) {
+
+            this.hostIpGrpcChannelStubMap.put(hostIp + port, createGrpcChannelStubArrayList(hostIp, port));
             LOG.info("[getOrCreateGrpcChannel] Created a channel and stub to host IP: " + hostIp);
         }
         int usingChannelWithThisIndex = ThreadLocalRandom.current().nextInt(0, numberOfGrpcChannelPerHost);
-        ManagedChannel chan = this.hostIpGrpcChannelStubMap.get(hostIp).get(usingChannelWithThisIndex).channel;
+        ManagedChannel chan = this.hostIpGrpcChannelStubMap.get(hostIp + port).get(usingChannelWithThisIndex).channel;
         //checks the channel status, reconnects if the channel is IDLE
 
         ConnectivityState channelState = chan.getState(true);
         if (channelState != ConnectivityState.READY && channelState != ConnectivityState.CONNECTING && channelState != ConnectivityState.IDLE) {
-            GrpcChannelStub newChannelStub = createGrpcChannelStub(hostIp);
-            this.hostIpGrpcChannelStubMap.get(hostIp).set(usingChannelWithThisIndex, newChannelStub);
+            GrpcChannelStub newChannelStub = createGrpcChannelStub(hostIp, port);
+            this.hostIpGrpcChannelStubMap.get(hostIp + port).set(usingChannelWithThisIndex, newChannelStub);
             LOG.info("[getOrCreateGrpcChannel] Replaced a channel and stub to host IP: " + hostIp);
         }
         LOG.info("[getOrCreateGrpcChannel] Using channel and stub index " + usingChannelWithThisIndex + " to host IP: " + hostIp);
-        return this.hostIpGrpcChannelStubMap.get(hostIp).get(usingChannelWithThisIndex);
+        return this.hostIpGrpcChannelStubMap.get(hostIp + port).get(usingChannelWithThisIndex);
     }
 
-    private ArrayList<GrpcChannelStub> createGrpcChannelStubArrayList(String hostIp) {
+    private ArrayList<GrpcChannelStub> createGrpcChannelStubArrayList(String hostIp, int port) {
         long start = System.currentTimeMillis();
         ArrayList<GrpcChannelStub> arr = new ArrayList<>();
         for (int i = 0; i < numberOfGrpcChannelPerHost; i++) {
-            GrpcChannelStub channelStub = createGrpcChannelStub(hostIp);
+            GrpcChannelStub channelStub = createGrpcChannelStub(hostIp, port);
             // Using Linkerd load balance
             //warmUpChannelStub(channelStub, hostIp);
             arr.add(channelStub);
@@ -232,10 +245,34 @@ public void onCompleted() {
         return;
     }
 
+    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>() {
+            @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 doSendGoalState(Goalstate.GoalStateV2 goalStateV2, CountDownLatch finishLatch, List<String> replies) {
         String hostIp = netwconfigmanagerGrpcServiceUrl;
         long start = System.currentTimeMillis();
-        GrpcChannelStub channelStub = getOrCreateGrpcChannel(hostIp);
+        GrpcChannelStub channelStub = getOrCreateGrpcChannel(hostIp, hostAgentPort);
         long chan_established = System.currentTimeMillis();
         LOG.info("[doSendGoalState] Established channel, elapsed Time in milli seconds: " + (chan_established - start));
         GoalStateProvisionerGrpc.GoalStateProvisionerStub asyncStub = channelStub.stub;
@@ -371,10 +408,10 @@ private Goalstate.GoalStateV2.Builder getGoalState(Goalstate.GoalStateV2.Builder
 
 
                 if (goalStateBuilder.containsRouterStates(unicastGoalStateV2.getHostIp() + "/" + entry.getKey())) {
-                   Router.RouterConfiguration.Builder routerConfigurationBuilder = goalStateBuilder.getRouterStatesMap().get(unicastGoalStateV2.getHostIp() + "/" + entry.getKey()).getConfiguration().toBuilder();
-                   routerConfigurationBuilder.addAllSubnetRoutingTables(entry.getValue().getConfiguration().getSubnetRoutingTablesList());
-                   Router.RouterState.Builder routerStateBuilder = goalStateBuilder.getRouterStatesMap().get(unicastGoalStateV2.getHostIp() + "/" + entry.getKey()).toBuilder();
-                   routerStateBuilder.setConfiguration(routerConfigurationBuilder);
+                    Router.RouterConfiguration.Builder routerConfigurationBuilder = goalStateBuilder.getRouterStatesMap().get(unicastGoalStateV2.getHostIp() + "/" + entry.getKey()).getConfiguration().toBuilder();
+                    routerConfigurationBuilder.addAllSubnetRoutingTables(entry.getValue().getConfiguration().getSubnetRoutingTablesList());
+                    Router.RouterState.Builder routerStateBuilder = goalStateBuilder.getRouterStatesMap().get(unicastGoalStateV2.getHostIp() + "/" + entry.getKey()).toBuilder();
+                    routerStateBuilder.setConfiguration(routerConfigurationBuilder);
                 } else {
                     goalStateBuilder.putRouterStates(unicastGoalStateV2.getHostIp() + "/" + entry.getKey(), entry.getValue());
                 }