[DPM] Refactor db/cache codes to support GoalStateV2 design (#713)

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

@@ -43,11 +43,15 @@ public class LocalCacheImpl implements LocalCache {
     @Autowired
     private SubnetPortsCache subnetPortsCache;
 
+    @Autowired
+    private PortHostInfoCache portHostInfoCache;
+
     @Autowired
     private NodeInfoCache nodeInfoCache;
 
     @Override
     public void setSubnetPorts(NetworkConfiguration networkConfig) throws Exception {
+
         List<InternalPortEntity> portEntities = networkConfig.getPortEntities();
         if (portEntities == null) {
             return;
@@ -70,6 +74,7 @@ public void setSubnetPorts(NetworkConfiguration networkConfig) throws Exception
                 portHostInfo.setPortId(portEntity.getId());
                 portHostInfo.setHostIp(portEntity.getBindingHostIP());
                 portHostInfo.setHostId(portEntity.getBindingHostId());
+                portHostInfo.setSubnetId(subnetId);
 
                 InternalSubnetPorts subnetPorts = subnetPortsMap.get(subnetId);
                 if (subnetPorts == null) {
@@ -85,7 +90,6 @@ public void setSubnetPorts(NetworkConfiguration networkConfig) throws Exception
                     subnetPorts.setVpcId(subnetEntity.getVpcId());
                     subnetPorts.setTunnelId(subnetEntity.getTunnelId());
                     subnetPorts.setDhcpEnable(subnetEntity.getDhcpEnable());
-                    subnetPorts.setPorts(new ArrayList<>());
 
                     List<InternalRouterInfo> routers = networkConfig.getInternalRouterInfos();
                     if (routers != null && routers.size() > 0) {
@@ -106,14 +110,13 @@ public void setSubnetPorts(NetworkConfiguration networkConfig) throws Exception
 
                     subnetPortsMap.put(subnetId, subnetPorts);
                 }
-
-                subnetPorts.getPorts().add(portHostInfo);
             }
         }
 
         for (Map.Entry<String, InternalSubnetPorts> entry: subnetPortsMap.entrySet()) {
             subnetPortsCache.updateSubnetPorts(entry.getValue());
         }
+
     }
 
     private InternalSubnetEntity getSubnetEntity(NetworkConfiguration networkConfig, String subnetId) throws Exception {

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

@@ -0,0 +1,87 @@
+/*
+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.web.entity.dataplane.v2.NetworkConfiguration;
+import com.futurewei.alcor.web.entity.port.PortHostInfo;
+import com.futurewei.alcor.web.entity.subnet.InternalSubnetPorts;
+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.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Repository
+@ComponentScan(value="com.futurewei.alcor.common.db")
+public class PortHostInfoCache {
+
+    private ICache<String, PortHostInfo> portHostInfoCache;
+    private CacheFactory cacheFactory;
+
+    @Autowired
+    public PortHostInfoCache(CacheFactory cacheFactory) {
+        this.cacheFactory = cacheFactory;
+        portHostInfoCache = cacheFactory.getCache(PortHostInfo.class);
+    }
+
+    public Map<String, PortHostInfo> getPortHostInfo(NetworkConfiguration networkConfig) {
+        Map<String, PortHostInfo> portHostInfoMap =  networkConfig
+                .getPortEntities()
+                .stream()
+                .filter(portEntity -> portEntity.getFixedIps().size() > 0)
+                .flatMap(portEntity -> portEntity.getFixedIps()
+                        .stream()
+                        .filter(fixedIp -> fixedIp != null)
+                        .map(fixedIp -> new PortHostInfo(portEntity.getId()
+                                , fixedIp.getIpAddress()
+                                , portEntity.getMacAddress()
+                                , portEntity.getBindingHostId()
+                                , portEntity.getBindingHostIP()
+                                , fixedIp.getSubnetId())))
+                .collect(Collectors.toMap(portHostInfo -> portHostInfo.getSubnetId() + cacheFactory.KEY_DELIMITER + portHostInfo.getPortIp(), Function.identity()));
+        return portHostInfoMap;
+    }
+
+    public void updatePortHostInfo(Map<String, PortHostInfo> portHostInfoMap) throws CacheException {
+        portHostInfoCache.putAll(portHostInfoMap);
+    }
+
+    public synchronized Collection<PortHostInfo> getPortHostInfos(String subnetId) throws CacheException {
+        Map<String, Object[]> queryParams = new HashMap<>();
+        Object[] values = new Object[1];
+        values[0] = subnetId;
+        queryParams.put("subnetId", values);
+        // Use sql index
+        return portHostInfoCache.getAll(queryParams).values();
+    }
+
+    public synchronized PortHostInfo getPortHostInfoByIp(String subnetId, String ip) throws CacheException {
+        return portHostInfoCache.get(subnetId + cacheFactory.KEY_DELIMITER + ip);
+    }
+
+    public synchronized void deletePortHostInfo(String subnetId, String portIp) throws CacheException {
+        portHostInfoCache.remove(subnetId + cacheFactory.KEY_DELIMITER + portIp);
+    }
+
+}

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

@@ -20,89 +20,62 @@ free of charge, to any person obtaining a copy of this software and associated d
 import com.futurewei.alcor.common.db.ICache;
 import com.futurewei.alcor.common.db.repo.ICacheRepository;
 import com.futurewei.alcor.common.stats.DurationStatistics;
+import com.futurewei.alcor.dataplane.entity.InternalSubnetRouterMap;
 import com.futurewei.alcor.dataplane.entity.InternalSubnets;
+import com.futurewei.alcor.web.entity.dataplane.v2.NetworkConfiguration;
+import com.futurewei.alcor.web.entity.port.PortHostInfo;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.stereotype.Repository;
 
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
 @Slf4j
 @Repository
 @ComponentScan(value="com.futurewei.alcor.common.db")
-public class RouterSubnetsCache implements ICacheRepository<InternalSubnets> {
+public class RouterSubnetsCache {
     // The cache is a map(routerId, subnetIds)
-    private final ICache<String, InternalSubnets> routerSubnetsCache;
+    private final ICache<String, InternalSubnetRouterMap> routerSubnetsCache;
+    private CacheFactory cacheFactory;
 
     @Autowired
     public RouterSubnetsCache(CacheFactory cacheFactory) {
-        this.routerSubnetsCache = cacheFactory.getCache(InternalSubnets.class);
+        this.cacheFactory = cacheFactory;
+        this.routerSubnetsCache = cacheFactory.getCache(InternalSubnetRouterMap.class);
     }
 
     @DurationStatistics
-    public InternalSubnets getRouterSubnets(String routerId) throws CacheException {
-        return routerSubnetsCache.get(routerId);
-    }
+    public Collection<InternalSubnetRouterMap> getRouterSubnets(String routerId) throws CacheException {
+        Map<String, Object[]> queryParams = new HashMap<>();
+        Object[] values = new Object[1];
+        values[0] = routerId;
+        queryParams.put("routerId", values);
+        return routerSubnetsCache.getAll(queryParams).values();
 
-    @DurationStatistics
-    public Map<String, InternalSubnets> getAllSubnets() throws CacheException {
-        return routerSubnetsCache.getAll();
-    }
-
-    @DurationStatistics
-    public Map<String, InternalSubnets> getAllSubnets(Map<String, Object[]> queryParams) throws CacheException {
-        return routerSubnetsCache.getAll(queryParams);
     }
 
     @DurationStatistics
-    public synchronized void addVpcSubnets(InternalSubnets subnets) throws CacheException {
-        routerSubnetsCache.put(subnets.getRouterId(), subnets);
+    public Collection<InternalSubnetRouterMap> updateVpcSubnets(NetworkConfiguration networkConfig) throws CacheException {
+        Map<String, InternalSubnetRouterMap> internalSubnetsMap =  networkConfig
+                .getInternalRouterInfos()
+                .stream()
+                .filter(routerInfo -> routerInfo.getRouterConfiguration().getSubnetRoutingTables().size() > 0)
+                .flatMap(routerInfo -> routerInfo.getRouterConfiguration().getSubnetRoutingTables()
+                        .stream()
+                        .map(routingTable -> new InternalSubnetRouterMap(routerInfo.getRouterConfiguration().getId()
+                                , routingTable.getSubnetId())))
+                .collect(Collectors.toMap(routerInfo -> routerInfo.getRouterId() + cacheFactory.KEY_DELIMITER + routerInfo.getSubnetId(), Function.identity()));
+        routerSubnetsCache.putAll(internalSubnetsMap);
+        return internalSubnetsMap.values();
     }
 
     @DurationStatistics
-    public void updateVpcSubnets(InternalSubnets subnets) throws CacheException {
-        routerSubnetsCache.put(subnets.getRouterId(), subnets);
-    }
-
-    @DurationStatistics
-    public void deleteVpcGatewayInfo(String routerId) throws CacheException {
-        routerSubnetsCache.remove(routerId);
-    }
-
-    @Override
-    public InternalSubnets findItem(String id) throws CacheException {
-        return routerSubnetsCache.get(id);
+    public void deleteVpcGatewayInfo(String routerId, String subnetId) throws CacheException {
+        routerSubnetsCache.remove(routerId + cacheFactory.KEY_DELIMITER + subnetId);
     }
 
-    @Override
-    public Map<String, InternalSubnets> findAllItems() throws CacheException {
-        return routerSubnetsCache.getAll();
-    }
 
-    @Override
-    public Map<String, InternalSubnets> findAllItems(Map<String, Object[]> queryParams) throws CacheException {
-        return routerSubnetsCache.getAll(queryParams);
-    }
-
-    @Override
-    public void addItem(InternalSubnets subnets) throws CacheException {
-        log.debug("Add Subnets {} to Router {}", subnets.toString(), subnets.getRouterId());
-        routerSubnetsCache.put(subnets.getRouterId(), subnets);
-    }
-
-    @Override
-    public void addItems(List<InternalSubnets> items) throws CacheException {
-        Map<String, InternalSubnets> subnetsMap = items.stream().collect(Collectors.toMap(InternalSubnets::getRouterId, Function.identity()));
-        routerSubnetsCache.putAll(subnetsMap);
-    }
-
-    @Override
-    public void deleteItem(String id) throws CacheException {
-        log.debug("Delete Router {}", id);
-        routerSubnetsCache.remove(id);
-    }
 }

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

@@ -18,22 +18,34 @@ free of charge, to any person obtaining a copy of this software and associated d
 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.db.Transaction;
 import com.futurewei.alcor.common.stats.DurationStatistics;
+import com.futurewei.alcor.dataplane.entity.InternalSubnetRouterMap;
+import com.futurewei.alcor.dataplane.entity.InternalSubnets;
+import com.futurewei.alcor.schema.Subnet;
+import com.futurewei.alcor.web.entity.dataplane.InternalSubnetEntity;
+import com.futurewei.alcor.web.entity.dataplane.v2.NetworkConfiguration;
+import com.futurewei.alcor.web.entity.port.PortHostInfo;
 import com.futurewei.alcor.web.entity.subnet.InternalSubnetPorts;
+import com.futurewei.alcor.web.entity.subnet.SubnetEntity;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.stereotype.Repository;
 
-import java.util.Map;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 @Repository
 @ComponentScan(value="com.futurewei.alcor.common.db")
 public class SubnetPortsCache {
     // The cache is a map(subnetId, subnetPorts)
     private ICache<String, InternalSubnetPorts> subnetPortsCache;
+    private CacheFactory cacheFactory;
 
     @Autowired
     public SubnetPortsCache(CacheFactory cacheFactory) {
+        this.cacheFactory = cacheFactory;
         subnetPortsCache = cacheFactory.getCache(InternalSubnetPorts.class);
     }
 
@@ -42,6 +54,16 @@ public InternalSubnetPorts getSubnetPorts(String subnetId) throws CacheException
         return subnetPortsCache.get(subnetId);
     }
 
+    @DurationStatistics
+    public Map<String, InternalSubnetPorts> getSubnetPortsByRouterId(String routerId) throws CacheException {
+        Map<String, Object[]> queryParams = new HashMap<>();
+        Object[] values = new Object[1];
+        values[0] = routerId;
+        queryParams.put("routerId", values);
+        // Use sql index
+        return subnetPortsCache.getAll(queryParams);
+    }
+
     @DurationStatistics
     public Map<String, InternalSubnetPorts> getAllSubnetPorts() throws CacheException {
         return subnetPortsCache.getAll();
@@ -57,6 +79,74 @@ public synchronized void addSubnetPorts(InternalSubnetPorts internalSubnetPorts)
         subnetPortsCache.put(internalSubnetPorts.getSubnetId(), internalSubnetPorts);
     }
 
+    @DurationStatistics
+    public Map<String, String> getInternalSubnetRouterMap(NetworkConfiguration networkConfig) {
+        if (networkConfig.getInternalRouterInfos() != null) {
+            Map<String, String> internalSubnetsRouterMap = networkConfig
+                    .getInternalRouterInfos()
+                    .stream()
+                    .filter(routerInfo -> routerInfo.getRouterConfiguration().getSubnetRoutingTables().size() > 0)
+                    .flatMap(routerInfo -> routerInfo.getRouterConfiguration().getSubnetRoutingTables()
+                            .stream()
+                            .map(routingTable -> new InternalSubnetRouterMap(routerInfo.getRouterConfiguration().getId()
+                                    , routingTable.getSubnetId())))
+                    .distinct()
+                    .collect(Collectors.toMap(routerInfo -> routerInfo.getSubnetId(), routerInfo -> routerInfo.getRouterId()));
+            return internalSubnetsRouterMap;
+        }
+        return new HashMap<>();
+    }
+
+    @DurationStatistics
+    public Map<String, InternalSubnetPorts> attacheRouter(Map<String, String> subnetIdRouterIdMap) throws CacheException {
+        Map<String, InternalSubnetPorts> internalSubnetEntityMap = new TreeMap<>();
+        for (Map.Entry<String, String> subnetIdRouterId : subnetIdRouterIdMap.entrySet()) {
+            InternalSubnetPorts internalSubnetPorts = null;
+            try {
+                if (subnetPortsCache.containsKey(subnetIdRouterId.getKey())) {
+                    internalSubnetPorts = subnetPortsCache.get(subnetIdRouterId.getKey());
+                    internalSubnetPorts.setRouterId(subnetIdRouterId.getValue());
+                    internalSubnetEntityMap.put(subnetIdRouterId.getKey(), internalSubnetPorts);
+                }
+            } catch (CacheException e) {
+                e.printStackTrace();
+            }
+        }
+
+        subnetPortsCache.putAll(internalSubnetEntityMap);
+        return internalSubnetEntityMap;
+    }
+
+
+    @DurationStatistics
+    public Map<String, InternalSubnetPorts> getSubnetPorts(NetworkConfiguration networkConfig) throws CacheException {
+        Map<String, String> internalSubnetsRouterMap = getInternalSubnetRouterMap(networkConfig);
+        Map<String, InternalSubnetPorts> internalSubnetPortsMap = new TreeMap<>();
+        for (InternalSubnetEntity subnetEntity : networkConfig.getSubnets()) {
+            internalSubnetPortsMap.put(subnetEntity.getId(), getInternalSubnetPorts(subnetEntity, internalSubnetsRouterMap.getOrDefault(subnetEntity.getId(), null)));
+        }
+        return internalSubnetPortsMap;
+    }
+
+    private InternalSubnetPorts getInternalSubnetPorts(InternalSubnetEntity subnetEntity, String routerId) {
+        InternalSubnetPorts internalSubnetPorts = new InternalSubnetPorts(subnetEntity.getId()
+                ,subnetEntity.getGatewayPortDetail().getGatewayPortId()
+                ,subnetEntity.getGatewayIp()
+                ,subnetEntity.getGatewayPortDetail().getGatewayMacAddress()
+                ,subnetEntity.getName()
+                ,subnetEntity.getCidr()
+                ,subnetEntity.getVpcId()
+                ,subnetEntity.getTunnelId()
+                ,subnetEntity.getDhcpEnable()
+                ,routerId);
+        return internalSubnetPorts;
+    }
+
+    @DurationStatistics
+    public void updateSubnetPorts(Map<String, InternalSubnetPorts> internalSubnetPortsMap) throws Exception {
+        subnetPortsCache.putAll(internalSubnetPortsMap);
+    }
+
     @DurationStatistics
     public void updateSubnetPorts(InternalSubnetPorts internalSubnetPorts) throws Exception {
         subnetPortsCache.put(internalSubnetPorts.getSubnetId(), internalSubnetPorts);
@@ -67,4 +157,8 @@ public void deleteSubnetPorts(String subnetId) throws Exception {
         subnetPortsCache.remove(subnetId);
     }
 
+    public Transaction getTransaction() {
+        return subnetPortsCache.getTransaction();
+    }
+
 }