natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Enhance metrics (#12714) 

* Enhance metrics

* Update failed

* Enhance option

* rename

* enhance
This commit is contained in:
Albumen Kevin 2023-07-13 09:41:33 +08:00 committed by GitHub
parent d8a314ff33
commit 097463a83e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 115 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

53
dubbo-common/src/main/java/org/apache/dubbo/config/nested/AggregationConfig.java
View File

@ -25,6 +25,14 @@ public class AggregationConfig implements Serializable {
*/
private Boolean enabled;
private Boolean enableQps;
private Boolean enableRtPxx;
private Boolean enableRt;
private Boolean enableRequest;
/**
* Bucket num for time window quantile
*/
@ -35,6 +43,11 @@ public class AggregationConfig implements Serializable {
*/
private Integer timeWindowSeconds;
/**
* Time window mill seconds for qps
*/
private Integer qpsTimeWindowMillSeconds;
public Boolean getEnabled() {
return enabled;
}
@ -58,4 +71,44 @@ public class AggregationConfig implements Serializable {
public void setTimeWindowSeconds(Integer timeWindowSeconds) {
this.timeWindowSeconds = timeWindowSeconds;
}
public Boolean getEnableQps() {
return enableQps;
}
public void setEnableQps(Boolean enableQps) {
this.enableQps = enableQps;
}
public Boolean getEnableRtPxx() {
return enableRtPxx;
}
public void setEnableRtPxx(Boolean enableRtPxx) {
this.enableRtPxx = enableRtPxx;
}
public Boolean getEnableRt() {
return enableRt;
}
public void setEnableRt(Boolean enableRt) {
this.enableRt = enableRt;
}
public Boolean getEnableRequest() {
return enableRequest;
}
public void setEnableRequest(Boolean enableRequest) {
this.enableRequest = enableRequest;
}
public Integer getQpsTimeWindowMillSeconds() {
return qpsTimeWindowMillSeconds;
}
public void setQpsTimeWindowMillSeconds(Integer qpsTimeWindowMillSeconds) {
this.qpsTimeWindowMillSeconds = qpsTimeWindowMillSeconds;
}
}

4
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/aggregate/TimeWindowCounter.java
View File

@ -45,6 +45,10 @@ public class TimeWindowCounter {
return TimeUnit.MILLISECONDS.toSeconds(this.slidingWindow.values().size() * this.slidingWindow.getPaneIntervalInMs());
}
public long bucketLivedMillSeconds() {
return this.slidingWindow.getIntervalInMs() - (System.currentTimeMillis() - this.slidingWindow.currentPane().getEndInMs());
}
public void increment() {
this.increment(1L);
}

4
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/model/MetricsSupport.java
View File

@ -112,6 +112,8 @@ public class MetricsSupport {
if (e.isNetwork()) {
targetKey = MetricsKey.METRIC_REQUESTS_NETWORK_FAILED;
}
} else {
targetKey = MetricsKey.METRIC_REQUEST_BUSINESS_FAILED;
}
return targetKey;
}
@ -135,6 +137,8 @@ public class MetricsSupport {
if (e.isNetwork()) {
targetKey = MetricsKey.METRIC_REQUESTS_NETWORK_FAILED_AGG;
}
} else {
targetKey = MetricsKey.METRIC_REQUEST_BUSINESS_FAILED_AGG;
}
return targetKey;
}

74
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
View File

@ -56,8 +56,9 @@ import static org.apache.dubbo.metrics.model.MetricsCategory.RT;
* This collector only enabled when metrics aggregation config is enabled.
*/
public class AggregateMetricsCollector implements MetricsCollector<RequestEvent> {
private int bucketNum;
private int timeWindowSeconds;
private int bucketNum = DEFAULT_BUCKET_NUM;
private int timeWindowSeconds = DEFAULT_TIME_WINDOW_SECONDS;
private int qpsTimeWindowMillSeconds = DEFAULT_QPS_TIME_WINDOW_MILL_SECONDS;
private final Map<MetricsKeyWrapper, ConcurrentHashMap<MethodMetric, TimeWindowCounter>> methodTypeCounter = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowQuantile> rt = new ConcurrentHashMap<>();
private final ConcurrentHashMap<MethodMetric, TimeWindowCounter> qps = new ConcurrentHashMap<>();
@ -65,7 +66,13 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
private static final Integer DEFAULT_COMPRESSION = 100;
private static final Integer DEFAULT_BUCKET_NUM = 10;
private static final Integer DEFAULT_TIME_WINDOW_SECONDS = 120;
private static final Integer DEFAULT_QPS_TIME_WINDOW_MILL_SECONDS = 3000;
private Boolean collectEnabled = null;
private boolean enableQps;
private boolean enableRtPxx;
private boolean enableRt;
private boolean enableRequest;
private final ConcurrentMap<MethodMetric, TimeWindowAggregator> rtAgr = new ConcurrentHashMap<>();
@ -78,8 +85,15 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
if (optional.isPresent()) {
registerListener();
AggregationConfig aggregation = optional.get().getAggregation();
this.bucketNum = aggregation.getBucketNum() == null ? DEFAULT_BUCKET_NUM : aggregation.getBucketNum();
this.timeWindowSeconds = aggregation.getTimeWindowSeconds() == null ? DEFAULT_TIME_WINDOW_SECONDS : aggregation.getTimeWindowSeconds();
this.bucketNum = Optional.ofNullable(aggregation.getBucketNum()).orElse(DEFAULT_BUCKET_NUM);
this.timeWindowSeconds = Optional.ofNullable(aggregation.getTimeWindowSeconds())
.orElse(DEFAULT_TIME_WINDOW_SECONDS);
this.qpsTimeWindowMillSeconds = Optional.ofNullable(aggregation.getQpsTimeWindowMillSeconds())
.orElse(DEFAULT_QPS_TIME_WINDOW_MILL_SECONDS);
this.enableQps = Optional.ofNullable(aggregation.getEnableQps()).orElse(true);
this.enableRtPxx = Optional.ofNullable(aggregation.getEnableRtPxx()).orElse(true);
this.enableRt = Optional.ofNullable(aggregation.getEnableRt()).orElse(true);
this.enableRequest = Optional.ofNullable(aggregation.getEnableRequest()).orElse(true);
}
}
}
@ -89,7 +103,7 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
this.collectEnabled = collectEnabled;
}
}
@Override
public boolean isCollectEnabled() {
@ -107,9 +121,12 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
@Override
public void onEvent(RequestEvent event) {
MethodMetric metric = calcWindowCounter(event, MetricsKey.METRIC_REQUESTS);
TimeWindowCounter qpsCounter = ConcurrentHashMapUtils.computeIfAbsent(qps, metric, methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
qpsCounter.increment();
if (enableQps) {
MethodMetric metric = calcWindowCounter(event, MetricsKey.METRIC_REQUESTS);
TimeWindowCounter qpsCounter = ConcurrentHashMapUtils.computeIfAbsent(qps, metric,
methodMetric -> new TimeWindowCounter(bucketNum, qpsTimeWindowMillSeconds));
qpsCounter.increment();
}
}
@Override
@ -125,23 +142,33 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
@Override
public void onEventError(RequestEvent event) {
MetricsKey targetKey = MetricsKey.METRIC_REQUESTS_FAILED;
Object throwableObj = event.getAttachmentValue(METRIC_THROWABLE);
if (throwableObj != null) {
targetKey = MetricsSupport.getAggMetricsKey((Throwable) throwableObj);
if (enableRequest) {
MetricsKey targetKey = MetricsKey.METRIC_REQUESTS_FAILED;
Object throwableObj = event.getAttachmentValue(METRIC_THROWABLE);
if (throwableObj != null) {
targetKey = MetricsSupport.getAggMetricsKey((Throwable) throwableObj);
}
calcWindowCounter(event, targetKey);
}
if (enableRt || enableRtPxx) {
onRTEvent(event);
}
calcWindowCounter(event, targetKey);
onRTEvent(event);
}
private void onRTEvent(RequestEvent event) {
MethodMetric metric = new MethodMetric(applicationModel, event.getAttachmentValue(MetricsConstants.INVOCATION));
long responseTime = event.getTimePair().calc();
TimeWindowQuantile quantile = ConcurrentHashMapUtils.computeIfAbsent(rt, metric, k -> new TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds));
quantile.add(responseTime);
if (enableRt) {
TimeWindowQuantile quantile = ConcurrentHashMapUtils.computeIfAbsent(rt, metric,
k -> new TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds));
quantile.add(responseTime);
}
TimeWindowAggregator timeWindowAggregator = ConcurrentHashMapUtils.computeIfAbsent(rtAgr, metric, methodMetric -> new TimeWindowAggregator(bucketNum, timeWindowSeconds));
timeWindowAggregator.add(responseTime);
if (enableRtPxx) {
TimeWindowAggregator timeWindowAggregator = ConcurrentHashMapUtils.computeIfAbsent(rtAgr, metric,
methodMetric -> new TimeWindowAggregator(bucketNum, timeWindowSeconds));
timeWindowAggregator.add(responseTime);
}
}
@ -152,7 +179,8 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
ConcurrentMap<MethodMetric, TimeWindowCounter> counter = methodTypeCounter.computeIfAbsent(metricsKeyWrapper, k -> new ConcurrentHashMap<>());
TimeWindowCounter windowCounter = ConcurrentHashMapUtils.computeIfAbsent(counter, metric, methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
TimeWindowCounter windowCounter = ConcurrentHashMapUtils.computeIfAbsent(counter, metric,
methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
windowCounter.increment();
return metric;
}
@ -196,7 +224,11 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
private void collectQPS(List<MetricSample> list) {
qps.forEach((k, v) -> list.add(new GaugeMetricSample<>(MetricsKey.METRIC_QPS.getNameByType(k.getSide()),
MetricsKey.METRIC_QPS.getDescription(), k.getTags(), QPS, v, value -> value.get() / value.bucketLivedSeconds())));
MetricsKey.METRIC_QPS.getDescription(), k.getTags(), QPS, v, value -> {
double total = value.get();
long millSeconds = value.bucketLivedMillSeconds();
return total / millSeconds * 1000;
})));
}
private void collectRT(List<MetricSample> list) {
@ -211,7 +243,7 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
MetricsKey.METRIC_RT_P50.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.50)));
});
rtAgr.forEach((k,v)->{
rtAgr.forEach((k, v) -> {
list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_MIN_AGG.getNameByType(k.getSide()),
MetricsKey.METRIC_RT_MIN_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getMin()));

1
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/event/DefaultSubDispatcher.java
View File

@ -82,6 +82,7 @@ public final class DefaultSubDispatcher extends SimpleMetricsEventMulticaster {
targetKey = key;
} else {
targetKey = MetricsSupport.getMetricsKey((Throwable) throwableObj);
MetricsSupport.increment(MetricsKey.METRIC_REQUESTS_TOTAL_FAILED, dynamicPlaceType, (MethodMetricsCollector) collector, event);
}
MetricsSupport.incrAndAddRt(targetKey, dynamicPlaceType, (MethodMetricsCollector) collector, event);
})),