修复一个监控关闭失效的bug

bishe-client-starter/src/main/java/top/ninwoo/bishe/starter/service/NetworkService.java

@@ -20,5 +20,7 @@ public interface NetworkService {
 
     String enableNetworkMonitor(Long clusterId, String appName);
 
+    String cancelNetworkMonitor(Long clusterId, String appName);
+
     List<ContainerMonitor> getContainerMonitors(Long clusterId, String appName);
 }

bishe-client-starter/src/main/java/top/ninwoo/bishe/starter/service/NetworkServiceImpl.java

 package top.ninwoo.bishe.starter.service;
 
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.http.HttpMethod;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.RestTemplate;
@@ -10,6 +12,7 @@ import top.ninwoo.common.entity.ContainerMonitorInfo;
 import top.ninwoo.common.entity.NetworkTopology;
 
 import javax.annotation.Resource;
+import java.awt.*;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -24,9 +27,10 @@ public class NetworkServiceImpl implements NetworkService {
 
     private final static String GET_IP_LIST_BY_APP_NAME = "/network/getIpListByAppName?clusterId={clusterId}&appName={appName}";
 
-    private final static String ENABLE_NETWORK_MONITOR = "/enableNetworkMonitor?clusterId={clusterId}&appName={appName}";
+    private final static String ENABLE_NETWORK_MONITOR = "/network/enableNetworkMonitor?clusterId={clusterId}&appName={appName}";
 
-    private final static String GET_NETWORK_MONITOR_INFO = "/getNetworkMonitorInfo?clusterId={clusterId}&appName={appName}";
+    private final static String CANCEL_NETWORK_MONITOR = "/network/cancelNetworkMonitor?clusterId={clusterId}&appName={appName}";
+    private final static String GET_NETWORK_MONITOR_INFO = "/network/getNetworkMonitorInfo?clusterId={clusterId}&appName={appName}";
 
     @Resource
     ClientProperties clientProperties;
@@ -95,6 +99,18 @@ public class NetworkServiceImpl implements NetworkService {
         return restTemplate.getForObject("http://" + cloudUrl + ENABLE_NETWORK_MONITOR, String.class, param);
     }
 
+    @Override
+    public String cancelNetworkMonitor(Long clusterId, String appName) {
+        String cloudUrl = clientProperties.getCloudUrl();
+
+        Map<String, Object> param = new HashMap<>();
+
+        param.put("clusterId", clusterId);
+        param.put("appName", appName);
+
+        return restTemplate.getForObject("http://" + cloudUrl + CANCEL_NETWORK_MONITOR, String.class, param);
+    }
+
     /**
      * 获取容器监控的钩子程序
      * @param clusterId
@@ -110,12 +126,16 @@ public class NetworkServiceImpl implements NetworkService {
         param.put("clusterId", clusterId);
         param.put("appName", appName);
         // 这里需要构建这个ContainerMonitor
-        List<ContainerMonitorInfo> containerMonitorInfos
-                = restTemplate.getForObject("http://" + cloudUrl + GET_NETWORK_MONITOR_INFO, List.class, param);
+        //List<ContainerMonitorInfo> containerMonitorInfos
+        //        = restTemplate.getForObject("http://" + cloudUrl + GET_NETWORK_MONITOR_INFO, List.class, param);
+        ResponseEntity<List<ContainerMonitorInfo>> responseEntity
+                = restTemplate.exchange("http://" + cloudUrl + GET_NETWORK_MONITOR_INFO,
+                HttpMethod.GET, null, new ParameterizedTypeReference<List<ContainerMonitorInfo>>() {}, clusterId, appName);
+        List<ContainerMonitorInfo> containerMonitorInfos = responseEntity.getBody();
         List<ContainerMonitor> containerMonitors = new ArrayList<>();
-        containerMonitorInfos.forEach(c -> {
-            containerMonitors.add(new ContainerMonitor(c, restTemplate));
-        });
+        for (ContainerMonitorInfo containerMonitorInfo : containerMonitorInfos) {
+            containerMonitors.add(new ContainerMonitor(containerMonitorInfo, restTemplate));
+        }
         return containerMonitors;
     }
 }
\ No newline at end of file

bishe-cloud-center/src/main/java/top/ninwoo/cloudcenter/controller/NetworkController.java

@@ -34,6 +34,11 @@ public class NetworkController {
         return cloudService.enableNetworkMonitor(clusterId, appName);
     }
 
+    @RequestMapping(value = "/cancelNetworkMonitor")
+    public String cancelNetworkMonitor(Long clusterId, String appName) {
+        return cloudService.cancelNetworkMonitor(clusterId, appName);
+    }
+
     @RequestMapping(value = "/getNetworkMonitorInfo")
     public List<ContainerMonitorInfo> getNetworkMonitor(Long clusterId, String appName) {
         // 这里可以理解为是返回一个网络监控的钩子程序，具体的查询还需要通过钩子进行执行

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

@@ -27,4 +27,6 @@ public interface CloudService {
     String enableNetworkMonitor(Long clusterId, String appName);
 
     List<ContainerMonitorInfo> getContainerMonitorInfoByAppName(Long clusterId, String appName);
+
+    String cancelNetworkMonitor(Long clusterId, String appName);
 }

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

@@ -24,7 +24,7 @@ public class CloudServiceImpl implements CloudService {
     private static final String CANCEL_DROP_DOCKER_NETWORK = "/cancelDropDockerNetwork?clusterId={clusterId}&appName={appName}&ipList={ipList}";
     private static final String REMOTE_IP_LIST_BY_APPNAME = "/getIpListByAppName?clusterId={clusterId}&appName={appName}";
     private static final String ENABLE_NETWORK_MONITOR = "/enableContainerMonitor?clusterId={clusterId}&containerName={appName}";
-
+    private static final String CANCEL_NETWORK_MONITOR = "/cancelNetworkMonitor?clusterId={clusterId}&containerName={appName}";
     @Value("${bs.ipservice.url}")
     private String ipServiceUrl;
     // 全部的逻辑拓扑
@@ -33,10 +33,11 @@ public class CloudServiceImpl implements CloudService {
     // 用于存储配置文件
     private ConcurrentHashMap<Long, List<SeparatedClusterConfig>> allClusterConfig
             = new ConcurrentHashMap<>();
-
+    // 存储app在哪一个边缘节点
     private ConcurrentHashMap<Long, Map<String, Set<String>>> allAppStatus
             = new ConcurrentHashMap<>();
 
+    private ConcurrentHashMap<Long, Map<String, Map<String, Set<String>>>> allContainerIds = new ConcurrentHashMap<>();
     @Resource
     private CloudRegisterCenter cloudRegisterCenter;
 
@@ -225,12 +226,30 @@ public class CloudServiceImpl implements CloudService {
                 c.setEdgeNodeId(edgeNode);
             }
             // 下发配置
-            ResponseEntity<Long> response = restTemplate.postForEntity("http://" + c.getEdgeNodeId() + CREATE_CLUSTER, c.getClusterConfig(), Long.class);
+            ResponseEntity<Map> response = restTemplate.postForEntity("http://" + c.getEdgeNodeId() + CREATE_CLUSTER, c.getClusterConfig(), Map.class);
             if(!response.getStatusCode().is2xxSuccessful()) {
                 LOG.error("集群配置下发失败:{}:{}", c.getEdgeNodeId(), response.getBody());
 
             }
-
+            // 更新配置
+            //c.setClusterConfig(response.getBody());
+            // TODO: 存储appName和containerId的映射
+            if (!allContainerIds.containsKey(c.getClusterConfig().getId())) {
+                allContainerIds.put(c.getClusterConfig().getId(), new HashMap<>());
+            }
+            Map<String, Map<String, Set<String>>> allNames = allContainerIds.get(c.getClusterConfig().getId());
+            if (!allNames.containsKey(c.getEdgeNodeId())) {
+                allNames.put(c.getEdgeNodeId(), new HashMap<>());
+            }
+            Map<String, Set<String>> edgeContainerIds = allNames.get(c.getEdgeNodeId());
+            Map<String, List<String>> nameToIds = response.getBody();
+            nameToIds.forEach((name, v) -> {
+                if(!edgeContainerIds.containsKey(name)) {
+                    edgeContainerIds.put(name, new HashSet<>());
+                }
+                edgeContainerIds.get(name).addAll(v);
+            });
+            // TODO: 对应需要设计一个删除这个配置的操作
             // 构建初始化的逻辑拓扑
             for (String appName : c.getClusterConfig().getTopology().getAppNames()) {
                 if(!appName.startsWith("br")) {
@@ -428,32 +447,64 @@ public class CloudServiceImpl implements CloudService {
         return allLogicalTopo.get(clusterId);
     }
 
+    @Override
+    public String cancelNetworkMonitor(Long clusterId, String appName) {
+        if(!allAppStatus.containsKey(clusterId)) {
+            LOG.warn("正在取消一个不存在集群的网络监控");
+            return "failed";
+        }
+        if (!allAppStatus.get(clusterId).containsKey(appName)) {
+            LOG.warn("正在取消一个不存在的AppName的监控");
+            return "failed";
+        }
+        Set<String> nodeIds = allAppStatus.get(clusterId).get(appName);
+
+        // 遍历clusterAppName下发，容器监控模块
+        nodeIds.forEach(c -> {
+            if (!"".equals(c)) {
+                // 下发启用monitor监控的指令
+                Map<String, Object> param = new HashMap<>();
+                param.put("clusterId", clusterId);
+                param.put("appName", appName);
+                ResponseEntity<Boolean> response
+                        = restTemplate.getForEntity("http://" + c + CANCEL_NETWORK_MONITOR, Boolean.class, param);
+                if(!response.getStatusCode().is2xxSuccessful()) {
+                    LOG.warn("集群{}通信失败", c);
+                }
+                if(!response.getBody()) {
+                    LOG.warn("集群{},appName{}指令下发失败", c, appName);
+                }
+            }
+        });
+        return "success";
+    }
+
     @Override
     public String enableNetworkMonitor(Long clusterId, String appName) {
-        if(!allClusterConfig.containsKey(clusterId)) {
+        if(!allAppStatus.containsKey(clusterId)) {
             LOG.warn("正在启用一个不存在集群的网络监控");
             return "failed";
         }
-        if (!allClusterConfig.containsKey(appName)) {
+        if (!allAppStatus.get(clusterId).containsKey(appName)) {
             LOG.warn("正在启用一个不存在的AppName的监控");
             return "failed";
         }
-        List<SeparatedClusterConfig> separatedClusterConfigs = allClusterConfig.get(appName);
+        Set<String> nodeIds = allAppStatus.get(clusterId).get(appName);
 
         // 遍历clusterAppName下发，容器监控模块
-        separatedClusterConfigs.forEach(c -> {
-            if (c.getClusterConfig().getTopology().containsAppName(appName)) {
+        nodeIds.forEach(c -> {
+            if (!"".equals(c)) {
                 // 下发启用monitor监控的指令
                 Map<String, Object> param = new HashMap<>();
                 param.put("clusterId", clusterId);
                 param.put("appName", appName);
                 ResponseEntity<Boolean> response
-                        = restTemplate.getForEntity("http://" + c.getEdgeNodeId() + ENABLE_NETWORK_MONITOR, Boolean.class, param);
+                        = restTemplate.getForEntity("http://" + c + ENABLE_NETWORK_MONITOR, Boolean.class, param);
                 if(!response.getStatusCode().is2xxSuccessful()) {
-                    LOG.warn("集群{}通信失败", c.getEdgeNodeId());
+                    LOG.warn("集群{}通信失败", c);
                 }
                 if(!response.getBody()) {
-                    LOG.warn("集群{},appName{}指令下发失败", c.getEdgeNodeId(), appName);
+                    LOG.warn("集群{},appName{}指令下发失败", c, appName);
                 }
             }
         });
@@ -468,19 +519,21 @@ public class CloudServiceImpl implements CloudService {
             return containerMonitorInfos;
         }
         List<SeparatedClusterConfig> separatedClusterConfigs = allClusterConfig.get(clusterId);
-        if(!separatedClusterConfigs.contains(appName)) {
-            return containerMonitorInfos;
-        }
 
+        // TODO: 这里边没有需要的containerId
         separatedClusterConfigs.forEach( c -> {
             if(c.getClusterConfig().getTopology().containsAppName(appName)) {
                 for (ContainerDescription docker : c.getClusterConfig().getDockers()) {
-                    ContainerMonitorInfo monitorInfo = new ContainerMonitorInfo();
-                    monitorInfo.setAppName(appName);
-                    monitorInfo.setEdgeNodeId(c.getEdgeNodeId());
-                    monitorInfo.setClusterId(clusterId);
-                    monitorInfo.setContainerId(docker.getDockerContainer().getId());
-                    containerMonitorInfos.add(monitorInfo);
+                    // 从找到对应容器id
+                    Set<String> cids = allContainerIds.get(clusterId).get(c.getEdgeNodeId()).get(appName);
+                    cids.forEach( cid -> {
+                        ContainerMonitorInfo monitorInfo = new ContainerMonitorInfo();
+                        monitorInfo.setAppName(appName);
+                        monitorInfo.setEdgeNodeId(c.getEdgeNodeId());
+                        monitorInfo.setClusterId(clusterId);
+                        monitorInfo.setContainerId(cid);
+                        containerMonitorInfos.add(monitorInfo);
+                    });
                 }
             }
         });

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

@@ -85,12 +85,15 @@ public class IndexController {
             return false;
         }
         cids.forEach(cid -> {
-            // 为每一个cid容器创建一个定时任务
-            NetworkMonitorThread networkMonitorThread = new NetworkMonitorThread(cid, containerInfoMap, clusterService);
-            // 放入线程池执行任务
-            ScheduledFuture<?> future = scheduledExecutorService.scheduleAtFixedRate(networkMonitorThread, 0, 1, TimeUnit.SECONDS);
-            // 保存任务，以方便进行定时任务的关闭
-            scheduledFutureMap.put(cid, future);
+            // 为每一个cid容器创建一个定时任务 TODO 应该加一个判断，防止重复提交
+            if (!scheduledFutureMap.containsKey(cid)) {
+                containerInfoMap.put(cid, new ContainerInfo());
+                NetworkMonitorThread networkMonitorThread = new NetworkMonitorThread(cid, containerInfoMap, clusterService);
+                // 放入线程池执行任务
+                ScheduledFuture<?> future = scheduledExecutorService.scheduleAtFixedRate(networkMonitorThread, 0, 1, TimeUnit.SECONDS);
+                // 保存任务，以方便进行定时任务的关闭
+                scheduledFutureMap.put(cid, future);
+            }
         });
 
         return true;
@@ -105,9 +108,16 @@ public class IndexController {
         }
 
         cids.forEach(cid -> {
-            scheduledFutureMap.get(cid).cancel(true);
-            scheduledFutureMap.remove(cid);
-            containerInfoMap.remove(cid);
+            // TODO: 不安全的操作
+            //  cancel不安全，get(cid)不安全
+            if(scheduledFutureMap.containsKey(cid)) {
+                scheduledFutureMap.get(cid).cancel(false);
+                while (!scheduledFutureMap.get(cid).isCancelled()) {
+                    System.out.println("等待线程停止" + cid);
+                }
+                scheduledFutureMap.remove(cid);
+                containerInfoMap.remove(cid);
+            }
         });
 
         return true;
@@ -115,13 +125,13 @@ public class IndexController {
 
 
     @RequestMapping(value = "/createCluster", method = RequestMethod.POST)
-    public long createCluster(@RequestBody ClusterConfig clusterConfig) {
+    public Map<String, Set<String>> createCluster(@RequestBody ClusterConfig clusterConfig) {
         // 创建一个解析器
-        long l = clusterService.initContainers(clusterConfig);
+        Map<String, Set<String>> resMap = clusterService.initContainers(clusterConfig);
         // 创建网络
         // 使用topo创建工具
-        topologyService.createTopo(l, clusterConfig.getTopology());
-        return l;
+        topologyService.createTopo(clusterConfig.getId(), clusterConfig.getTopology());
+        return resMap;
     }
 
     @RequestMapping(value = "/getContainerIds")
@@ -150,7 +160,10 @@ public class IndexController {
             containerInfo.setId(containerId);
             containerInfo.setNetworkInfo(networkInfo);
             containerInfo.setMem(mem);
-            map.put(containerId, containerInfo);
+            if(map.containsKey(containerId)) {
+                map.put(containerId, containerInfo); // TODO: 这里有又重新把容器ｉｄ注册回去了
+            }
+            System.out.println("监控运行中" + containerId);
         }
 
         public void setContainerId(String containerId) {

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

@@ -6,6 +6,7 @@ import top.ninwoo.utils.entity.NetworkInfo;
 import top.ninwoo.utils.entity.OvsBridge;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -19,7 +20,7 @@ public interface ClusterService {
     List<DockerContainer> getContainerIds(boolean isAll);
 
     // 创建容器组
-    long initContainers(ClusterConfig clusterConfig);
+    Map<String, Set<String>> initContainers(ClusterConfig clusterConfig);
 
     // 动态调整容器规模，涉及到网络的一个调整
     String adjustCluster(ClusterConfig clusterConfig);

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

@@ -87,7 +87,7 @@ public class ClusterServiceImpl implements ClusterService {
      * @return
      */
     @Override
-    public long initContainers(ClusterConfig clusterConfig) {
+    public Map<String, Set<String>> initContainers(ClusterConfig clusterConfig) {
         // 校验ClusterConfig是否符合要求
         if(clusterConfig.getId() == 0L) {
             // 需要保存到数据库中，获取全局唯一的集群id
@@ -125,7 +125,7 @@ public class ClusterServiceImpl implements ClusterService {
             }
         });
         clusterConfig.setCreateTime(new Date());
-        return clusterConfig.getId();
+        return clustersInfo.get(clusterConfig.getId());
     }
 
     // 动态调整容器规模，涉及到网络的一个调整

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

@@ -58,8 +58,8 @@ public class ClusterServiceTest {
 
         clusterConfig.setDockers(cds);
 
-        long id = clusterService.initContainers(clusterConfig);
-        Assert.assertEquals(id, clusterId);
+        //long id = clusterService.initContainers(clusterConfig).getId();
+        //Assert.assertEquals(id, clusterId);
 
         Assert.assertTrue(clusterService.getClusterIds().contains(clusterId));
         System.out.println(clusterService.getContainerIdsByClusterId(clusterId, "APP"));

bishe-test/src/test/java/top/ninwoo/BisheTests.java

@@ -5,6 +5,7 @@ import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
+import top.ninwoo.bishe.starter.entity.ContainerMonitor;
 import top.ninwoo.bishe.starter.service.ClusterService;
 import top.ninwoo.bishe.starter.service.NetworkService;
 import top.ninwoo.common.entity.*;
@@ -85,4 +86,29 @@ public class BisheTests {
         List<String> ipList = networkService.getIpListByAppName(11111L, "Run");
         System.out.println(ipList);
     }
+
+    @Test
+    public void enableNetworkMonitorTest() throws InterruptedException {
+        String res = networkService.enableNetworkMonitor(11111L, "Run");
+        System.out.println(res);
+    }
+
+    @Test
+    public void cancelNetworkMonitorTest() {
+        String run = networkService.cancelNetworkMonitor(11111L, "Run");
+        System.out.println(run);
+    }
+
+    @Test
+    public void getNetworkMonitorTest() throws InterruptedException {
+        List<ContainerMonitor> run = networkService.getContainerMonitors(11111L, "Run");
+        int i = 0;
+        while(i < 10) {
+            run.forEach(c -> {
+                System.out.println(c.getContainerInfo());
+            });
+            Thread.sleep(500);
+            i++;
+        }
+    }
 }

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

@@ -98,7 +98,8 @@ public class DockerServiceImpl implements DockerService, InitializingBean {
         if("".equals(cid)) {
             throw new RuntimeException("不存在这个容器名称[" + name + "]");
         }
-        dockerUtils.deleteDockerById(name);
+        // TODO: 待测试
+        dockerUtils.deleteDockerById(cid);
         return cid;
     }