本次更新的代码未进行测试

bishe-cloud-center/src/main/java/top/ninwoo/cloudcenter/register/CloudRegisterCenter.java

@@ -10,6 +10,7 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
 import top.ninwoo.common.EdgeNodeEntity;
+import top.ninwoo.common.entity.NetworkTopology;
 import top.ninwoo.utils.util.impl.IpUtils;
 
 import javax.annotation.PostConstruct;

bishe-cloud-center/src/main/java/top/ninwoo/cloudcenter/service/impl/CloudServiceImpl.java

@@ -22,10 +22,18 @@ public class CloudServiceImpl implements CloudService {
     private static final Random randomInt = new Random(14);
     private static final String CREATE_CLUSTER = "/createCluster";
     private static final String DELETE_CLUSTER = "/delCluster";
+    private static final String DROP_DOCKER_NETWORK = "dropDockerNetwork";
+    private static final String CANCEL_DROP_DOCKER_NETWORK = "cancdelDropDockerNetwork";
+    // 全部的逻辑拓扑
+    // 全部的逻辑拓扑
+    private ConcurrentHashMap<Long, NetworkTopology> allLogicalTopo = new ConcurrentHashMap<>();
     // 用于存储配置文件
     private ConcurrentHashMap<Long, List<SeparatedClusterConfig>> allClusterConfig
             = new ConcurrentHashMap<>();
 
+    private ConcurrentHashMap<Long, Map<String, Set<String>>> allAppStatus
+            = new ConcurrentHashMap<>();
+
     @Resource
     private CloudRegisterCenter cloudRegisterCenter;
 
@@ -189,6 +197,11 @@ public class CloudServiceImpl implements CloudService {
      */
     @Override
     public List<SeparatedClusterConfig> sendClusterConfigToEdgeNode(List<SeparatedClusterConfig> clusterConfigs) {
+        if(clusterConfigs.size() == 0) {
+            LOG.warn("下发的集群配置为空。");
+            return clusterConfigs;
+        }
+        Set<String> allAppNames = new HashSet<>();
         // 这里应该做一个判断，edgeNode的ip是否可用，如果不可用，或者未设定，将随机挑选一个进行设置
         clusterConfigs.forEach(c -> {
             if (c.getEdgeNodeId() == null || "".equals(c.getEdgeNodeId())) {
@@ -199,15 +212,50 @@ public class CloudServiceImpl implements CloudService {
             ResponseEntity<Long> response = restTemplate.postForEntity("http://" + c.getEdgeNodeId() + CREATE_CLUSTER, c.getClusterConfig(), Long.class);
             if(!response.getStatusCode().is2xxSuccessful()) {
                 LOG.error("集群配置下发失败:{}:{}", c.getEdgeNodeId(), response.getBody());
+
+            }
+
+            // 构建初始化的逻辑拓扑
+            for (String appName : c.getClusterConfig().getTopology().getAppNames()) {
+                if(!appName.startsWith("br")) {
+                    allAppNames.add(appName);
+                    if (!allAppStatus.containsKey(c.getClusterConfig().getId())) {
+                        allAppStatus.put(c.getClusterConfig().getId(), new HashMap<>());
+                    }
+                    // 保存appnames
+                    Map<String, Set<String>> clusterAppConfig = allAppStatus.get(c.getClusterConfig().getId());
+                    if(!clusterAppConfig.containsKey(appName)) {
+                        clusterAppConfig.put(appName, new HashSet<>());
+                    }
+                    Set<String> edgeNodes = clusterAppConfig.get(appName);
+                    edgeNodes.add(c.getEdgeNodeId());
+                }
             }
             if(!allClusterConfig.containsKey(c.getClusterConfig().getId())) {
                 allClusterConfig.put(c.getClusterConfig().getId(), new ArrayList<>());
             }
             allClusterConfig.get(c.getClusterConfig().getId()).add(c);
         });
+
+        allLogicalTopo.put(clusterConfigs.get(0).getClusterConfig().getId(), initNetworkTopology(allAppNames));
+
         return clusterConfigs;
     }
 
+    private NetworkTopology initNetworkTopology(Set<String> allAppNames) {
+        NetworkTopology networkTopology = new NetworkTopology();
+        networkTopology.setTopologyId(1);
+        networkTopology.setAppNames(allAppNames.toArray(new String[allAppNames.size()]));
+        int[][] topo = new int[allAppNames.size()][allAppNames.size()];
+        for (int i = 0; i < allAppNames.size(); i++) {
+            for (int j = 0; j < allAppNames.size(); j++) {
+                topo[i][j] = 1;
+            }
+        }
+        networkTopology.setTopology(topo);
+        return networkTopology;
+    }
+
     /**
      * 删除全部的容器集群
      * @param clusterId
@@ -225,4 +273,66 @@ public class CloudServiceImpl implements CloudService {
         return true;
     }
 
+    /**
+     * 获取cluster的全网拓扑
+     * @param clusterId
+     * @return
+     */
+    public NetworkTopology getClusterTopology(long clusterId) {
+        return null;
+    }
+
+    /**
+     * 设置网络的逻辑拓扑
+     * @param clusterId
+     * @param topology
+     */
+    public void updateLogicalTopology(long clusterId, NetworkTopology topology) {
+        // 校验下topo是否合法
+        for (String appName : topology.getAppNames()) {
+            if(appName.contains("br")) {
+                throw new RuntimeException("暂时不支持非逻辑网络拓扑！");
+            }
+        }
+        // 更新topo配置
+        // 获取原来的网络topo
+        if (!allLogicalTopo.containsKey(clusterId)) {
+            throw new RuntimeException("不存在的topo，无法进行更新");
+        }
+
+        NetworkTopology networkTopology = allLogicalTopo.get(clusterId);
+        String[] appNames = networkTopology.getAppNames();
+        if(appNames.length != topology.getAppNames().length) {
+            throw new RuntimeException("更新的topo不一致");
+        }
+        for (int i = 0; i < appNames.length; i++) {
+            if (appNames[i] != topology.getAppNames()[i]) {
+                throw new RuntimeException("更新的名称不一致");
+            }
+        }
+
+        int[][] oldTopo = networkTopology.getTopology();
+        int[][] newTopo = topology.getTopology();
+
+        // 对比更新
+        for (int i = 0; i < appNames.length; i++) {
+            for (int j = 0; j < appNames.length; j++) {
+                if(oldTopo[i][j] == 1 && newTopo[i][j] == 0) {
+                    // 阻断连接
+                    // TODO step1 找到appNames对应的边缘服务器
+                    Set<String> edgeNodesIdSrc = allAppStatus.get(clusterId).get(appNames[i]);
+                    // TODO step2 更新网络
+                    Set<String> edgeNodesIdDst = allAppStatus.get(clusterId).get(appNames[j]);
+                    // 获取全部的Dst 容器网络
+                    // step2 调用远程接口，添加阻断连接
+                }
+                else if(oldTopo[i][j] == 0 && newTopo[i][j] == 1) {
+                    // 创建连接
+                    // step1 找到appNames对应的边缘服务器
+                    // step2 调用远程接口，删除阻断连接
+                }
+            }
+        }
+    }
+
 }

bishe-edge-center/src/main/java/top/ninwoo/edgecenter/controller/IndexController.java

@@ -2,11 +2,12 @@ package top.ninwoo.edgecenter.controller;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
+import top.ninwoo.common.entity.ClusterConfig;
 import top.ninwoo.common.entity.DockerContainer;
-import top.ninwoo.edgecenter.entity.ClusterConfig;
+import top.ninwoo.common.entity.NetworkTopology;
 import top.ninwoo.edgecenter.entity.ContainerInfo;
-import top.ninwoo.edgecenter.entity.NetworkTopology;
 import top.ninwoo.edgecenter.service.ClusterService;
+import top.ninwoo.edgecenter.service.IpService;
 import top.ninwoo.edgecenter.service.TopologyService;
 import top.ninwoo.utils.entity.NetworkInfo;
 import top.ninwoo.utils.entity.OvsBridge;
@@ -26,6 +27,9 @@ public class IndexController {
     @Autowired
     TopologyService topologyService;
 
+    @Autowired
+    IpService ipService;
+
     @RequestMapping(value = "/index")
     public List<DockerContainer> index(int flag) {
 
@@ -214,4 +218,27 @@ public class IndexController {
         String res = topologyService.modifyTopology(clusterId, topo);
         return res;
     }
+
+    @RequestMapping(value = "/dropDockerNetwork")
+    public String dropDockerNetwork(long clusterId, String appName, List<String> ipList) {
+        topologyService.dropDockerNetwork(clusterId, appName, ipList);
+        return "success";
+    }
+
+    @RequestMapping(value = "/cancelDropDockerNetwork")
+    public String cancelDropDockerNetwork(long clusterId, String appName, List<String> ipList) {
+        topologyService.cancelDropDockerNetwork(clusterId, appName, ipList);
+        return "success";
+    }
+
+    @RequestMapping(value = "/getAppIpsByAppName")
+    public Set<String> getAppIpsByAppName(long clusterId, String appName) {
+        Set<String> ids = clusterService.getContainerIdsByClusterId(clusterId, appName);
+        HashSet<String> ips = new HashSet<>();
+        ids.forEach(id -> {
+            ips.add(ipService.getContainerIp(id));
+        });
+
+        return ips;
+    }
 }
\ No newline at end of file

bishe-edge-center/src/main/java/top/ninwoo/edgecenter/entity/ClusterConfig.java

-package top.ninwoo.edgecenter.entity;
-
-import lombok.Data;
-
-import java.util.Date;
-import java.util.List;
-
-/**
- * @author joliu
- * @date 2019-10-20
- * @description 集群的配置类
- */
-@Data
-public class ClusterConfig {
-    private long id;
-    private Date createTime;
-    private String owner;
-    private List<ContainerDescription> dockers;
-    private NetworkTopology topology;
-}

bishe-edge-center/src/main/java/top/ninwoo/edgecenter/entity/ContainerDescription.java

-package top.ninwoo.edgecenter.entity;
-
-import lombok.Data;
-import top.ninwoo.common.entity.DockerContainer;
-
-/**
- * @Author joliu
- * @Description 容器具体的描述信息
- * @Date　Create in 下午2:17 2019/10/20
- */
-@Data
-public class ContainerDescription {
-    private String mode;
-    private int replicas;
-    // dockerContainer模板
-    private DockerContainer dockerContainer;
-
-}

bishe-edge-center/src/main/java/top/ninwoo/edgecenter/entity/NetworkTopology.java

-package top.ninwoo.edgecenter.entity;
-
-import lombok.Data;
-
-/**
- * @Author joliu
- * @Description 描述网络拓扑, 这个数据结构最终要存储到数据库中
- * @Date　Create in 下午9:30 2019/11/6
- */
-@Data
-public class NetworkTopology {
-    private long topologyId;
-    private String[] appNames;
-    private int[][] topology;
-
-/*    public NetworkTopology(long topologyId, String[] appNames) {
-        this.topologyId = topologyId;
-        this.appNames = appNames;
-        if(appNames == null || appNames.length < 1) {
-            throw new RuntimeException("容器的数组不能为空");
-        }
-        topology = new int[appNames.length][appNames.length];
-    }*/
-}

bishe-edge-center/src/main/java/top/ninwoo/edgecenter/service/ClusterService.java

 package top.ninwoo.edgecenter.service;
 
+import top.ninwoo.common.entity.ClusterConfig;
 import top.ninwoo.common.entity.DockerContainer;
-import top.ninwoo.edgecenter.entity.ClusterConfig;
 import top.ninwoo.utils.entity.NetworkInfo;
 import top.ninwoo.utils.entity.OvsBridge;
 

bishe-edge-center/src/main/java/top/ninwoo/edgecenter/service/TopologyService.java

 package top.ninwoo.edgecenter.service;
 
-import top.ninwoo.edgecenter.entity.ClusterConfig;
-import top.ninwoo.edgecenter.entity.NetworkTopology;
+import top.ninwoo.common.entity.ClusterConfig;
+import top.ninwoo.common.entity.NetworkTopology;
+
+import java.util.List;
 
 /**
  * @Author joliu
@@ -29,4 +31,8 @@ public interface TopologyService {
     String modifyTopology(long clusterId, NetworkTopology topology);
 
     void removeTopoByCluterId(long clusterId);
+
+    void dropDockerNetwork(long clusterId, String appName, List<String> ipList);
+
+    void cancelDropDockerNetwork(long clusterId, String appName, List<String> ipList);
 }

bishe-edge-center/src/main/java/top/ninwoo/edgecenter/service/impl/ClusterServiceImpl.java

@@ -4,10 +4,9 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import top.ninwoo.common.entity.ClusterConfig;
+import top.ninwoo.common.entity.ContainerDescription;
 import top.ninwoo.common.entity.DockerContainer;
-import top.ninwoo.edgecenter.entity.ClusterConfig;
-import top.ninwoo.edgecenter.entity.ContainerDescription;
-import top.ninwoo.edgecenter.entity.NetworkTopology;
 import top.ninwoo.edgecenter.service.ClusterService;
 import top.ninwoo.edgecenter.service.TopologyService;
 import top.ninwoo.utils.entity.NetworkInfo;

bishe-edge-center/src/main/java/top/ninwoo/edgecenter/service/impl/IpServiceImpl.java

@@ -22,6 +22,9 @@ public class IpServiceImpl implements IpService {
         // networkSegment 10.100.1.0/24
         // 在分配IP前还需要校验下这个网段是否是空闲的网段
         AtomicInteger atomicInteger;
+        if(ipMap.containsKey(containerId)) {
+            return ipMap.get(containerId);
+        }
         if(!countMap.containsKey(networkSegment)) {
             countMap.put(networkSegment, new AtomicInteger(2));
         }

bishe-edge-center/src/main/java/top/ninwoo/edgecenter/service/impl/TopologyServiceImpl.java

@@ -4,8 +4,8 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
-import top.ninwoo.edgecenter.entity.ClusterConfig;
-import top.ninwoo.edgecenter.entity.NetworkTopology;
+import top.ninwoo.common.entity.ClusterConfig;
+import top.ninwoo.common.entity.NetworkTopology;
 import top.ninwoo.edgecenter.service.ClusterService;
 import top.ninwoo.edgecenter.service.IpService;
 import top.ninwoo.edgecenter.service.TopologyService;
@@ -13,6 +13,7 @@ import top.ninwoo.utils.service.IptablesService;
 import top.ninwoo.utils.service.OVSService;
 import top.ninwoo.utils.service.OvsDockerService;
 
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -156,9 +157,24 @@ public class TopologyServiceImpl implements TopologyService {
                 ovsDockerService.addPort(ovsName, "eth1", cid, ip);
             });
         }
-        // TODO: 如果两个都是容器，这里可以配置逻辑连接
+        // 这里非常特殊，只有1的时候才中断容器和容器之间的网络
         else {
-            throw new RuntimeException("Physical topology does not support c2c link.");
+            // 这里先不配置逻辑拓扑
+            /*// throw new RuntimeException("Physical topology does not support c2c link.");
+            // step1: 获取容器id
+            Set<String> app1s = clusterService.getContainerIdsByClusterId(clusterId, appName1);
+            Set<String> app2s = clusterService.getContainerIdsByClusterId(clusterId, appName2);
+
+            // step:3 下发drop流表
+            app1s.forEach(c -> {
+                // 获取source ip
+                String sourceIp = ipService.assignIpString("10.10.1.0/24", c);
+                app2s.forEach( c1 -> {
+                    String destinationIp = ipService.assignIpString("10.10.1.0/24", c1);
+                    iptablesService.dropTraffic(c, sourceIp, destinationIp);
+                    iptablesService.dropTraffic(c1, destinationIp, sourceIp);
+                });
+            });*/
         }
     }
 
@@ -353,9 +369,27 @@ public class TopologyServiceImpl implements TopologyService {
                 ovsDockerService.delPort(ovsName, "eth1", cid);
             });
         }
-        // TODO: 如果两个都是容器，这里可以配置逻辑连接
         else {
-            throw new RuntimeException("Physical topology does not support c2c link.");
+            // 这里不配置逻辑拓扑
+            // step1 先获取两个容器的id iptablesUtils.delIptable(dockerContainer.getId(), ChainType.INPUT, "172.0.17.3", "", "DROP");
+            // step2 再获取两个容器的ip地址
+            // step3 设置a->b的流向
+            // step4 设置b->a的流向
+            // throw new RuntimeException("Physical topology does not support c2c link.");
+            // step1: 获取容器id
+            /*Set<String> app1s = clusterService.getContainerIdsByClusterId(clusterId, appName1);
+            Set<String> app2s = clusterService.getContainerIdsByClusterId(clusterId, appName2);
+
+            // step:3 下发drop流表
+            app1s.forEach(c -> {
+                // 获取source ip
+                String sourceIp = ipService.assignIpString("10.10.1.0/24", c);
+                app2s.forEach( c1 -> {
+                    String destinationIp = ipService.assignIpString("10.10.1.0/24", c1);
+                    iptablesService.cancelDropTraffic(c, sourceIp, destinationIp);
+                    iptablesService.cancelDropTraffic(c1, destinationIp, sourceIp);
+                });
+            });*/
         }
     }
 
@@ -384,4 +418,25 @@ public class TopologyServiceImpl implements TopologyService {
             }
         }
     }
+
+    @Override
+    public void dropDockerNetwork(long clusterId, String appName, List<String> ipList) {
+        // 找到全部的appName的容器id
+        Set<String> cids = clusterService.getContainerIdsByClusterId(clusterId, appName);
+        cids.forEach(cid -> {
+            ipList.forEach( dstIp -> {
+                iptablesService.dropTraffic(cid, ipService.getContainerIp(cid), dstIp);
+            });
+        });
+    }
+
+    @Override
+    public void cancelDropDockerNetwork(long clusterId, String appName, List<String> ipList) {
+        Set<String> cids = clusterService.getContainerIdsByClusterId(clusterId, appName);
+        cids.forEach(cid -> {
+            ipList.forEach( dstIp -> {
+                iptablesService.cancelDropTraffic(cid, ipService.getContainerIp(cid), dstIp);
+            });
+        });
+    }
 }

bishe-edge-center/src/test/java/top/ninwoo/edgecenter/ClusterServiceTest.java

@@ -8,8 +8,6 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
 import top.ninwoo.common.entity.DockerContainer;
-import top.ninwoo.edgecenter.entity.ClusterConfig;
-import top.ninwoo.edgecenter.entity.ContainerDescription;
 import top.ninwoo.edgecenter.service.ClusterService;
 
 import java.util.ArrayList;

bishe-utils/src/main/java/top/ninwoo/utils/service/IptablesService.java

@@ -30,4 +30,8 @@ public interface IptablesService {
     void deleteIptableById(String containerId, ChainType type, int lineNumber);
 
     boolean forwardTraffic(String containerId, String selfIp, int selfPort, String toIp, int toPort);
+
+    void dropTraffic(String containerId, String sourceIp, String destinationIp);
+
+    void cancelDropTraffic(String containerId, String souceIp, String destinationIp);
 }

bishe-utils/src/main/java/top/ninwoo/utils/service/impl/IptablesServiceImpl.java

@@ -336,4 +336,17 @@ public class IptablesServiceImpl implements IptablesService {
 
         return true;
     }
+
+    @Override
+    public void dropTraffic(String containerId, String sourceIp, String destinationIp) {
+        String s = iptablesUtils.addIptable(containerId, "append", ChainType.INPUT, sourceIp, destinationIp, "DROP");
+        if(!s.equals("")) {
+            throw new RuntimeException("container下发流表出错:" + containerId);
+        }
+    }
+
+    @Override
+    public void cancelDropTraffic(String containerId, String souceIp, String destinationIp) {
+        iptablesUtils.delIptable(containerId, ChainType.INPUT, souceIp, destinationIp, "DROP");
+    }
 }