Skip to content

Commit 714e9a5

Browse files
zengqiaoZQKC
zengqiao
authored and
ZQKC
committed
[Optimize] 优化ZK指标的获取，减少重复采集的出现 (#709)
1、避免不同集群，相同的ZK地址时，指标重复获取的情况； 2、避免集群某个ZK地址获取指标失败时，下一个周期还会继续尝试从该地址获取指标；
1 parent 88d0a60 commit 714e9a5

File tree

2 files changed

+102
-32
lines changed

2 files changed

+102
-32
lines changed

‎km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/cache/ZookeeperLocalCache.java‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,46 @@
1+
package com.xiaojukeji.know.streaming.km.core.cache;
2+
3+
import com.github.benmanes.caffeine.cache.Cache;
4+
import com.github.benmanes.caffeine.cache.Caffeine;
5+
import com.xiaojukeji.know.streaming.km.common.bean.entity.zookeeper.fourletterword.BaseFourLetterWordCmdData;
6+
7+
import java.util.concurrent.TimeUnit;
8+
9+
public class ZookeeperLocalCache {
10+
private static final Cache<String, String> fourLetterCmdFailedServerCache = Caffeine.newBuilder()
11+
.expireAfterWrite(10, TimeUnit.MINUTES)
12+
.maximumSize(10000)
13+
.build();
14+
15+
private static final Cache<String, BaseFourLetterWordCmdData> fourLetterCmdDataCache = Caffeine.newBuilder()
16+
.expireAfterWrite(60, TimeUnit.SECONDS)
17+
.maximumSize(10000)
18+
.build();
19+
20+
public static boolean canUse(String host, int port, String cmd) {
21+
String data = fourLetterCmdFailedServerCache.getIfPresent(gen4lwFailedKey(host, port, cmd));
22+
23+
return data == null;
24+
}
25+
26+
public static void setFailed(String host, int port, String cmd) {
27+
fourLetterCmdFailedServerCache.put(gen4lwFailedKey(host, port, cmd), "");
28+
}
29+
30+
public static BaseFourLetterWordCmdData getData(String host, int port, String cmd) {
31+
return fourLetterCmdDataCache.getIfPresent(gen4lwFailedKey(host, port, cmd));
32+
}
33+
34+
public static void putData(String host, int port, String cmd, BaseFourLetterWordCmdData cmdData) {
35+
fourLetterCmdDataCache.put(gen4lwFailedKey(host, port, cmd), cmdData);
36+
}
37+
38+
/**************************************************** private method ****************************************************/
39+
40+
private static String gen4lwFailedKey(String host, int port, String cmd) {
41+
return host + "@" + port + "@" + cmd;
42+
}
43+
44+
45+
46+
}

‎km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/zookeeper/impl/ZookeeperMetricServiceImpl.java‎

Lines changed: 56 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
1111
import com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus;
1212
import com.xiaojukeji.know.streaming.km.common.bean.entity.version.VersionJmxInfo;
13+
import com.xiaojukeji.know.streaming.km.common.bean.entity.zookeeper.fourletterword.BaseFourLetterWordCmdData;
1314
import com.xiaojukeji.know.streaming.km.common.bean.entity.zookeeper.fourletterword.ServerCmdData;
1415
import com.xiaojukeji.know.streaming.km.common.bean.entity.zookeeper.fourletterword.parser.MonitorCmdDataParser;
1516
import com.xiaojukeji.know.streaming.km.common.bean.entity.zookeeper.fourletterword.parser.ServerCmdDataParser;
@@ -26,8 +27,8 @@
2627
import com.xiaojukeji.know.streaming.km.common.bean.entity.metrics.ZookeeperMetrics;
2728
import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.ZookeeperMetricPO;
2829
import com.xiaojukeji.know.streaming.km.common.utils.zookeeper.FourLetterWordUtil;
30+
import com.xiaojukeji.know.streaming.km.core.cache.ZookeeperLocalCache;
2931
import com.xiaojukeji.know.streaming.km.core.service.cluster.ClusterPhyService;
30-
import com.xiaojukeji.know.streaming.km.core.service.kafkacontroller.KafkaControllerService;
3132
import com.xiaojukeji.know.streaming.km.core.service.version.BaseMetricService;
3233
import com.xiaojukeji.know.streaming.km.core.service.zookeeper.ZookeeperMetricService;
3334
import com.xiaojukeji.know.streaming.km.core.service.zookeeper.ZookeeperService;
@@ -53,6 +54,7 @@ public class ZookeeperMetricServiceImpl extends BaseMetricService implements Zoo
5354
public static final String ZOOKEEPER_METHOD_GET_METRIC_FROM_MONITOR_CMD = "getMetricFromMonitorCmd";
5455
public static final String ZOOKEEPER_METHOD_GET_METRIC_FROM_SERVER_CMD = "getMetricFromServerCmd";
5556
public static final String ZOOKEEPER_METHOD_GET_METRIC_FROM_KAFKA_BY_JMX = "getMetricFromKafkaByJMX";
57+
public static final String ZOOKEEPER_METHOD_GET_METRIC_FROM_HEALTH_SERVICE = "getMetricFromHealthService";
5658

5759
@Autowired
5860
private ClusterPhyService clusterPhyService;
@@ -66,9 +68,6 @@ public class ZookeeperMetricServiceImpl extends BaseMetricService implements Zoo
6668
@Autowired
6769
private KafkaJMXClient kafkaJMXClient;
6870

69-
@Autowired
70-
private KafkaControllerService kafkaControllerService;
71-
7271
@Override
7372
protected VersionItemTypeEnum getVersionItemType() {
7473
return VersionItemTypeEnum.METRIC_ZOOKEEPER;
@@ -171,24 +170,37 @@ private Result<ZookeeperMetrics> getMetricFromServerCmd(VersionItemParam metricP
171170

172171
Result<ZookeeperMetrics> rz = null;
173172
for (Tuple<String, Integer> hostPort: param.getZkAddressList()) {
174-
Result<ServerCmdData> cmdDataResult = FourLetterWordUtil.executeFourLetterCmd(
175-
param.getClusterPhyId(),
176-
hostPort.getV1(),
177-
hostPort.getV2(),
178-
param.getZkConfig() != null ? param.getZkConfig().getOpenSecure(): false,
179-
param.getZkConfig() != null ? param.getZkConfig().getRequestTimeoutUnitMs(): Constant.DEFAULT_REQUEST_TIMEOUT_UNIT_MS,
180-
new ServerCmdDataParser()
181-
);
182-
183-
if (cmdDataResult.failed()) {
184-
rz = Result.buildFromIgnoreData(cmdDataResult);
173+
ServerCmdData cmdData = null;
174+
175+
BaseFourLetterWordCmdData baseCmdData = ZookeeperLocalCache.getData(hostPort.getV1(), hostPort.getV2(), FourLetterWordUtil.ServerCmd);
176+
if (baseCmdData != null) {
177+
cmdData = (ServerCmdData) baseCmdData;
178+
} else if (ZookeeperLocalCache.canUse(hostPort.getV1(), hostPort.getV2(), FourLetterWordUtil.ServerCmd)) {
179+
Result<ServerCmdData> cmdDataResult = FourLetterWordUtil.executeFourLetterCmd(
180+
param.getClusterPhyId(),
181+
hostPort.getV1(),
182+
hostPort.getV2(),
183+
param.getZkConfig() != null ? param.getZkConfig().getOpenSecure(): false,
184+
param.getZkConfig() != null ? param.getZkConfig().getRequestTimeoutUnitMs(): Constant.DEFAULT_REQUEST_TIMEOUT_UNIT_MS,
185+
new ServerCmdDataParser()
186+
);
187+
188+
if (cmdDataResult.failed()) {
189+
ZookeeperLocalCache.setFailed(hostPort.getV1(), hostPort.getV2(), FourLetterWordUtil.ServerCmd);
190+
191+
rz = Result.buildFromIgnoreData(cmdDataResult);
192+
continue;
193+
}
194+
195+
cmdData = cmdDataResult.getData();
196+
ZookeeperLocalCache.putData(hostPort.getV1(), hostPort.getV2(), FourLetterWordUtil.ServerCmd, cmdData);
197+
} else {
198+
// baseCmdData为空 且 当前地址不可使用
185199
continue;
186200
}
187201

188-
ServerCmdData cmdData = cmdDataResult.getData();
189-
190202
ZookeeperMetrics metrics = new ZookeeperMetrics(param.getClusterPhyId());
191-
metrics.putMetric(ZOOKEEPER_METRIC_AVG_REQUEST_LATENCY, cmdData.getZkAvgLatency().floatValue());
203+
metrics.putMetric(ZOOKEEPER_METRIC_AVG_REQUEST_LATENCY, cmdData.getZkAvgLatency());
192204
metrics.putMetric(ZOOKEEPER_METRIC_MIN_REQUEST_LATENCY, cmdData.getZkMinLatency().floatValue());
193205
metrics.putMetric(ZOOKEEPER_METRIC_MAX_REQUEST_LATENCY, cmdData.getZkMaxLatency().floatValue());
194206
metrics.putMetric(ZOOKEEPER_METRIC_OUTSTANDING_REQUESTS, cmdData.getZkOutstandingRequests().floatValue());
@@ -208,24 +220,36 @@ private Result<ZookeeperMetrics> getMetricFromMonitorCmd(VersionItemParam metric
208220

209221
Result<ZookeeperMetrics> rz = null;
210222
for (Tuple<String, Integer> hostPort: param.getZkAddressList()) {
211-
Result<MonitorCmdData> cmdDataResult = FourLetterWordUtil.executeFourLetterCmd(
212-
param.getClusterPhyId(),
213-
hostPort.getV1(),
214-
hostPort.getV2(),
215-
param.getZkConfig() != null ? param.getZkConfig().getOpenSecure(): false,
216-
param.getZkConfig() != null ? param.getZkConfig().getRequestTimeoutUnitMs(): Constant.DEFAULT_REQUEST_TIMEOUT_UNIT_MS,
217-
new MonitorCmdDataParser()
218-
);
219-
220-
if (cmdDataResult.failed()) {
221-
rz = Result.buildFromIgnoreData(cmdDataResult);
223+
MonitorCmdData cmdData = null;
224+
225+
BaseFourLetterWordCmdData baseCmdData = ZookeeperLocalCache.getData(hostPort.getV1(), hostPort.getV2(), FourLetterWordUtil.MonitorCmd);
226+
if (baseCmdData != null) {
227+
cmdData = (MonitorCmdData) baseCmdData;
228+
} else if (ZookeeperLocalCache.canUse(hostPort.getV1(), hostPort.getV2(), FourLetterWordUtil.MonitorCmd)) {
229+
Result<MonitorCmdData> cmdDataResult = FourLetterWordUtil.executeFourLetterCmd(
230+
param.getClusterPhyId(),
231+
hostPort.getV1(),
232+
hostPort.getV2(),
233+
param.getZkConfig() != null ? param.getZkConfig().getOpenSecure(): false,
234+
param.getZkConfig() != null ? param.getZkConfig().getRequestTimeoutUnitMs(): Constant.DEFAULT_REQUEST_TIMEOUT_UNIT_MS,
235+
new MonitorCmdDataParser()
236+
);
237+
238+
if (cmdDataResult.failed()) {
239+
ZookeeperLocalCache.setFailed(hostPort.getV1(), hostPort.getV2(), FourLetterWordUtil.MonitorCmd);
240+
241+
rz = Result.buildFromIgnoreData(cmdDataResult);
242+
continue;
243+
}
244+
245+
cmdData = cmdDataResult.getData();
246+
ZookeeperLocalCache.putData(hostPort.getV1(), hostPort.getV2(), FourLetterWordUtil.MonitorCmd, cmdData);
247+
} else {
222248
continue;
223249
}
224250

225-
MonitorCmdData cmdData = cmdDataResult.getData();
226-
227251
ZookeeperMetrics metrics = new ZookeeperMetrics(param.getClusterPhyId());
228-
metrics.putMetric(ZOOKEEPER_METRIC_AVG_REQUEST_LATENCY, cmdData.getZkAvgLatency().floatValue());
252+
metrics.putMetric(ZOOKEEPER_METRIC_AVG_REQUEST_LATENCY, cmdData.getZkAvgLatency());
229253
metrics.putMetric(ZOOKEEPER_METRIC_MIN_REQUEST_LATENCY, cmdData.getZkMinLatency().floatValue());
230254
metrics.putMetric(ZOOKEEPER_METRIC_MAX_REQUEST_LATENCY, cmdData.getZkMaxLatency().floatValue());
231255
metrics.putMetric(ZOOKEEPER_METRIC_OUTSTANDING_REQUESTS, cmdData.getZkOutstandingRequests().floatValue());

0 commit comments

Comments
 (0)