新增核心业务区

src/main/java/com/yingxin/prms/api/ReliabilityApi.java

 package com.yingxin.prms.api;
 
+import com.yingxin.prms.domain.FlowNode;
+import com.yingxin.prms.dto.SingleHostStatus;
+import com.yingxin.prms.service.asyncTask.QueryHostStatusTask;
 import com.yingxin.prms.service.business.Manager;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.task.AsyncTaskExecutor;
@@ -12,7 +15,9 @@ import javax.ws.rs.Path;
 import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Future;
 
 
 @RestController
@@ -81,5 +86,23 @@ public class ReliabilityApi {
     public Map selectChart(@RequestBody String json) throws Exception{
         return manager.selectChart(json);
     }
+    /**
+     *  获取拓扑图数据
+     *
+     * @return 集群信息json字符串
+     */
+    @PostMapping("selectTopology")
+    public List<FlowNode> selectTopology(){
+        return manager.selectTopology();
+    }
+    /**
+     *  获取拓扑图数据
+     *
+     * @return 集群信息json字符串
+     */
+    @PostMapping("saveTopology")
+    public Map saveTopology(@RequestBody String json) throws Exception{
+        return manager.saveTopology();
+    }
 
 }

src/main/java/com/yingxin/prms/domain/FlowLine.java

+package com.yingxin.prms.domain;
+
+import org.springframework.data.cassandra.core.mapping.Column;
+import org.springframework.data.cassandra.core.mapping.PrimaryKey;
+import org.springframework.data.cassandra.core.mapping.Table;
+
+/**
+ * Description:
+ * Datetime:    2021/1/12   9:40
+ * Author:   gaozhentao
+ */
+@Table(value = "flowNode")
+public class FlowLine {
+
+    @PrimaryKey
+    String id;  //主键id
+    @Column("cluster_name_left")
+    String clusterNameLeft;  //左集群名称
+    @Column("cluster_name_right")
+    String clusterNameRight;  //右集群名称
+    @Column("coordinate_left")
+    String coordinateLeft;  //左坐标
+    @Column("coordinate_right")
+    String coordinateRight;  //右坐标
+
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getClusterNameLeft() {
+        return clusterNameLeft;
+    }
+
+    public void setClusterNameLeft(String clusterNameLeft) {
+        this.clusterNameLeft = clusterNameLeft;
+    }
+
+    public String getClusterNameRight() {
+        return clusterNameRight;
+    }
+
+    public void setClusterNameRight(String clusterNameRight) {
+        this.clusterNameRight = clusterNameRight;
+    }
+
+    public String getCoordinateLeft() {
+        return coordinateLeft;
+    }
+
+    public void setCoordinateLeft(String coordinateLeft) {
+        this.coordinateLeft = coordinateLeft;
+    }
+
+    public String getCoordinateRight() {
+        return coordinateRight;
+    }
+
+    public void setCoordinateRight(String coordinateRight) {
+        this.coordinateRight = coordinateRight;
+    }
+}

src/main/java/com/yingxin/prms/domain/FlowNode.java

+package com.yingxin.prms.domain;
+
+import org.springframework.data.cassandra.core.mapping.Column;
+import org.springframework.data.cassandra.core.mapping.PrimaryKey;
+import org.springframework.data.cassandra.core.mapping.Table;
+
+/**
+ * Description:
+ * Datetime:    2021/1/12   9:40
+ * Author:   gaozhentao
+ */
+@Table(value = "flowNode")
+public class FlowNode {
+
+    @PrimaryKey
+    @Column("cluster_name")
+    String clusterName;  //集群名称
+    String coordinate;   //坐标
+    String mark;         //标识
+    @Column("server_path")
+    String serverPath;   //服务路径
+    String faultCount;   //错误数
+
+    public String getClusterName() {
+        return clusterName;
+    }
+
+    public void setClusterName(String clusterName) {
+        this.clusterName = clusterName;
+    }
+
+    public String getCoordinate() {
+        return coordinate;
+    }
+
+    public void setCoordinate(String coordinate) {
+        this.coordinate = coordinate;
+    }
+
+    public String getMark() {
+        return mark;
+    }
+
+    public void setMark(String mark) {
+        this.mark = mark;
+    }
+
+    public String getServerPath() {
+        return serverPath;
+    }
+
+    public void setServerPath(String serverPath) {
+        this.serverPath = serverPath;
+    }
+
+    public String getFaultCount() {
+        return faultCount;
+    }
+
+    public void setFaultCount(String faultCount) {
+        this.faultCount = faultCount;
+    }
+}

src/main/java/com/yingxin/prms/service/business/Manager.java

@@ -6,14 +6,13 @@ import com.yingxin.prms.domain.*;
 import com.yingxin.prms.dto.*;
 import com.yingxin.prms.service.asyncTask.QueryHostStatusTask;
 import com.yingxin.prms.service.dao.CassandraDaoImpl;
+import com.yingxin.prms.service.dao.FlowDaoImpl;
 import com.yingxin.prms.utils.CurveFittingMethod;
 import com.yingxin.prms.utils.TimeUtil;
 import com.yingxin.prms.utils.TokenUtil;
 import net.sf.json.JSONObject;
 import org.kairosdb.client.HttpClient;
-import org.kairosdb.client.builder.AggregatorFactory;
-import org.kairosdb.client.builder.DataPoint;
-import org.kairosdb.client.builder.QueryBuilder;
+import org.kairosdb.client.builder.*;
 import org.kairosdb.client.response.QueryResponse;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -27,6 +26,7 @@ import java.math.RoundingMode;
 import java.text.SimpleDateFormat;
 import java.util.*;
 import java.util.concurrent.*;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import static sun.misc.MessageUtils.where;
@@ -45,6 +45,9 @@ public class Manager {
 
     private final CassandraDaoImpl cassandraDao;
 
+    @Autowired
+    private FlowDaoImpl flowDao;
+
     private RestTemplate restTemplate;
 
     private final HttpClient httpClient;
@@ -112,11 +115,13 @@ public class Manager {
         List<SingleHostStatus> accessList = new ArrayList<>();
         List<SingleHostStatus> dataList = new ArrayList<>();
         List<SingleHostStatus> passportList = new ArrayList<>();
+        List<SingleHostStatus> coreList = new ArrayList<>();
 
         List<Future> accessFuture = new ArrayList<>();
         List<Future> dataFuture = new ArrayList<>();
         List<Future> passportFuture = new ArrayList<>();
         List<Future> futureList = new ArrayList<>();
+        List<Future> coreFuture = new ArrayList<>();
         List<SingleHostStatus> hostStatusList = new ArrayList<>();
         /*线程池*/
         ExecutorService service = Executors.newFixedThreadPool(THREADPOOLSIZE);
@@ -130,6 +135,8 @@ public class Manager {
                 dataFuture.add(result);
             }else if(hostInfo.getLayer().equals("护照柜")){
                 passportFuture.add(result);
+            }else if(hostInfo.getLayer().equals("核心业务区")){
+                coreFuture.add(result);
             }
         }
         service.shutdown();
@@ -151,6 +158,10 @@ public class Manager {
                     SingleHostStatus singleHostStatus = (SingleHostStatus) future.get();
                     passportList.add(singleHostStatus);
                 }
+                for (Future future : coreFuture) {
+                    SingleHostStatus singleHostStatus = (SingleHostStatus) future.get();
+                    coreList.add(singleHostStatus);
+                }
             }
         } catch (InterruptedException | ExecutionException e) {
             e.printStackTrace();
@@ -159,10 +170,12 @@ public class Manager {
         Map accessMap = selectClusterStatusBylayer(accessList);
         Map dataMap = selectClusterStatusBylayer(dataList);
         Map passportMap = selectClusterStatusBylayer(passportList);
+        Map coreMap = selectClusterStatusBylayer(coreList);
         Map map  = new HashMap();
         map.put("接入区",accessMap);
         map.put("算法区",dataMap);
         map.put("护照柜",passportMap);
+        map.put("核心业务区",coreMap);
         return map;
     }
     public Map selectClusterStatusBylayer(List<SingleHostStatus> hostStatusList){
@@ -385,6 +398,64 @@ public class Manager {
         return dataPoints;
     }
 
+    public List<FlowNode> selectTopology(){
+        List<Server_Host_Info> serverHostInfoAll = authModeAndHostList.getClusterBeMonitoredList();
+
+        List<FlowNode> flowNodeAll = flowDao.findNodeAll();
+        List<FlowNode> list = new ArrayList<>();
+        if(flowNodeAll.size()>0){
+            list = flowNodeAll;
+        }else{
+            Map<String,String> map = new HashMap();
+            for(Server_Host_Info hostInfo : serverHostInfoAll){
+                if(!map.containsKey(hostInfo.getCluster_name())){
+                    map.put(hostInfo.getCluster_name(),hostInfo.getService_path());
+                    FlowNode flowNode = new FlowNode();
+                    flowNode.setClusterName(hostInfo.getCluster_name());
+                    flowNode.setServerPath(hostInfo.getService_path());
+                    flowNode.setCoordinate("初始坐标");
+                    long faultCount;
+                    try {
+                        faultCount = getFaultCount(hostInfo.getService_path());
+                    } catch (Exception e) {
+                        faultCount = 0;
+                    }
+                    flowNode.setFaultCount(String.valueOf(faultCount));
+                    if(faultCount>0){
+                        flowNode.setMark("red");
+                    }else{
+                        flowNode.setMark("green");
+                    }
+                    list.add(flowNode);
+                    flowDao.insertNode(flowNode);
+                }
+            }
+        }
+       return list;
+    }
+
+    private long getFaultCount(String servicePath) throws IOException, DataFormatException {
+        Aggregator aggregatorDay = AggregatorFactory.createSumAggregator(1, org.kairosdb.client.builder.TimeUnit.DAYS).withStartTimeAlignment(0);
+        QueryBuilder queryBuilder = QueryBuilder.getInstance();
+        queryBuilder.setStart(1, org.kairosdb.client.builder.TimeUnit.DAYS).setTimeZone(TimeZone.getDefault())
+                .addMetric("heartbeat")
+                .addTag("host", servicePath)
+                .addAggregator(aggregatorDay);
+        QueryResponse response = httpClient.query(queryBuilder);
+        List<DataPoint> dataPointsDay = response.getQueries().get(0).getResults().get(0).getDataPoints();
+        return dataPointsDay.size() == 0 ? 0 : dataPointsDay.get(dataPointsDay.size() - 1).longValue();
+    }
+
+    public Map saveTopology() {
+        List<Server_Host_Info> serverHostInfoAll = authModeAndHostList.getClusterBeMonitoredList();
+        Map map  = new HashMap();
+        /*线程池*/
+        for(Server_Host_Info hostInfo : serverHostInfoAll){
+
+        }
+        return null;
+    }
+
     /*
     public Map selectClusterStatus() {
         List<Server_Host_Info> host_info_List = authModeAndHostList.getClusterBeMonitoredList();

src/main/java/com/yingxin/prms/service/dao/CassandraDaoImpl.java

@@ -2,8 +2,6 @@ package com.yingxin.prms.service.dao;
 
 import com.yingxin.prms.config.Constant;
 import com.yingxin.prms.domain.*;
-import com.yingxin.prms.utils.TimeUtil;
-import org.apache.tomcat.util.bcel.Const;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.cassandra.core.CassandraTemplate;
 import org.springframework.stereotype.Repository;
@@ -11,8 +9,6 @@ import org.springframework.stereotype.Repository;
 import java.util.List;
 import java.util.Map;
 
-import static org.springframework.data.cassandra.core.query.Criteria.where;
-import static org.springframework.data.cassandra.core.query.Query.query;
 
 @Repository
 public class CassandraDaoImpl {

src/main/java/com/yingxin/prms/service/dao/FlowDaoImpl.java

+package com.yingxin.prms.service.dao;
+
+import com.yingxin.prms.domain.FlowNode;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.cassandra.core.CassandraTemplate;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+
+/**
+ * Description:
+ * Datetime:    2021/1/12   9:53
+ * Author:   gaozhentao
+ */
+@Repository
+public class FlowDaoImpl {
+
+    private final CassandraTemplate template;
+
+    @Autowired
+    public FlowDaoImpl(CassandraTemplate template) {
+            this.template = template;
+        }
+
+
+    public void insertNode(FlowNode flowNode) {
+        String cql ="INSERT INTO ctid_prof.flownode(cluster_name, server_path)" +
+                "VALUES('"+flowNode.getClusterName()+"', '"+flowNode.getServerPath()+"');";
+        template.getCqlOperations().execute(cql);
+    }
+    public boolean updateNode(FlowNode flowNode) {
+        String cql ="UPDATE ctid_prof.flownode set coordinate='"+flowNode.getCoordinate()+"', mark='"+flowNode.getMark()+
+                "' WHERE  cluster_name='"+flowNode.getClusterName()+"'";
+        return template.getCqlOperations().execute(cql);
+    }
+
+
+    public List<FlowNode> findNodeAll() {
+        String cql ="SELECT * FROM ctid_prof.flownode";
+
+        return template.select(cql,FlowNode.class);
+    }
+}

src/main/resources/application.yml

@@ -33,7 +33,7 @@ demo:
 #    url: "http://172.16.15.7:8080"
 #    url: "http://192.168.189.130:8080"
 #    url: "http://200.102.10.11:8080"
-    url: "http://121.22.111.251:8080"
+    url: "http://121.22.111.251:9008"
     maxConnTotal: 50
     maxConnPerRoute: 30