8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2025-12-24 03:42:07 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

[Optimize]无ZK模块时，巡检详情忽略对ZK的展示(#764)

Browse Source
This commit is contained in:
zengqiao
2023-01-03 13:16:54 +08:00
committed by EricZeng
parent c187b5246f
commit 4994639111
12 changed files with 93 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checker/AbstractHealthCheckService.java
View File

@@ -28,6 +28,8 @@ public abstract class AbstractHealthCheckService {
public abstract HealthCheckDimensionEnum getHealthCheckDimensionEnum();
public abstract Integer getDimensionCodeIfSupport(Long kafkaClusterPhyId);
public HealthCheckResult checkAndGetResult(ClusterParam clusterParam, BaseClusterHealthConfig clusterHealthConfig) {
if (ValidateUtils.anyNull(clusterParam, clusterHealthConfig)) {
return null;

5
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checker/broker/HealthCheckBrokerService.java
View File

@@ -57,6 +57,11 @@ public class HealthCheckBrokerService extends AbstractHealthCheckService {
return HealthCheckDimensionEnum.BROKER;
}
@Override
public Integer getDimensionCodeIfSupport(Long kafkaClusterPhyId) {
return this.getHealthCheckDimensionEnum().getDimension();
}
/**
* Broker网络处理线程平均值过低
*/

5
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checker/cluster/HealthCheckClusterService.java
View File

@@ -44,6 +44,11 @@ public class HealthCheckClusterService extends AbstractHealthCheckService {
return HealthCheckDimensionEnum.CLUSTER;
}
@Override
public Integer getDimensionCodeIfSupport(Long kafkaClusterPhyId) {
return this.getHealthCheckDimensionEnum().getDimension();
}
/**
* 检查NoController
*/

16
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checker/connect/HealthCheckConnectClusterService.java
View File

@@ -14,7 +14,9 @@ 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.enums.health.HealthCheckNameEnum;
import com.xiaojukeji.know.streaming.km.common.utils.Tuple;
import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
import com.xiaojukeji.know.streaming.km.core.service.connect.cluster.ConnectClusterMetricService;
import com.xiaojukeji.know.streaming.km.core.service.connect.cluster.ConnectClusterService;
import com.xiaojukeji.know.streaming.km.core.service.health.checker.AbstractHealthCheckService;
import com.xiaojukeji.know.streaming.km.persistence.connect.cache.LoadedConnectClusterCache;
import org.springframework.beans.factory.annotation.Autowired;
@@ -32,9 +34,11 @@ import static com.xiaojukeji.know.streaming.km.core.service.version.metrics.conn
*/
@Service
public class HealthCheckConnectClusterService extends AbstractHealthCheckService {
private static final ILog log = LogFactory.getLog(HealthCheckConnectClusterService.class);
@Autowired
private ConnectClusterService connectClusterService;
@Autowired
private ConnectClusterMetricService connectClusterMetricService;
@@ -57,6 +61,16 @@ public class HealthCheckConnectClusterService extends AbstractHealthCheckService
return HealthCheckDimensionEnum.CONNECT_CLUSTER;
}
@Override
public Integer getDimensionCodeIfSupport(Long kafkaClusterPhyId) {
List<ConnectCluster> clusterList = connectClusterService.listByKafkaCluster(kafkaClusterPhyId);
if (ValidateUtils.isEmptyList(clusterList)) {
return null;
}
return this.getHealthCheckDimensionEnum().getDimension();
}
private HealthCheckResult checkStartupFailurePercentage(Tuple<ClusterParam, BaseClusterHealthConfig> paramTuple) {
ConnectClusterParam param = (ConnectClusterParam) paramTuple.getV1();
HealthCompareValueConfig compareConfig = (HealthCompareValueConfig) paramTuple.getV2();

18
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checker/connect/HealthCheckConnectorService.java
View File

@@ -4,6 +4,7 @@ import com.didiglobal.logi.log.ILog;
import com.didiglobal.logi.log.LogFactory;
import com.xiaojukeji.know.streaming.km.common.bean.entity.config.healthcheck.BaseClusterHealthConfig;
import com.xiaojukeji.know.streaming.km.common.bean.entity.config.healthcheck.HealthCompareValueConfig;
import com.xiaojukeji.know.streaming.km.common.bean.entity.connect.ConnectCluster;
import com.xiaojukeji.know.streaming.km.common.bean.entity.health.HealthCheckResult;
import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.connect.ConnectorMetrics;
import com.xiaojukeji.know.streaming.km.common.bean.entity.param.cluster.ClusterParam;
@@ -13,6 +14,8 @@ 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.enums.health.HealthCheckNameEnum;
import com.xiaojukeji.know.streaming.km.common.utils.Tuple;
import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
import com.xiaojukeji.know.streaming.km.core.service.connect.cluster.ConnectClusterService;
import com.xiaojukeji.know.streaming.km.core.service.connect.connector.ConnectorMetricService;
import com.xiaojukeji.know.streaming.km.core.service.connect.connector.ConnectorService;
import com.xiaojukeji.know.streaming.km.core.service.health.checker.AbstractHealthCheckService;
@@ -32,11 +35,14 @@ import static com.xiaojukeji.know.streaming.km.core.service.version.metrics.conn
*/
@Service
public class HealthCheckConnectorService extends AbstractHealthCheckService {
private static final ILog log = LogFactory.getLog(HealthCheckConnectorService.class);
@Autowired
private ConnectorService connectorService;
@Autowired
private ConnectClusterService connectClusterService;
@Autowired
private ConnectorMetricService connectorMetricService;
@@ -66,6 +72,16 @@ public class HealthCheckConnectorService extends AbstractHealthCheckService {
return HealthCheckDimensionEnum.CONNECTOR;
}
@Override
public Integer getDimensionCodeIfSupport(Long kafkaClusterPhyId) {
List<ConnectCluster> clusterList = connectClusterService.listByKafkaCluster(kafkaClusterPhyId);
if (ValidateUtils.isEmptyList(clusterList)) {
return null;
}
return this.getHealthCheckDimensionEnum().getDimension();
}
private HealthCheckResult checkFailedTaskCount(Tuple<ClusterParam, BaseClusterHealthConfig> paramTuple) {
ConnectorParam param = (ConnectorParam) paramTuple.getV1();
HealthCompareValueConfig compareConfig = (HealthCompareValueConfig) paramTuple.getV2();

5
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checker/group/HealthCheckGroupService.java
View File

@@ -51,6 +51,11 @@ public class HealthCheckGroupService extends AbstractHealthCheckService {
return HealthCheckDimensionEnum.GROUP;
}
@Override
public Integer getDimensionCodeIfSupport(Long kafkaClusterPhyId) {
return this.getHealthCheckDimensionEnum().getDimension();
}
/**
* 检查Group re-balance太频繁
*/

5
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checker/topic/HealthCheckTopicService.java
View File

@@ -62,6 +62,11 @@ public class HealthCheckTopicService extends AbstractHealthCheckService {
return HealthCheckDimensionEnum.TOPIC;
}
@Override
public Integer getDimensionCodeIfSupport(Long kafkaClusterPhyId) {
return this.getHealthCheckDimensionEnum().getDimension();
}
/**
* 检查Topic长期未同步
*/

12
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/checker/zookeeper/HealthCheckZookeeperService.java
View File

@@ -19,6 +19,7 @@ import com.xiaojukeji.know.streaming.km.common.enums.health.HealthCheckNameEnum;
import com.xiaojukeji.know.streaming.km.common.enums.zookeeper.ZKRoleEnum;
import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
import com.xiaojukeji.know.streaming.km.common.utils.Tuple;
import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
import com.xiaojukeji.know.streaming.km.common.utils.zookeeper.ZookeeperUtils;
import com.xiaojukeji.know.streaming.km.core.service.health.checker.AbstractHealthCheckService;
import com.xiaojukeji.know.streaming.km.core.service.version.metrics.kafka.ZookeeperMetricVersionItems;
@@ -56,7 +57,7 @@ public class HealthCheckZookeeperService extends AbstractHealthCheckService {
@Override
public List<ClusterParam> getResList(Long clusterPhyId) {
ClusterPhy clusterPhy = LoadedClusterPhyCache.getByPhyId(clusterPhyId);
if (clusterPhy == null) {
if (clusterPhy == null || ValidateUtils.isBlank(clusterPhy.getZookeeper())) {
return new ArrayList<>();
}
@@ -80,6 +81,15 @@ public class HealthCheckZookeeperService extends AbstractHealthCheckService {
return HealthCheckDimensionEnum.ZOOKEEPER;
}
@Override
public Integer getDimensionCodeIfSupport(Long kafkaClusterPhyId) {
if (ValidateUtils.isEmptyList(this.getResList(kafkaClusterPhyId))) {
return null;
}
return this.getHealthCheckDimensionEnum().getDimension();
}
private HealthCheckResult checkBrainSplit(Tuple<ClusterParam, BaseClusterHealthConfig> singleConfigSimpleTuple) {
ZookeeperParam param = (ZookeeperParam) singleConfigSimpleTuple.getV1();

4
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/state/HealthStateService.java
View File

@@ -3,7 +3,6 @@ package com.xiaojukeji.know.streaming.km.core.service.health.state;
import com.xiaojukeji.know.streaming.km.common.bean.entity.health.HealthScoreResult;
import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.*;
import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.connect.ConnectorMetrics;
import com.xiaojukeji.know.streaming.km.common.enums.health.HealthCheckDimensionEnum;
import java.util.List;
@@ -22,8 +21,7 @@ public interface HealthStateService {
/**
* 获取集群健康检查结果
*/
List<HealthScoreResult> getClusterHealthResult(Long clusterPhyId);
List<HealthScoreResult> getDimensionHealthResult(Long clusterPhyId, HealthCheckDimensionEnum dimensionEnum);
List<HealthScoreResult> getAllDimensionHealthResult(Long clusterPhyId);
List<HealthScoreResult> getDimensionHealthResult(Long clusterPhyId, List<Integer> dimensionCodeList);
List<HealthScoreResult> getResHealthResult(Long clusterPhyId, Long clusterId, Integer dimension, String resNme);
}

35
km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/health/state/impl/HealthStateServiceImpl.java
View File

@@ -8,12 +8,14 @@ import com.xiaojukeji.know.streaming.km.common.bean.entity.health.HealthScoreRes
import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.*;
import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.connect.ConnectorMetrics;
import com.xiaojukeji.know.streaming.km.common.bean.po.health.HealthCheckResultPO;
import com.xiaojukeji.know.streaming.km.common.component.SpringTool;
import com.xiaojukeji.know.streaming.km.common.enums.health.HealthCheckDimensionEnum;
import com.xiaojukeji.know.streaming.km.common.enums.health.HealthCheckNameEnum;
import com.xiaojukeji.know.streaming.km.common.enums.health.HealthStateEnum;
import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerService;
import com.xiaojukeji.know.streaming.km.core.service.connect.cluster.ConnectClusterService;
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.core.service.health.state.HealthStateService;
import com.xiaojukeji.know.streaming.km.core.service.zookeeper.ZookeeperService;
@@ -224,17 +226,20 @@ public class HealthStateServiceImpl implements HealthStateService {
}
@Override
public List<HealthScoreResult> getClusterHealthResult(Long clusterPhyId) {
List<HealthCheckResultPO> poList = healthCheckResultService.listCheckResult(clusterPhyId);
public List<HealthScoreResult> getAllDimensionHealthResult(Long clusterPhyId) {
List<Integer> supportedDimensionCodeList = new ArrayList<>();
return this.convert2HealthScoreResultList(clusterPhyId, poList, null);
}
// 获取支持的code
for (AbstractHealthCheckService service: SpringTool.getBeansOfType(AbstractHealthCheckService.class).values()) {
Integer dimensionCode = service.getDimensionCodeIfSupport(clusterPhyId);
if (dimensionCode == null) {
continue;
}
@Override
public List<HealthScoreResult> getDimensionHealthResult(Long clusterPhyId, HealthCheckDimensionEnum dimensionEnum) {
List<HealthCheckResultPO> poList = healthCheckResultService.listCheckResult(clusterPhyId, dimensionEnum.getDimension());
supportedDimensionCodeList.add(dimensionCode);
}
return this.convert2HealthScoreResultList(clusterPhyId, poList, dimensionEnum.getDimension());
return this.getDimensionHealthResult(clusterPhyId, supportedDimensionCodeList);
}
@Override
@@ -242,22 +247,14 @@ public class HealthStateServiceImpl implements HealthStateService {
//查找健康巡查结果
List<HealthCheckResultPO> poList = new ArrayList<>();
for (Integer dimensionCode : dimensionCodeList) {
HealthCheckDimensionEnum dimensionEnum = HealthCheckDimensionEnum.getByCode(dimensionCode);
if (dimensionEnum.equals(HealthCheckDimensionEnum.UNKNOWN)) {
continue;
}
if (dimensionEnum.equals(HealthCheckDimensionEnum.CONNECTOR)) {
if (dimensionCode.equals(HealthCheckDimensionEnum.CONNECTOR.getDimension())) {
poList.addAll(healthCheckResultService.getConnectorHealthCheckResult(clusterPhyId));
} else {
poList.addAll(healthCheckResultService.listCheckResult(clusterPhyId, dimensionEnum.getDimension()));
poList.addAll(healthCheckResultService.listCheckResult(clusterPhyId, dimensionCode));
}
}
List<HealthScoreResult> resultList = this.getResHealthResult(clusterPhyId, dimensionCodeList, poList);
return resultList;
return this.getResHealthResult(clusterPhyId, dimensionCodeList, poList);
}
@Override