Merge remote-tracking branch 'origin/master'

docs/user_guide/faq.md

@@ -37,7 +37,7 @@
 
 ## 8.4、`Jmx`连接失败如何解决？
 
-- 参看 [Jmx 连接配置&问题解决](./9-attachment#jmx-连接失败问题解决) 说明。
+- 参看 [Jmx 连接配置&问题解决](https://doc.knowstreaming.com/product/9-attachment#91jmx-%E8%BF%9E%E6%8E%A5%E5%A4%B1%E8%B4%A5%E9%97%AE%E9%A2%98%E8%A7%A3%E5%86%B3) 说明。
 
  
 

km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/version/impl/VersionControlManagerImpl.java

@@ -14,7 +14,6 @@ import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.version.VersionControlItem;
 import com.xiaojukeji.know.streaming.km.common.bean.vo.config.metric.UserMetricConfigVO;
 import com.xiaojukeji.know.streaming.km.common.bean.vo.version.VersionItemVO;
-import com.xiaojukeji.know.streaming.km.common.constant.Constant;
 import com.xiaojukeji.know.streaming.km.common.enums.version.VersionEnum;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
 import com.xiaojukeji.know.streaming.km.common.utils.VersionUtil;
@@ -108,6 +107,7 @@ public class VersionControlManagerImpl implements VersionControlManager {
         allVersionItemVO.addAll(ConvertUtil.list2List(versionControlService.listVersionControlItem(METRIC_BROKER.getCode()), VersionItemVO.class));
         allVersionItemVO.addAll(ConvertUtil.list2List(versionControlService.listVersionControlItem(METRIC_PARTITION.getCode()), VersionItemVO.class));
         allVersionItemVO.addAll(ConvertUtil.list2List(versionControlService.listVersionControlItem(METRIC_REPLICATION.getCode()), VersionItemVO.class));
+        allVersionItemVO.addAll(ConvertUtil.list2List(versionControlService.listVersionControlItem(METRIC_ZOOKEEPER.getCode()), VersionItemVO.class));
         allVersionItemVO.addAll(ConvertUtil.list2List(versionControlService.listVersionControlItem(WEB_OP.getCode()), VersionItemVO.class));
 
         Map<String, VersionItemVO> map = allVersionItemVO.stream().collect(

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/vo/zookeeper/ClusterZookeepersOverviewVO.java

@@ -14,6 +14,9 @@ public class ClusterZookeepersOverviewVO  {
     @ApiModelProperty(value = "主机ip", example = "121.0.0.1")
     private String host;
 
+    @ApiModelProperty(value = "主机存活状态，1：Live，0：Down", example = "1")
+    private Integer status;
+
     @ApiModelProperty(value = "端口号", example = "2416")
     private Integer port;
 

km-extends/km-monitor/src/main/java/com/xiaojukeji/know/streaming/km/monitor/component/AbstractMonitorSinkService.java

@@ -37,29 +37,32 @@ public abstract class AbstractMonitorSinkService implements ApplicationListener<
     @Override
     public void onApplicationEvent(BaseMetricEvent event) {
         executor.execute( () -> {
-            if(event instanceof BrokerMetricEvent){
+            if (event instanceof BrokerMetricEvent) {
                 BrokerMetricEvent brokerMetricEvent = (BrokerMetricEvent)event;
                 sinkMetrics(brokerMetric2SinkPoint(brokerMetricEvent.getBrokerMetrics()));
 
-            }else if(event instanceof ClusterMetricEvent){
+            } else if(event instanceof ClusterMetricEvent) {
                 ClusterMetricEvent clusterMetricEvent = (ClusterMetricEvent)event;
                 sinkMetrics(clusterMetric2SinkPoint(clusterMetricEvent.getClusterMetrics()));
 
-            }else if(event instanceof TopicMetricEvent){
+            } else if(event instanceof TopicMetricEvent) {
                 TopicMetricEvent topicMetricEvent = (TopicMetricEvent)event;
                 sinkMetrics(topicMetric2SinkPoint(topicMetricEvent.getTopicMetrics()));
 
-            }else if(event instanceof PartitionMetricEvent){
+            } else if(event instanceof PartitionMetricEvent) {
                 PartitionMetricEvent   partitionMetricEvent = (PartitionMetricEvent)event;
                 sinkMetrics(partitionMetric2SinkPoint(partitionMetricEvent.getPartitionMetrics()));
 
-            }else if(event instanceof GroupMetricEvent){
+            } else if(event instanceof GroupMetricEvent) {
                 GroupMetricEvent groupMetricEvent = (GroupMetricEvent)event;
                 sinkMetrics(groupMetric2SinkPoint(groupMetricEvent.getGroupMetrics()));
 
-            }else if(event instanceof ReplicaMetricEvent){
+            } else if(event instanceof ReplicaMetricEvent) {
                 ReplicaMetricEvent       replicaMetricEvent = (ReplicaMetricEvent)event;
                 sinkMetrics(replicationMetric2SinkPoint(replicaMetricEvent.getReplicationMetrics()));
+            } else if(event instanceof ZookeeperMetricEvent) {
+                ZookeeperMetricEvent     zookeeperMetricEvent = (ZookeeperMetricEvent)event;
+                sinkMetrics(zookeeperMetric2SinkPoint(zookeeperMetricEvent.getZookeeperMetrics()));
             }
         } );
     }
@@ -72,6 +75,7 @@ public abstract class AbstractMonitorSinkService implements ApplicationListener<
     public abstract Boolean sinkMetrics(List<MetricSinkPoint> pointList);
 
     /**************************************************** private method ****************************************************/
+
     private List<MetricSinkPoint> brokerMetric2SinkPoint(List<BrokerMetrics> brokerMetrics){
         List<MetricSinkPoint> pointList = new ArrayList<>();
 
@@ -161,8 +165,23 @@ public abstract class AbstractMonitorSinkService implements ApplicationListener<
         return pointList;
     }
 
-    private List<MetricSinkPoint> genSinkPoint(String metricPre, Map<String, Float> metrics,
+    private List<MetricSinkPoint> zookeeperMetric2SinkPoint(List<ZookeeperMetrics> zookeeperMetricsList){
-                                               long timeStamp, Map<String, Object> tagsMap){
+        List<MetricSinkPoint> pointList = new ArrayList<>();
+
+        for(ZookeeperMetrics z : zookeeperMetricsList){
+            Map<String, Object> tagsMap = new HashMap<>();
+            tagsMap.put(CLUSTER_ID.getName(),     z.getClusterPhyId());
+
+            pointList.addAll(genSinkPoint("Zookeeper", z.getMetrics(), z.getTimestamp(), tagsMap));
+        }
+
+        return pointList;
+    }
+
+    private List<MetricSinkPoint> genSinkPoint(String metricPre,
+                                               Map<String, Float> metrics,
+                                               long timeStamp,
+                                               Map<String, Object> tagsMap) {
         List<MetricSinkPoint> pointList = new ArrayList<>();
 
         for(String metricName : metrics.keySet()){

km-rest/src/main/java/com/xiaojukeji/know/streaming/km/rest/api/v3/version/VersionController.java

@@ -15,7 +15,6 @@ import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 
 import javax.servlet.http.HttpServletRequest;
-import javax.validation.Valid;
 import java.util.List;
 import java.util.Map;
 import java.util.SortedMap;

km-rest/src/main/resources/application.yml

@@ -84,7 +84,8 @@ client-pool:
 es:
   client:
     address: 127.0.0.1:8091,127.0.0.1:8061,127.0.0.1:8061
-    client-cnt: 10
+    pass:                         # ES账号密码，如果有账号密码，按照 username:password 的格式填写，没有则不需要填写
+    client-cnt: 10                # 创建的ES客户端数
     io-thread-cnt: 2
     max-retry-cnt: 5
 

km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/health/AbstractHealthCheckTask.java

@@ -0,0 +1,115 @@
+package com.xiaojukeji.know.streaming.km.task.health;
+
+import com.didiglobal.logi.job.common.TaskResult;
+import com.didiglobal.logi.log.ILog;
+import com.didiglobal.logi.log.LogFactory;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.config.healthcheck.BaseClusterHealthConfig;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.health.HealthCheckResult;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.param.cluster.ClusterPhyParam;
+import com.xiaojukeji.know.streaming.km.common.constant.Constant;
+import com.xiaojukeji.know.streaming.km.common.enums.health.HealthCheckDimensionEnum;
+import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
+import com.xiaojukeji.know.streaming.km.core.service.health.checker.AbstractHealthCheckService;
+import com.xiaojukeji.know.streaming.km.core.service.health.checkresult.HealthCheckResultService;
+import com.xiaojukeji.know.streaming.km.task.metrics.AbstractAsyncMetricsDispatchTask;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+
+public abstract class AbstractHealthCheckTask extends AbstractAsyncMetricsDispatchTask {
+    private static final ILog log = LogFactory.getLog(AbstractHealthCheckTask.class);
+
+    @Autowired
+    private HealthCheckResultService healthCheckResultService;
+
+    public abstract AbstractHealthCheckService getCheckService();
+
+    @Override
+    public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) {
+        return this.calAndUpdateHealthCheckResult(clusterPhy, triggerTimeUnitMs);
+    }
+
+    private TaskResult calAndUpdateHealthCheckResult(ClusterPhy clusterPhy, long triggerTimeUnitMs) {
+        // 获取配置，<配置名，配置信息>
+        Map<String, BaseClusterHealthConfig> healthConfigMap = healthCheckResultService.getClusterHealthConfig(clusterPhy.getId());
+
+        // 检查结果
+        List<HealthCheckResult> resultList = new ArrayList<>();
+
+        // 遍历Check-Service
+        List<ClusterPhyParam> paramList = this.getCheckService().getResList(clusterPhy.getId());
+        if (ValidateUtils.isEmptyList(paramList)) {
+            // 当前无该维度的资源，则直接设置为
+            resultList.addAll(this.getNoResResult(clusterPhy.getId(), this.getCheckService(), healthConfigMap));
+        }
+
+        // 遍历资源
+        for (ClusterPhyParam clusterPhyParam: paramList) {
+            resultList.addAll(this.checkAndGetResult(clusterPhyParam, healthConfigMap));
+        }
+
+        for (HealthCheckResult checkResult: resultList) {
+            try {
+                healthCheckResultService.replace(checkResult);
+            } catch (Exception e) {
+                log.error("class=AbstractHealthCheckTask||method=processSubTask||clusterPhyId={}||checkResult={}||errMsg=exception!", clusterPhy.getId(), checkResult, e);
+            }
+        }
+
+        // 删除10分钟之前的检查结果
+        try {
+            healthCheckResultService.deleteByUpdateTimeBeforeInDB(clusterPhy.getId(), new Date(triggerTimeUnitMs - 10 * 60 * 1000));
+        } catch (Exception e) {
+            log.error("class=AbstractHealthCheckTask||method=processSubTask||clusterPhyId={}||errMsg=exception!", clusterPhy.getId(), e);
+        }
+
+        return TaskResult.SUCCESS;
+    }
+
+    private List<HealthCheckResult> getNoResResult(Long clusterPhyId, AbstractHealthCheckService healthCheckService, Map<String, BaseClusterHealthConfig> healthConfigMap) {
+        List<HealthCheckResult> resultList = new ArrayList<>();
+
+        // 进行检查
+        for (BaseClusterHealthConfig clusterHealthConfig: healthConfigMap.values()) {
+            HealthCheckDimensionEnum dimensionEnum = healthCheckService.getHealthCheckDimensionEnum();
+            if (!clusterHealthConfig.getCheckNameEnum().getDimensionEnum().equals(dimensionEnum)) {
+                // 类型不匹配
+                continue;
+            }
+
+            // 记录
+            HealthCheckResult checkResult = new HealthCheckResult(
+                    dimensionEnum.getDimension(),
+                    clusterHealthConfig.getCheckNameEnum().getConfigName(),
+                    clusterPhyId,
+                    "-1"
+            );
+            checkResult.setPassed(Constant.YES);
+            resultList.add(checkResult);
+        }
+
+        return resultList;
+    }
+
+    private List<HealthCheckResult> checkAndGetResult(ClusterPhyParam clusterPhyParam,
+                                                      Map<String, BaseClusterHealthConfig> healthConfigMap) {
+        List<HealthCheckResult> resultList = new ArrayList<>();
+
+        // 进行检查
+        for (BaseClusterHealthConfig clusterHealthConfig: healthConfigMap.values()) {
+            HealthCheckResult healthCheckResult = this.getCheckService().checkAndGetResult(clusterPhyParam, clusterHealthConfig);
+            if (healthCheckResult == null) {
+                continue;
+            }
+
+            // 记录
+            resultList.add(healthCheckResult);
+        }
+
+        return resultList;
+    }
+}

km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/health/BrokerHealthCheckTask.java

@@ -1,30 +1,13 @@
 package com.xiaojukeji.know.streaming.km.task.health;
 
 import com.didiglobal.logi.job.annotation.Task;
-import com.didiglobal.logi.job.common.TaskResult;
 import com.didiglobal.logi.job.core.consensual.ConsensualEnum;
-import com.didiglobal.logi.log.ILog;
-import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.config.healthcheck.BaseClusterHealthConfig;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.health.HealthCheckResult;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.param.cluster.ClusterPhyParam;
-import com.xiaojukeji.know.streaming.km.common.constant.Constant;
-import com.xiaojukeji.know.streaming.km.common.enums.health.HealthCheckDimensionEnum;
-import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
 import com.xiaojukeji.know.streaming.km.core.service.health.checker.AbstractHealthCheckService;
 import com.xiaojukeji.know.streaming.km.core.service.health.checker.broker.HealthCheckBrokerService;
-import com.xiaojukeji.know.streaming.km.core.service.health.checkresult.HealthCheckResultService;
-import com.xiaojukeji.know.streaming.km.task.metrics.AbstractAsyncMetricsDispatchTask;
 import lombok.AllArgsConstructor;
 import lombok.NoArgsConstructor;
 import org.springframework.beans.factory.annotation.Autowired;
 
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
-
 @NoArgsConstructor
 @AllArgsConstructor
 @Task(name = "BrokerHealthCheckTask",
@@ -33,98 +16,12 @@ import java.util.Map;
         autoRegister = true,
         consensual = ConsensualEnum.BROADCAST,
         timeout = 2 * 60)
-public class BrokerHealthCheckTask extends AbstractAsyncMetricsDispatchTask {
+public class BrokerHealthCheckTask extends AbstractHealthCheckTask {
-    private static final ILog log = LogFactory.getLog(BrokerHealthCheckTask.class);
-
-    @Autowired
-    private HealthCheckResultService healthCheckResultService;
-
     @Autowired
     private HealthCheckBrokerService healthCheckBrokerService;
 
     @Override
-    public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) {
+    public AbstractHealthCheckService getCheckService() {
-        return this.calAndUpdateHealthCheckResult(clusterPhy, triggerTimeUnitMs);
+        return healthCheckBrokerService;
-    }
-
-    private TaskResult calAndUpdateHealthCheckResult(ClusterPhy clusterPhy, long triggerTimeUnitMs) {
-        // 获取配置，<配置名，配置信息>
-        Map<String, BaseClusterHealthConfig> healthConfigMap = healthCheckResultService.getClusterHealthConfig(clusterPhy.getId());
-
-        // 检查结果
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 遍历Check-Service
-        List<ClusterPhyParam> paramList = healthCheckBrokerService.getResList(clusterPhy.getId());
-        if (ValidateUtils.isEmptyList(paramList)) {
-            // 当前无该维度的资源，则直接设置为
-            resultList.addAll(this.getNoResResult(clusterPhy.getId(), healthCheckBrokerService, healthConfigMap));
-        }
-
-        // 遍历资源
-        for (ClusterPhyParam clusterPhyParam: paramList) {
-            resultList.addAll(this.checkAndGetResult(healthCheckBrokerService, clusterPhyParam, healthConfigMap));
-        }
-
-        for (HealthCheckResult checkResult: resultList) {
-            try {
-                healthCheckResultService.replace(checkResult);
-            } catch (Exception e) {
-                log.error("class=BrokerHealthCheckTask||method=processSubTask||clusterPhyId={}||checkResult={}||errMsg=exception!", clusterPhy.getId(), checkResult, e);
-            }
-        }
-
-        // 删除10分钟之前的检查结果
-        try {
-            healthCheckResultService.deleteByUpdateTimeBeforeInDB(clusterPhy.getId(), new Date(triggerTimeUnitMs - 10 * 60 * 1000));
-        } catch (Exception e) {
-            log.error("class=BrokerHealthCheckTask||method=processSubTask||clusterPhyId={}||errMsg=exception!", clusterPhy.getId(), e);
-        }
-
-        return TaskResult.SUCCESS;
-    }
-
-    private List<HealthCheckResult> getNoResResult(Long clusterPhyId, AbstractHealthCheckService healthCheckService, Map<String, BaseClusterHealthConfig> healthConfigMap) {
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 进行检查
-        for (BaseClusterHealthConfig clusterHealthConfig: healthConfigMap.values()) {
-            HealthCheckDimensionEnum dimensionEnum = healthCheckService.getHealthCheckDimensionEnum();
-            if (!clusterHealthConfig.getCheckNameEnum().getDimensionEnum().equals(dimensionEnum)) {
-                // 类型不匹配
-                continue;
-            }
-
-            // 记录
-            HealthCheckResult checkResult = new HealthCheckResult(
-                    dimensionEnum.getDimension(),
-                    clusterHealthConfig.getCheckNameEnum().getConfigName(),
-                    clusterPhyId,
-                    "-1"
-            );
-            checkResult.setPassed(Constant.YES);
-            resultList.add(checkResult);
-        }
-
-        return resultList;
-    }
-
-    private List<HealthCheckResult> checkAndGetResult(AbstractHealthCheckService healthCheckService,
-                                                      ClusterPhyParam clusterPhyParam,
-                                                      Map<String, BaseClusterHealthConfig> healthConfigMap) {
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 进行检查
-        for (BaseClusterHealthConfig clusterHealthConfig: healthConfigMap.values()) {
-            HealthCheckResult healthCheckResult = healthCheckService.checkAndGetResult(clusterPhyParam, clusterHealthConfig);
-            if (healthCheckResult == null) {
-                continue;
-            }
-
-            // 记录
-            resultList.add(healthCheckResult);
-        }
-
-        return resultList;
     }
 }

km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/health/ClusterHealthCheckTask.java

@@ -1,30 +1,13 @@
 package com.xiaojukeji.know.streaming.km.task.health;
 
 import com.didiglobal.logi.job.annotation.Task;
-import com.didiglobal.logi.job.common.TaskResult;
 import com.didiglobal.logi.job.core.consensual.ConsensualEnum;
-import com.didiglobal.logi.log.ILog;
-import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.config.healthcheck.BaseClusterHealthConfig;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.health.HealthCheckResult;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.param.cluster.ClusterPhyParam;
-import com.xiaojukeji.know.streaming.km.common.constant.Constant;
-import com.xiaojukeji.know.streaming.km.common.enums.health.HealthCheckDimensionEnum;
-import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
 import com.xiaojukeji.know.streaming.km.core.service.health.checker.AbstractHealthCheckService;
 import com.xiaojukeji.know.streaming.km.core.service.health.checker.cluster.HealthCheckClusterService;
-import com.xiaojukeji.know.streaming.km.core.service.health.checkresult.HealthCheckResultService;
-import com.xiaojukeji.know.streaming.km.task.metrics.AbstractAsyncMetricsDispatchTask;
 import lombok.AllArgsConstructor;
 import lombok.NoArgsConstructor;
 import org.springframework.beans.factory.annotation.Autowired;
 
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
-
 @NoArgsConstructor
 @AllArgsConstructor
 @Task(name = "ClusterHealthCheckTask",
@@ -33,98 +16,12 @@ import java.util.Map;
         autoRegister = true,
         consensual = ConsensualEnum.BROADCAST,
         timeout = 2 * 60)
-public class ClusterHealthCheckTask extends AbstractAsyncMetricsDispatchTask {
+public class ClusterHealthCheckTask extends AbstractHealthCheckTask {
-    private static final ILog log = LogFactory.getLog(ClusterHealthCheckTask.class);
-
-    @Autowired
-    private HealthCheckResultService healthCheckResultService;
-
     @Autowired
     private HealthCheckClusterService healthCheckClusterService;
 
     @Override
-    public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) {
+    public AbstractHealthCheckService getCheckService() {
-        return this.calAndUpdateHealthCheckResult(clusterPhy, triggerTimeUnitMs);
+        return healthCheckClusterService;
-    }
-
-    private TaskResult calAndUpdateHealthCheckResult(ClusterPhy clusterPhy, long triggerTimeUnitMs) {
-        // 获取配置，<配置名，配置信息>
-        Map<String, BaseClusterHealthConfig> healthConfigMap = healthCheckResultService.getClusterHealthConfig(clusterPhy.getId());
-
-        // 检查结果
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 遍历Check-Service
-        List<ClusterPhyParam> paramList = healthCheckClusterService.getResList(clusterPhy.getId());
-        if (ValidateUtils.isEmptyList(paramList)) {
-            // 当前无该维度的资源，则直接设置为
-            resultList.addAll(this.getNoResResult(clusterPhy.getId(), healthCheckClusterService, healthConfigMap));
-        }
-
-        // 遍历资源
-        for (ClusterPhyParam clusterPhyParam: paramList) {
-            resultList.addAll(this.checkAndGetResult(healthCheckClusterService, clusterPhyParam, healthConfigMap));
-        }
-
-        for (HealthCheckResult checkResult: resultList) {
-            try {
-                healthCheckResultService.replace(checkResult);
-            } catch (Exception e) {
-                log.error("class=ClusterHealthCheckTask||method=processSubTask||clusterPhyId={}||checkResult={}||errMsg=exception!", clusterPhy.getId(), checkResult, e);
-            }
-        }
-
-        // 删除10分钟之前的检查结果
-        try {
-            healthCheckResultService.deleteByUpdateTimeBeforeInDB(clusterPhy.getId(), new Date(triggerTimeUnitMs - 10 * 60 * 1000));
-        } catch (Exception e) {
-            log.error("class=ClusterHealthCheckTask||method=processSubTask||clusterPhyId={}||errMsg=exception!", clusterPhy.getId(), e);
-        }
-
-        return TaskResult.SUCCESS;
-    }
-
-    private List<HealthCheckResult> getNoResResult(Long clusterPhyId, AbstractHealthCheckService healthCheckService, Map<String, BaseClusterHealthConfig> healthConfigMap) {
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 进行检查
-        for (BaseClusterHealthConfig clusterHealthConfig: healthConfigMap.values()) {
-            HealthCheckDimensionEnum dimensionEnum = healthCheckService.getHealthCheckDimensionEnum();
-            if (!clusterHealthConfig.getCheckNameEnum().getDimensionEnum().equals(dimensionEnum)) {
-                // 类型不匹配
-                continue;
-            }
-
-            // 记录
-            HealthCheckResult checkResult = new HealthCheckResult(
-                    dimensionEnum.getDimension(),
-                    clusterHealthConfig.getCheckNameEnum().getConfigName(),
-                    clusterPhyId,
-                    "-1"
-            );
-            checkResult.setPassed(Constant.YES);
-            resultList.add(checkResult);
-        }
-
-        return resultList;
-    }
-
-    private List<HealthCheckResult> checkAndGetResult(AbstractHealthCheckService healthCheckService,
-                                                      ClusterPhyParam clusterPhyParam,
-                                                      Map<String, BaseClusterHealthConfig> healthConfigMap) {
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 进行检查
-        for (BaseClusterHealthConfig clusterHealthConfig: healthConfigMap.values()) {
-            HealthCheckResult healthCheckResult = healthCheckService.checkAndGetResult(clusterPhyParam, clusterHealthConfig);
-            if (healthCheckResult == null) {
-                continue;
-            }
-
-            // 记录
-            resultList.add(healthCheckResult);
-        }
-
-        return resultList;
     }
 }

km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/health/GroupHealthCheckTask.java

@@ -1,29 +1,13 @@
 package com.xiaojukeji.know.streaming.km.task.health;
 
 import com.didiglobal.logi.job.annotation.Task;
-import com.didiglobal.logi.job.common.TaskResult;
 import com.didiglobal.logi.job.core.consensual.ConsensualEnum;
-import com.didiglobal.logi.log.ILog;
-import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.config.healthcheck.BaseClusterHealthConfig;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.health.HealthCheckResult;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.param.cluster.ClusterPhyParam;
-import com.xiaojukeji.know.streaming.km.common.constant.Constant;
-import com.xiaojukeji.know.streaming.km.common.enums.health.HealthCheckDimensionEnum;
-import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
 import com.xiaojukeji.know.streaming.km.core.service.health.checker.AbstractHealthCheckService;
 import com.xiaojukeji.know.streaming.km.core.service.health.checker.group.HealthCheckGroupService;
-import com.xiaojukeji.know.streaming.km.core.service.health.checkresult.HealthCheckResultService;
-import com.xiaojukeji.know.streaming.km.task.metrics.AbstractAsyncMetricsDispatchTask;
 import lombok.AllArgsConstructor;
 import lombok.NoArgsConstructor;
 import org.springframework.beans.factory.annotation.Autowired;
 
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
 
 @NoArgsConstructor
 @AllArgsConstructor
@@ -33,98 +17,13 @@ import java.util.Map;
         autoRegister = true,
         consensual = ConsensualEnum.BROADCAST,
         timeout = 2 * 60)
-public class GroupHealthCheckTask extends AbstractAsyncMetricsDispatchTask {
+public class GroupHealthCheckTask extends AbstractHealthCheckTask {
-    private static final ILog log = LogFactory.getLog(GroupHealthCheckTask.class);
-
-    @Autowired
-    private HealthCheckResultService healthCheckResultService;
 
     @Autowired
     private HealthCheckGroupService healthCheckGroupService;
 
     @Override
-    public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) {
+    public AbstractHealthCheckService getCheckService() {
-        return this.calAndUpdateHealthCheckResult(clusterPhy, triggerTimeUnitMs);
+        return healthCheckGroupService;
-    }
-
-    private TaskResult calAndUpdateHealthCheckResult(ClusterPhy clusterPhy, long triggerTimeUnitMs) {
-        // 获取配置，<配置名，配置信息>
-        Map<String, BaseClusterHealthConfig> healthConfigMap = healthCheckResultService.getClusterHealthConfig(clusterPhy.getId());
-
-        // 检查结果
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 遍历Check-Service
-        List<ClusterPhyParam> paramList = healthCheckGroupService.getResList(clusterPhy.getId());
-        if (ValidateUtils.isEmptyList(paramList)) {
-            // 当前无该维度的资源，则直接设置为
-            resultList.addAll(this.getNoResResult(clusterPhy.getId(), healthCheckGroupService, healthConfigMap));
-        }
-
-        // 遍历资源
-        for (ClusterPhyParam clusterPhyParam: paramList) {
-            resultList.addAll(this.checkAndGetResult(healthCheckGroupService, clusterPhyParam, healthConfigMap));
-        }
-
-        for (HealthCheckResult checkResult: resultList) {
-            try {
-                healthCheckResultService.replace(checkResult);
-            } catch (Exception e) {
-                log.error("class=GroupHealthCheckTask||method=processSubTask||clusterPhyId={}||checkResult={}||errMsg=exception!", clusterPhy.getId(), checkResult, e);
-            }
-        }
-
-        // 删除10分钟之前的检查结果
-        try {
-            healthCheckResultService.deleteByUpdateTimeBeforeInDB(clusterPhy.getId(), new Date(triggerTimeUnitMs - 10 * 60 * 1000));
-        } catch (Exception e) {
-            log.error("class=GroupHealthCheckTask||method=processSubTask||clusterPhyId={}||errMsg=exception!", clusterPhy.getId(), e);
-        }
-
-        return TaskResult.SUCCESS;
-    }
-
-    private List<HealthCheckResult> getNoResResult(Long clusterPhyId, AbstractHealthCheckService healthCheckService, Map<String, BaseClusterHealthConfig> healthConfigMap) {
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 进行检查
-        for (BaseClusterHealthConfig clusterHealthConfig: healthConfigMap.values()) {
-            HealthCheckDimensionEnum dimensionEnum = healthCheckService.getHealthCheckDimensionEnum();
-            if (!clusterHealthConfig.getCheckNameEnum().getDimensionEnum().equals(dimensionEnum)) {
-                // 类型不匹配
-                continue;
-            }
-
-            // 记录
-            HealthCheckResult checkResult = new HealthCheckResult(
-                    dimensionEnum.getDimension(),
-                    clusterHealthConfig.getCheckNameEnum().getConfigName(),
-                    clusterPhyId,
-                    "-1"
-            );
-            checkResult.setPassed(Constant.YES);
-            resultList.add(checkResult);
-        }
-
-        return resultList;
-    }
-
-    private List<HealthCheckResult> checkAndGetResult(AbstractHealthCheckService healthCheckService,
-                                                      ClusterPhyParam clusterPhyParam,
-                                                      Map<String, BaseClusterHealthConfig> healthConfigMap) {
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 进行检查
-        for (BaseClusterHealthConfig clusterHealthConfig: healthConfigMap.values()) {
-            HealthCheckResult healthCheckResult = healthCheckService.checkAndGetResult(clusterPhyParam, clusterHealthConfig);
-            if (healthCheckResult == null) {
-                continue;
-            }
-
-            // 记录
-            resultList.add(healthCheckResult);
-        }
-
-        return resultList;
     }
 }

km-task/src/main/java/com/xiaojukeji/know/streaming/km/task/health/TopicHealthCheckTask.java

@@ -1,30 +1,13 @@
 package com.xiaojukeji.know.streaming.km.task.health;
 
 import com.didiglobal.logi.job.annotation.Task;
-import com.didiglobal.logi.job.common.TaskResult;
 import com.didiglobal.logi.job.core.consensual.ConsensualEnum;
-import com.didiglobal.logi.log.ILog;
-import com.didiglobal.logi.log.LogFactory;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.config.healthcheck.BaseClusterHealthConfig;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.health.HealthCheckResult;
-import com.xiaojukeji.know.streaming.km.common.bean.entity.param.cluster.ClusterPhyParam;
-import com.xiaojukeji.know.streaming.km.common.constant.Constant;
-import com.xiaojukeji.know.streaming.km.common.enums.health.HealthCheckDimensionEnum;
-import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
 import com.xiaojukeji.know.streaming.km.core.service.health.checker.AbstractHealthCheckService;
 import com.xiaojukeji.know.streaming.km.core.service.health.checker.topic.HealthCheckTopicService;
-import com.xiaojukeji.know.streaming.km.core.service.health.checkresult.HealthCheckResultService;
-import com.xiaojukeji.know.streaming.km.task.metrics.AbstractAsyncMetricsDispatchTask;
 import lombok.AllArgsConstructor;
 import lombok.NoArgsConstructor;
 import org.springframework.beans.factory.annotation.Autowired;
 
-import java.util.ArrayList;
-import java.util.Date;
-import java.util.List;
-import java.util.Map;
-
 @NoArgsConstructor
 @AllArgsConstructor
 @Task(name = "TopicHealthCheckTask",
@@ -33,98 +16,13 @@ import java.util.Map;
         autoRegister = true,
         consensual = ConsensualEnum.BROADCAST,
         timeout = 2 * 60)
-public class TopicHealthCheckTask extends AbstractAsyncMetricsDispatchTask {
+public class TopicHealthCheckTask extends AbstractHealthCheckTask {
-    private static final ILog log = LogFactory.getLog(TopicHealthCheckTask.class);
-
-    @Autowired
-    private HealthCheckResultService healthCheckResultService;
 
     @Autowired
     private HealthCheckTopicService healthCheckTopicService;
 
     @Override
-    public TaskResult processClusterTask(ClusterPhy clusterPhy, long triggerTimeUnitMs) {
+    public AbstractHealthCheckService getCheckService() {
-        return this.calAndUpdateHealthCheckResult(clusterPhy, triggerTimeUnitMs);
+        return healthCheckTopicService;
-    }
-
-    private TaskResult calAndUpdateHealthCheckResult(ClusterPhy clusterPhy, long triggerTimeUnitMs) {
-        // 获取配置，<配置名，配置信息>
-        Map<String, BaseClusterHealthConfig> healthConfigMap = healthCheckResultService.getClusterHealthConfig(clusterPhy.getId());
-
-        // 检查结果
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 遍历Check-Service
-        List<ClusterPhyParam> paramList = healthCheckTopicService.getResList(clusterPhy.getId());
-        if (ValidateUtils.isEmptyList(paramList)) {
-            // 当前无该维度的资源，则直接设置为
-            resultList.addAll(this.getNoResResult(clusterPhy.getId(), healthCheckTopicService, healthConfigMap));
-        }
-
-        // 遍历资源
-        for (ClusterPhyParam clusterPhyParam: paramList) {
-            resultList.addAll(this.checkAndGetResult(healthCheckTopicService, clusterPhyParam, healthConfigMap));
-        }
-
-        for (HealthCheckResult checkResult: resultList) {
-            try {
-                healthCheckResultService.replace(checkResult);
-            } catch (Exception e) {
-                log.error("class=TopicHealthCheckTask||method=processSubTask||clusterPhyId={}||checkResult={}||errMsg=exception!", clusterPhy.getId(), checkResult, e);
-            }
-        }
-
-        // 删除10分钟之前的检查结果
-        try {
-            healthCheckResultService.deleteByUpdateTimeBeforeInDB(clusterPhy.getId(), new Date(triggerTimeUnitMs - 10 * 60 * 1000));
-        } catch (Exception e) {
-            log.error("class=TopicHealthCheckTask||method=processSubTask||clusterPhyId={}||errMsg=exception!", clusterPhy.getId(), e);
-        }
-
-        return TaskResult.SUCCESS;
-    }
-
-    private List<HealthCheckResult> getNoResResult(Long clusterPhyId, AbstractHealthCheckService healthCheckService, Map<String, BaseClusterHealthConfig> healthConfigMap) {
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 进行检查
-        for (BaseClusterHealthConfig clusterHealthConfig: healthConfigMap.values()) {
-            HealthCheckDimensionEnum dimensionEnum = healthCheckService.getHealthCheckDimensionEnum();
-            if (!clusterHealthConfig.getCheckNameEnum().getDimensionEnum().equals(dimensionEnum)) {
-                // 类型不匹配
-                continue;
-            }
-
-            // 记录
-            HealthCheckResult checkResult = new HealthCheckResult(
-                    dimensionEnum.getDimension(),
-                    clusterHealthConfig.getCheckNameEnum().getConfigName(),
-                    clusterPhyId,
-                    "-1"
-            );
-            checkResult.setPassed(Constant.YES);
-            resultList.add(checkResult);
-        }
-
-        return resultList;
-    }
-
-    private List<HealthCheckResult> checkAndGetResult(AbstractHealthCheckService healthCheckService,
-                                                      ClusterPhyParam clusterPhyParam,
-                                                      Map<String, BaseClusterHealthConfig> healthConfigMap) {
-        List<HealthCheckResult> resultList = new ArrayList<>();
-
-        // 进行检查
-        for (BaseClusterHealthConfig clusterHealthConfig: healthConfigMap.values()) {
-            HealthCheckResult healthCheckResult = healthCheckService.checkAndGetResult(clusterPhyParam, clusterHealthConfig);
-            if (healthCheckResult == null) {
-                continue;
-            }
-
-            // 记录
-            resultList.add(healthCheckResult);
-        }
-
-        return resultList;
     }
 }