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

Add metrics version compare (#13104)

This commit is contained in:
Albumen Kevin 2023-09-30 10:06:21 +08:00 committed by GitHub
parent 9dc5bf1944
commit 70600eb632
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
26 changed files with 316 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/MetricsCollector.java
View File

@ -43,6 +43,14 @@ public interface MetricsCollector<E extends TimeCounterEvent> extends MetricsLif
*/
List<MetricSample> collect();
/**
* Check if samples have been changed.
* Note that this method will reset the changed flag to false using CAS.
*
* @return true if samples have been changed
*/
boolean calSamplesChanged();
default void initMetrics(MetricsEvent event) {};
}

14
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/ApplicationStatComposite.java
View File

@ -30,6 +30,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
/**
@ -46,11 +47,16 @@ public class ApplicationStatComposite extends AbstractMetricsExport {
private final Map<MetricsKey, AtomicLong> applicationNumStats = new ConcurrentHashMap<>();
private final AtomicBoolean samplesChanged = new AtomicBoolean(true);
public void init(List<MetricsKey> appKeys) {
if (CollectionUtils.isEmpty(appKeys)) {
return;
}
appKeys.forEach(appKey -> applicationNumStats.put(appKey, new AtomicLong(0L)));
appKeys.forEach(appKey -> {
applicationNumStats.put(appKey, new AtomicLong(0L));
});
samplesChanged.set(true);
}
public void incrementSize(MetricsKey metricsKey, int size) {
@ -78,4 +84,10 @@ public class ApplicationStatComposite extends AbstractMetricsExport {
return applicationNumStats;
}
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
return samplesChanged.compareAndSet(true, false);
}
}

10
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/BaseStatComposite.java
View File

@ -133,4 +133,14 @@ public abstract class BaseStatComposite implements MetricsExport {
public RtStatComposite getRtStatComposite() {
return rtStatComposite;
}
@Override
public boolean calSamplesChanged() {
// Should ensure that all the composite's samplesChanged have been compareAndSet, and cannot flip the `or` logic
boolean changed = applicationStatComposite.calSamplesChanged();
changed = rtStatComposite.calSamplesChanged() || changed;
changed = serviceStatComposite.calSamplesChanged() || changed;
changed = methodStatComposite.calSamplesChanged() || changed;
return changed;
}
}

17
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/MethodStatComposite.java
View File

@ -33,6 +33,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
/**
@ -43,6 +44,8 @@ import java.util.concurrent.atomic.AtomicLong;
public class MethodStatComposite extends AbstractMetricsExport {
private boolean serviceLevel;
private final AtomicBoolean samplesChanged = new AtomicBoolean(true);
public MethodStatComposite(ApplicationModel applicationModel) {
super(applicationModel);
this.serviceLevel = MethodMetric.isServiceLevel(getApplicationModel());
@ -54,7 +57,10 @@ public class MethodStatComposite extends AbstractMetricsExport {
if (CollectionUtils.isEmpty(metricsKeyWrappers)) {
return;
}
metricsKeyWrappers.forEach(appKey -> methodNumStats.put(appKey, new ConcurrentHashMap<>()));
metricsKeyWrappers.forEach(appKey -> {
methodNumStats.put(appKey, new ConcurrentHashMap<>());
});
samplesChanged.set(true);
}
public void initMethodKey(MetricsKeyWrapper wrapper, Invocation invocation) {
@ -63,6 +69,7 @@ public class MethodStatComposite extends AbstractMetricsExport {
}
methodNumStats.get(wrapper).computeIfAbsent(new MethodMetric(getApplicationModel(), invocation, serviceLevel), k -> new AtomicLong(0L));
samplesChanged.set(true);
}
public void incrementMethodKey(MetricsKeyWrapper wrapper, MethodMetric methodMetric, int size) {
@ -71,7 +78,8 @@ public class MethodStatComposite extends AbstractMetricsExport {
}
AtomicLong stat = methodNumStats.get(wrapper).get(methodMetric);
if (stat == null) {
methodNumStats.get(wrapper).putIfAbsent(methodMetric, new AtomicLong(0L));
methodNumStats.get(wrapper).computeIfAbsent(methodMetric, (k)-> new AtomicLong(0L));
samplesChanged.set(true);
stat = methodNumStats.get(wrapper).get(methodMetric);
}
stat.getAndAdd(size);
@ -97,4 +105,9 @@ public class MethodStatComposite extends AbstractMetricsExport {
return list;
}
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
return samplesChanged.compareAndSet(true, false);
}
}

18
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/RtStatComposite.java
View File

@ -38,6 +38,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAccumulator;
import java.util.function.BiConsumer;
@ -52,6 +53,8 @@ import java.util.stream.Collectors;
public class RtStatComposite extends AbstractMetricsExport {
private boolean serviceLevel;
private final AtomicBoolean samplesChanged = new AtomicBoolean(true);
public RtStatComposite(ApplicationModel applicationModel) {
super(applicationModel);
this.serviceLevel = MethodMetric.isServiceLevel(getApplicationModel());
@ -66,10 +69,10 @@ public class RtStatComposite extends AbstractMetricsExport {
for (MetricsPlaceValue placeValue : placeValues) {
List<LongContainer<? extends Number>> containers = initStats(placeValue);
for (LongContainer<? extends Number> container : containers) {
rtStats.computeIfAbsent(container.getMetricsKeyWrapper().getType(), k -> new ArrayList<>())
.add(container);
rtStats.computeIfAbsent(container.getMetricsKeyWrapper().getType(), k -> new ArrayList<>()).add(container);
}
}
samplesChanged.set(true);
}
private List<LongContainer<? extends Number>> initStats(MetricsPlaceValue placeValue) {
@ -95,6 +98,7 @@ public class RtStatComposite extends AbstractMetricsExport {
Number current = (Number) container.get(key);
if (current == null) {
container.putIfAbsent(key, container.getInitFunc().apply(key));
samplesChanged.set(true);
current = (Number) container.get(key);
}
container.getConsumerFunc().accept(responseTime, current);
@ -114,6 +118,7 @@ public class RtStatComposite extends AbstractMetricsExport {
if (actions == null) {
actions = calServiceRtActions(invocation, registryOpType);
cache.putIfAbsent(registryOpType, actions);
samplesChanged.set(true);
actions = cache.get(registryOpType);
}
} else {
@ -134,6 +139,7 @@ public class RtStatComposite extends AbstractMetricsExport {
Number current = (Number) container.get(key);
if (current == null) {
container.putIfAbsent(key, container.getInitFunc().apply(key));
samplesChanged.set(true);
current = (Number) container.get(key);
}
actions.add(new Action(container.getConsumerFunc(), current));
@ -155,6 +161,7 @@ public class RtStatComposite extends AbstractMetricsExport {
if (actions == null) {
actions = calMethodRtActions(invocation, registryOpType);
cache.putIfAbsent(registryOpType, actions);
samplesChanged.set(true);
actions = cache.get(registryOpType);
}
} else {
@ -174,6 +181,7 @@ public class RtStatComposite extends AbstractMetricsExport {
Number current = (Number) container.get(key);
if (current == null) {
container.putIfAbsent(key, container.getInitFunc().apply(key));
samplesChanged.set(true);
current = (Number) container.get(key);
}
actions.add(new Action(container.getConsumerFunc(), current));
@ -218,4 +226,10 @@ public class RtStatComposite extends AbstractMetricsExport {
consumerFunc.accept(responseTime, initValue);
}
}
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
return samplesChanged.compareAndSet(true, false);
}
}

29
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/ServiceStatComposite.java
View File

@ -30,6 +30,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
/**
@ -39,6 +40,8 @@ import java.util.concurrent.atomic.AtomicLong;
*/
public class ServiceStatComposite extends AbstractMetricsExport {
private final AtomicBoolean samplesChanged = new AtomicBoolean(true);
public ServiceStatComposite(ApplicationModel applicationModel) {
super(applicationModel);
}
@ -49,7 +52,10 @@ public class ServiceStatComposite extends AbstractMetricsExport {
if (CollectionUtils.isEmpty(metricsKeyWrappers)) {
return;
}
metricsKeyWrappers.forEach(appKey -> serviceWrapperNumStats.put(appKey, new ConcurrentHashMap<>()));
metricsKeyWrappers.forEach(appKey -> {
serviceWrapperNumStats.put(appKey, new ConcurrentHashMap<>());
});
samplesChanged.set(true);
}
public void incrementServiceKey(MetricsKeyWrapper wrapper, String serviceKey, int size) {
@ -64,7 +70,13 @@ public class ServiceStatComposite extends AbstractMetricsExport {
if (extra != null) {
serviceKeyMetric.setExtraInfo(extra);
}
serviceWrapperNumStats.get(wrapper).computeIfAbsent(serviceKeyMetric, k -> new AtomicLong(0L)).getAndAdd(size);
Map<ServiceKeyMetric, AtomicLong> map = serviceWrapperNumStats.get(wrapper);
AtomicLong metrics = map.get(serviceKeyMetric);
if (metrics == null) {
metrics = map.computeIfAbsent(serviceKeyMetric, k -> new AtomicLong(0L));
samplesChanged.set(true);
}
metrics.getAndAdd(size);
// MetricsSupport.fillZero(serviceWrapperNumStats);
}
@ -80,7 +92,13 @@ public class ServiceStatComposite extends AbstractMetricsExport {
if (extra != null) {
serviceKeyMetric.setExtraInfo(extra);
}
serviceWrapperNumStats.get(wrapper).computeIfAbsent(serviceKeyMetric, k -> new AtomicLong(0L)).set(num);
Map<ServiceKeyMetric, AtomicLong> stats = serviceWrapperNumStats.get(wrapper);
AtomicLong metrics = stats.get(serviceKeyMetric);
if (metrics == null) {
metrics = stats.computeIfAbsent(serviceKeyMetric, k -> new AtomicLong(0L));
samplesChanged.set(true);
}
metrics.set(num);
}
@Override
@ -95,4 +113,9 @@ public class ServiceStatComposite extends AbstractMetricsExport {
return list;
}
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
return samplesChanged.compareAndSet(true, false);
}
}

7
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/report/MetricsExport.java
View File

@ -33,4 +33,11 @@ public interface MetricsExport {
*/
List<MetricSample> export(MetricsCategory category);
/**
* Check if samples have been changed.
* Note that this method will reset the changed flag to false using CAS.
*
* @return true if samples have been changed
*/
boolean calSamplesChanged();
}

4
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/report/MetricsReporter.java
View File

@ -28,10 +28,10 @@ public interface MetricsReporter {
*/
void init();
void refreshData();
void resetIfSamplesChanged();
String getResponse();
default String getResponseWithName(String metricsName) {
return null;
}

15
dubbo-metrics/dubbo-metrics-config-center/src/main/java/org/apache/dubbo/metrics/config/collector/ConfigCenterMetricsCollector.java
View File

@ -34,6 +34,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import static org.apache.dubbo.metrics.model.MetricsCategory.CONFIGCENTER;
@ -47,6 +48,7 @@ public class ConfigCenterMetricsCollector extends CombMetricsCollector<ConfigCen
private Boolean collectEnabled = null;
private final ApplicationModel applicationModel;
private final AtomicBoolean samplesChanged = new AtomicBoolean(true);
private final Map<ConfigCenterMetric, AtomicLong> updatedMetrics = new ConcurrentHashMap<>();
@ -76,7 +78,12 @@ public class ConfigCenterMetricsCollector extends CombMetricsCollector<ConfigCen
return;
}
ConfigCenterMetric metric = new ConfigCenterMetric(applicationModel.getApplicationName(), key, group, protocol, changeTypeName);
updatedMetrics.computeIfAbsent(metric, k -> new AtomicLong(0L)).addAndGet(size);
AtomicLong metrics = updatedMetrics.get(metric);
if (metrics == null) {
metrics = updatedMetrics.computeIfAbsent(metric, k -> new AtomicLong(0L));
samplesChanged.set(true);
}
metrics.addAndGet(size);
}
@ -91,5 +98,9 @@ public class ConfigCenterMetricsCollector extends CombMetricsCollector<ConfigCen
return list;
}
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
return samplesChanged.compareAndSet(true, false);
}
}

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

@ -44,6 +44,7 @@ import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SIDE;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_SIDE;
@ -74,6 +75,7 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
private boolean enableRtPxx;
private boolean enableRt;
private boolean enableRequest;
private final AtomicBoolean samplesChanged = new AtomicBoolean(true);
private final ConcurrentMap<MethodMetric, TimeWindowAggregator> rtAgr = new ConcurrentHashMap<>();
@ -90,9 +92,9 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
AggregationConfig aggregation = optional.get().getAggregation();
this.bucketNum = Optional.ofNullable(aggregation.getBucketNum()).orElse(DEFAULT_BUCKET_NUM);
this.timeWindowSeconds = Optional.ofNullable(aggregation.getTimeWindowSeconds())
.orElse(DEFAULT_TIME_WINDOW_SECONDS);
.orElse(DEFAULT_TIME_WINDOW_SECONDS);
this.qpsTimeWindowMillSeconds = Optional.ofNullable(aggregation.getQpsTimeWindowMillSeconds())
.orElse(DEFAULT_QPS_TIME_WINDOW_MILL_SECONDS);
.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);
@ -127,8 +129,12 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
public void onEvent(RequestEvent event) {
if (enableQps) {
MethodMetric metric = calcWindowCounter(event, MetricsKey.METRIC_REQUESTS);
TimeWindowCounter qpsCounter = ConcurrentHashMapUtils.computeIfAbsent(qps, metric,
methodMetric -> new TimeWindowCounter(bucketNum, TimeUnit.MILLISECONDS.toSeconds(qpsTimeWindowMillSeconds)));
TimeWindowCounter qpsCounter = qps.get(metric);
if (qpsCounter == null) {
qpsCounter = ConcurrentHashMapUtils.computeIfAbsent(qps, metric,
methodMetric -> new TimeWindowCounter(bucketNum, TimeUnit.MILLISECONDS.toSeconds(qpsTimeWindowMillSeconds)));
samplesChanged.set(true);
}
qpsCounter.increment();
}
}
@ -163,14 +169,22 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
MethodMetric metric = new MethodMetric(applicationModel, event.getAttachmentValue(MetricsConstants.INVOCATION), serviceLevel);
long responseTime = event.getTimePair().calc();
if (enableRt) {
TimeWindowQuantile quantile = ConcurrentHashMapUtils.computeIfAbsent(rt, metric,
k -> new TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds));
TimeWindowQuantile quantile = rt.get(metric);
if (quantile == null) {
quantile = ConcurrentHashMapUtils.computeIfAbsent(rt, metric,
k -> new TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds));
samplesChanged.set(true);
}
quantile.add(responseTime);
}
if (enableRtPxx) {
TimeWindowAggregator timeWindowAggregator = ConcurrentHashMapUtils.computeIfAbsent(rtAgr, metric,
methodMetric -> new TimeWindowAggregator(bucketNum, timeWindowSeconds));
TimeWindowAggregator timeWindowAggregator = rtAgr.get(metric);
if (timeWindowAggregator == null) {
timeWindowAggregator = ConcurrentHashMapUtils.computeIfAbsent(rtAgr, metric,
methodMetric -> new TimeWindowAggregator(bucketNum, timeWindowSeconds));
samplesChanged.set(true);
}
timeWindowAggregator.add(responseTime);
}
}
@ -183,8 +197,12 @@ 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 = counter.get(metric);
if (windowCounter == null) {
windowCounter = ConcurrentHashMapUtils.computeIfAbsent(counter, metric,
methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
samplesChanged.set(true);
}
windowCounter.increment();
return metric;
}
@ -225,13 +243,13 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
ConcurrentHashMap<MethodMetric, TimeWindowCounter> windowCounter = methodTypeCounter.get(metricsKeyWrapper);
if (windowCounter != null) {
windowCounter.forEach((k, v) -> list.add(new GaugeMetricSample<>(metricsKey.getNameByType(k.getSide()),
metricsKey.getDescription(), k.getTags(), REQUESTS, v, TimeWindowCounter::get)));
metricsKey.getDescription(), k.getTags(), REQUESTS, v, TimeWindowCounter::get)));
}
}
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 -> {
MetricsKey.METRIC_QPS.getDescription(), k.getTags(), QPS, v, value -> {
double total = value.get();
long millSeconds = value.bucketLivedMillSeconds();
return total / millSeconds * 1000;
@ -241,24 +259,24 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
private void collectRT(List<MetricSample> list) {
rt.forEach((k, v) -> {
list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_P99.getNameByType(k.getSide()),
MetricsKey.METRIC_RT_P99.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.99)));
MetricsKey.METRIC_RT_P99.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.99)));
list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_P95.getNameByType(k.getSide()),
MetricsKey.METRIC_RT_P95.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.95)));
MetricsKey.METRIC_RT_P95.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.95)));
list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_P90.getNameByType(k.getSide()),
MetricsKey.METRIC_RT_P90.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.90)));
MetricsKey.METRIC_RT_P90.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.90)));
list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_P50.getNameByType(k.getSide()),
MetricsKey.METRIC_RT_P50.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.50)));
MetricsKey.METRIC_RT_P50.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.50)));
});
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()));
MetricsKey.METRIC_RT_MIN_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getMin()));
list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_MAX_AGG.getNameByType(k.getSide()),
MetricsKey.METRIC_RT_MAX_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getMax()));
MetricsKey.METRIC_RT_MAX_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getMax()));
list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_AVG_AGG.getNameByType(k.getSide()),
MetricsKey.METRIC_RT_AVG_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getAvg()));
MetricsKey.METRIC_RT_AVG_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getAvg()));
});
}
@ -288,14 +306,17 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
public void initQpsMetric(MethodMetric metric) {
ConcurrentHashMapUtils.computeIfAbsent(qps, metric, methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
samplesChanged.set(true);
}
public void initRtMetric(MethodMetric metric) {
ConcurrentHashMapUtils.computeIfAbsent(rt, metric, k -> new TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds));
samplesChanged.set(true);
}
public void initRtAgrMetric(MethodMetric metric) {
ConcurrentHashMapUtils.computeIfAbsent(rtAgr, metric, k -> new TimeWindowAggregator(bucketNum, timeWindowSeconds));
samplesChanged.set(true);
}
public void initWindowCounter(MetricsEvent event, MetricsKey targetKey) {
@ -307,6 +328,13 @@ public class AggregateMetricsCollector implements MetricsCollector<RequestEvent>
ConcurrentMap<MethodMetric, TimeWindowCounter> counter = methodTypeCounter.computeIfAbsent(metricsKeyWrapper, k -> new ConcurrentHashMap<>());
ConcurrentHashMapUtils.computeIfAbsent(counter, metric, methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
samplesChanged.set(true);
}
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
return samplesChanged.compareAndSet(true, false);
}
}

20
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/DefaultMetricsCollector.java
View File

@ -72,6 +72,7 @@ public class DefaultMetricsCollector extends CombMetricsCollector<RequestEvent>
private final AtomicBoolean initialized = new AtomicBoolean();
private final AtomicBoolean samplesChanged = new AtomicBoolean();
public DefaultMetricsCollector(ApplicationModel applicationModel) {
super(new BaseStatComposite(applicationModel) {
@ -91,11 +92,13 @@ public class DefaultMetricsCollector extends CombMetricsCollector<RequestEvent>
super.setEventMulticaster(new DefaultSubDispatcher(this));
samplers.add(applicationSampler);
samplers.add(threadPoolSampler);
samplesChanged.set(true);
this.applicationModel = applicationModel;
}
public void addSampler(MetricsSampler sampler) {
samplers.add(sampler);
samplesChanged.set(true);
}
public void setApplicationName(String applicationName) {
@ -203,5 +206,22 @@ public class DefaultMetricsCollector extends CombMetricsCollector<RequestEvent>
MetricsCountSampleConfigurer<String, MetricsEvent.Type, ApplicationMetric> sampleConfigure) {
sampleConfigure.configureMetrics(configure -> new ApplicationMetric(applicationModel));
}
@Override
public boolean calSamplesChanged() {
return false;
}
};
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
boolean changed = samplesChanged.compareAndSet(true, false);
// Should ensure that all the sampler's samplesChanged have been compareAndSet, and cannot flip the `or` logic
changed = stats.calSamplesChanged() || changed;
for (MetricsSampler sampler : samplers) {
changed = sampler.calSamplesChanged() || changed;
}
return changed;
}
}

9
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/HistogramMetricsCollector.java
View File

@ -17,7 +17,6 @@
package org.apache.dubbo.metrics.collector;
import io.micrometer.core.instrument.Timer;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.config.MetricsConfig;
import org.apache.dubbo.config.context.ConfigManager;
@ -33,6 +32,8 @@ import org.apache.dubbo.metrics.register.HistogramMetricRegister;
import org.apache.dubbo.metrics.sample.HistogramMetricSample;
import org.apache.dubbo.rpc.model.ApplicationModel;
import io.micrometer.core.instrument.Timer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
@ -110,4 +111,10 @@ public class HistogramMetricsCollector extends AbstractMetricsListener<RequestEv
public List<MetricSample> collect() {
return new ArrayList<>();
}
@Override
public boolean calSamplesChanged() {
// Histogram is directly register micrometer
return false;
}
}

9
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MetricsSampler.java
View File

@ -18,9 +18,18 @@
package org.apache.dubbo.metrics.collector.sample;
import org.apache.dubbo.metrics.model.sample.MetricSample;
import java.util.List;
public interface MetricsSampler {
List<MetricSample> sample();
/**
* Check if samples have been changed.
* Note that this method will reset the changed flag to false using CAS.
*
* @return true if samples have been changed
*/
boolean calSamplesChanged();
}

25
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadPoolMetricsSampler.java
View File

@ -23,13 +23,12 @@ import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.metrics.collector.DefaultMetricsCollector;
import org.apache.dubbo.metrics.model.key.MetricsKey;
import org.apache.dubbo.metrics.model.ThreadPoolMetric;
import org.apache.dubbo.metrics.model.key.MetricsKey;
import org.apache.dubbo.metrics.model.sample.GaugeMetricSample;
import org.apache.dubbo.metrics.model.sample.MetricSample;
import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@ -38,10 +37,10 @@ import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SHARED_EXECUTOR_SERVICE_COMPONENT_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_SERVICE_COMPONENT_KEY;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_METRICS_COLLECTOR_EXCEPTION;
import static org.apache.dubbo.config.Constants.CLIENT_THREAD_POOL_NAME;
import static org.apache.dubbo.config.Constants.SERVER_THREAD_POOL_NAME;
@ -56,6 +55,8 @@ public class ThreadPoolMetricsSampler implements MetricsSampler {
private DataStore dataStore;
private final Map<String, ThreadPoolExecutor> sampleThreadPoolExecutor = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, ThreadPoolMetric> threadPoolMetricMap = new ConcurrentHashMap<>();
private final AtomicBoolean samplesChanged = new AtomicBoolean(true);
public ThreadPoolMetricsSampler(DefaultMetricsCollector collector) {
this.collector = collector;
}
@ -63,7 +64,10 @@ public class ThreadPoolMetricsSampler implements MetricsSampler {
public void addExecutors(String name, ExecutorService executorService) {
Optional.ofNullable(executorService).filter(Objects::nonNull).filter(e -> e instanceof ThreadPoolExecutor)
.map(e -> (ThreadPoolExecutor) e)
.ifPresent(threadPoolExecutor -> sampleThreadPoolExecutor.put(name, threadPoolExecutor));
.ifPresent(threadPoolExecutor -> {
sampleThreadPoolExecutor.put(name, threadPoolExecutor);
samplesChanged.set(true);
});
}
@Override
@ -113,7 +117,7 @@ public class ThreadPoolMetricsSampler implements MetricsSampler {
for (Map.Entry<String, Object> entry : executors.entrySet()) {
ExecutorService executor = (ExecutorService) entry.getValue();
if (executor instanceof ThreadPoolExecutor) {
this.addExecutors( SERVER_THREAD_POOL_NAME + "-" + entry.getKey(), executor);
this.addExecutors(SERVER_THREAD_POOL_NAME + "-" + entry.getKey(), executor);
}
}
executors = dataStore.get(CONSUMER_SHARED_EXECUTOR_SERVICE_COMPONENT_KEY);
@ -125,9 +129,9 @@ public class ThreadPoolMetricsSampler implements MetricsSampler {
}
ThreadRejectMetricsCountSampler threadRejectMetricsCountSampler = new ThreadRejectMetricsCountSampler(collector);
this.sampleThreadPoolExecutor.entrySet().stream().filter(entry->entry.getKey().startsWith(SERVER_THREAD_POOL_NAME)).forEach(entry->{
if(entry.getValue().getRejectedExecutionHandler() instanceof AbortPolicyWithReport) {
MetricThreadPoolExhaustedListener metricThreadPoolExhaustedListener=new MetricThreadPoolExhaustedListener(entry.getKey(),threadRejectMetricsCountSampler);
this.sampleThreadPoolExecutor.entrySet().stream().filter(entry -> entry.getKey().startsWith(SERVER_THREAD_POOL_NAME)).forEach(entry -> {
if (entry.getValue().getRejectedExecutionHandler() instanceof AbortPolicyWithReport) {
MetricThreadPoolExhaustedListener metricThreadPoolExhaustedListener = new MetricThreadPoolExhaustedListener(entry.getKey(), threadRejectMetricsCountSampler);
((AbortPolicyWithReport) entry.getValue().getRejectedExecutionHandler()).addThreadPoolExhaustedEventListener(metricThreadPoolExhaustedListener);
}
});
@ -137,4 +141,9 @@ public class ThreadPoolMetricsSampler implements MetricsSampler {
}
}
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
return samplesChanged.compareAndSet(true, false);
}
}

15
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadRejectMetricsCountSampler.java
View File

@ -16,19 +16,23 @@
*/
package org.apache.dubbo.metrics.collector.sample;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.metrics.collector.DefaultMetricsCollector;
import org.apache.dubbo.metrics.model.Metric;
import org.apache.dubbo.metrics.model.MetricsCategory;
import org.apache.dubbo.metrics.model.key.MetricsKey;
import org.apache.dubbo.metrics.model.ThreadPoolRejectMetric;
import org.apache.dubbo.metrics.model.key.MetricsKey;
import org.apache.dubbo.metrics.model.sample.GaugeMetricSample;
import org.apache.dubbo.metrics.model.sample.MetricSample;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.ToDoubleFunction;
import static org.apache.dubbo.metrics.model.MetricsCategory.THREAD_POOL;
public class ThreadRejectMetricsCountSampler extends SimpleMetricsCountSampler<String, String, ThreadPoolRejectMetric> {
@ -36,6 +40,8 @@ public class ThreadRejectMetricsCountSampler extends SimpleMetricsCountSampler<S
private final DefaultMetricsCollector collector;
private final Set<String> metricNames = new ConcurrentHashSet<>();
private final AtomicBoolean samplesChanged = new AtomicBoolean(true);
public ThreadRejectMetricsCountSampler(DefaultMetricsCollector collector) {
this.collector = collector;
this.collector.addSampler(this);
@ -44,6 +50,7 @@ public class ThreadRejectMetricsCountSampler extends SimpleMetricsCountSampler<S
public void addMetricName(String name){
this.metricNames.add(name);
this.initMetricsCounter(name,name);
samplesChanged.set(true);
}
@Override
@ -82,4 +89,10 @@ public class ThreadRejectMetricsCountSampler extends SimpleMetricsCountSampler<S
protected void countConfigure(MetricsCountSampleConfigurer<String, String, ThreadPoolRejectMetric> sampleConfigure) {
sampleConfigure.configureMetrics(configure -> new ThreadPoolRejectMetric(collector.getApplicationName(),configure.getSource()));
}
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
return samplesChanged.compareAndSet(true, false);
}
}

68
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/AbstractMetricsReporter.java
View File

@ -148,38 +148,22 @@ public abstract class AbstractMetricsReporter implements MetricsReporter {
NamedThreadFactory threadFactory = new NamedThreadFactory("metrics-collector-sync-job", true);
collectorSyncJobExecutor = Executors.newScheduledThreadPool(1, threadFactory);
collectorSyncJobExecutor.scheduleWithFixedDelay(this::refreshData, DEFAULT_SCHEDULE_INITIAL_DELAY, collectSyncPeriod, TimeUnit.SECONDS);
collectorSyncJobExecutor.scheduleWithFixedDelay(this::resetIfSamplesChanged, DEFAULT_SCHEDULE_INITIAL_DELAY, collectSyncPeriod, TimeUnit.SECONDS);
}
}
@SuppressWarnings({"unchecked", "rawtypes"})
public void refreshData() {
@SuppressWarnings({"unchecked"})
public void resetIfSamplesChanged() {
collectors.forEach(collector -> {
if (!collector.calSamplesChanged()) {
// Metrics has not been changed since last time, no need to reload
return;
}
// Collect all the samples and register them to the micrometer registry
List<MetricSample> samples = collector.collect();
for (MetricSample sample : samples) {
try {
switch (sample.getType()) {
case GAUGE:
GaugeMetricSample gaugeSample = (GaugeMetricSample) sample;
List<Tag> tags = getTags(gaugeSample);
Gauge.builder(gaugeSample.getName(), gaugeSample.getValue(), gaugeSample.getApply())
.description(gaugeSample.getDescription()).tags(tags).register(compositeRegistry);
break;
case COUNTER:
CounterMetricSample counterMetricSample = (CounterMetricSample) sample;
FunctionCounter.builder(counterMetricSample.getName(), counterMetricSample.getValue(),
Number::doubleValue).description(counterMetricSample.getDescription())
.tags(getTags(counterMetricSample))
.register(compositeRegistry);
case TIMER:
case LONG_TASK_TIMER:
case DISTRIBUTION_SUMMARY:
// TODO
break;
default:
break;
}
registerSample(sample);
} catch (Exception e) {
logger.error(COMMON_METRICS_COLLECTOR_EXCEPTION, "", "", "error occurred when synchronize metrics collector.", e);
}
@ -187,6 +171,40 @@ public abstract class AbstractMetricsReporter implements MetricsReporter {
});
}
@SuppressWarnings({"rawtypes"})
private void registerSample(MetricSample sample) {
switch (sample.getType()) {
case GAUGE:
registerGaugeSample((GaugeMetricSample) sample);
break;
case COUNTER:
registerCounterSample((CounterMetricSample) sample);
case TIMER:
case LONG_TASK_TIMER:
case DISTRIBUTION_SUMMARY:
// TODO
break;
default:
break;
}
}
@SuppressWarnings({"rawtypes"})
private void registerCounterSample(CounterMetricSample sample) {
FunctionCounter.builder(sample.getName(), sample.getValue(), Number::doubleValue)
.description(sample.getDescription())
.tags(getTags(sample))
.register(compositeRegistry);
}
@SuppressWarnings({"unchecked", "rawtypes"})
private void registerGaugeSample(GaugeMetricSample sample) {
Gauge.builder(sample.getName(), sample.getValue(), sample.getApply())
.description(sample.getDescription())
.tags(getTags(sample))
.register(compositeRegistry);
}
private static List<Tag> getTags(MetricSample gaugeSample) {
List<Tag> tags = new ArrayList<>();
gaugeSample.getTags().forEach((k, v) -> {

2
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/nop/NopMetricsReporter.java
View File

@ -35,7 +35,7 @@ public class NopMetricsReporter implements MetricsReporter {
}
@Override
public void refreshData() {
public void resetIfSamplesChanged() {
}

4
dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/collector/MetadataMetricsCollector.java
View File

@ -99,4 +99,8 @@ public class MetadataMetricsCollector extends CombMetricsCollector<MetadataEvent
return list;
}
@Override
public boolean calSamplesChanged() {
return stats.calSamplesChanged();
}
}

2
dubbo-metrics/dubbo-metrics-prometheus/src/main/java/org/apache/dubbo/metrics/prometheus/NopPrometheusMetricsReporter.java
View File

@ -31,7 +31,7 @@ public class NopPrometheusMetricsReporter implements MetricsReporter {
}
@Override
public void refreshData() {
public void resetIfSamplesChanged() {
}

11
dubbo-metrics/dubbo-metrics-prometheus/src/main/java/org/apache/dubbo/metrics/prometheus/PrometheusMetricsReporter.java
View File

@ -17,10 +17,6 @@
package org.apache.dubbo.metrics.prometheus;
import io.micrometer.prometheus.PrometheusConfig;
import io.micrometer.prometheus.PrometheusMeterRegistry;
import io.prometheus.client.exporter.BasicAuthHttpConnectionFactory;
import io.prometheus.client.exporter.PushGateway;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
@ -29,6 +25,11 @@ import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.metrics.report.AbstractMetricsReporter;
import org.apache.dubbo.rpc.model.ApplicationModel;
import io.micrometer.prometheus.PrometheusConfig;
import io.micrometer.prometheus.PrometheusMeterRegistry;
import io.prometheus.client.exporter.BasicAuthHttpConnectionFactory;
import io.prometheus.client.exporter.PushGateway;
import java.io.IOException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@ -90,7 +91,7 @@ public class PrometheusMetricsReporter extends AbstractMetricsReporter {
protected void push(PushGateway pushGateway, String job) {
try {
refreshData();
resetIfSamplesChanged();
pushGateway.pushAdd(prometheusRegistry.getPrometheusRegistry(), job);
} catch (IOException e) {
logger.error(COMMON_METRICS_COLLECTOR_EXCEPTION, "", "", "Error occurred when pushing metrics to prometheus: ", e);

2
dubbo-metrics/dubbo-metrics-prometheus/src/main/java/org/apache/dubbo/metrics/prometheus/PrometheusMetricsReporterCmd.java
View File

@ -116,7 +116,7 @@ public class PrometheusMetricsReporterCmd implements BaseCommand {
logger.debug("scrape begin");
}
metricsReporter.refreshData();
metricsReporter.resetIfSamplesChanged();
if (logger.isDebugEnabled()) {
logger.debug(String.format("scrape end,Elapsed Time：%s", System.currentTimeMillis() - begin));

7
dubbo-metrics/dubbo-metrics-prometheus/src/test/java/org/apache/dubbo/metrics/prometheus/PrometheusMetricsReporterTest.java
View File

@ -17,14 +17,15 @@
package org.apache.dubbo.metrics.prometheus;
import com.sun.net.httpserver.HttpServer;
import io.micrometer.prometheus.PrometheusMeterRegistry;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.config.MetricsConfig;
import org.apache.dubbo.config.nested.PrometheusConfig;
import org.apache.dubbo.metrics.collector.DefaultMetricsCollector;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.FrameworkModel;
import com.sun.net.httpserver.HttpServer;
import io.micrometer.prometheus.PrometheusMeterRegistry;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
@ -143,7 +144,7 @@ class PrometheusMetricsReporterTest {
try {
HttpServer prometheusExporterHttpServer = HttpServer.create(new InetSocketAddress(port), 0);
prometheusExporterHttpServer.createContext("/metrics", httpExchange -> {
reporter.refreshData();
reporter.resetIfSamplesChanged();
String response = reporter.getPrometheusRegistry().scrape();
httpExchange.sendResponseHeaders(200, response.getBytes().length);
try (OutputStream os = httpExchange.getResponseBody()) {

5
dubbo-metrics/dubbo-metrics-prometheus/src/test/java/org/apache/dubbo/metrics/prometheus/PrometheusMetricsThreadPoolTest.java
View File

@ -16,7 +16,6 @@
*/
package org.apache.dubbo.metrics.prometheus;
import com.sun.net.httpserver.HttpServer;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.config.MetricsConfig;
import org.apache.dubbo.config.nested.PrometheusConfig;
@ -25,6 +24,8 @@ import org.apache.dubbo.metrics.collector.sample.ThreadRejectMetricsCountSampler
import org.apache.dubbo.metrics.model.sample.GaugeMetricSample;
import org.apache.dubbo.metrics.model.sample.MetricSample;
import org.apache.dubbo.rpc.model.ApplicationModel;
import com.sun.net.httpserver.HttpServer;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
@ -122,7 +123,7 @@ public class PrometheusMetricsThreadPoolTest {
try {
HttpServer prometheusExporterHttpServer = HttpServer.create(new InetSocketAddress(port), 0);
prometheusExporterHttpServer.createContext("/metrics", httpExchange -> {
reporter.refreshData();
reporter.resetIfSamplesChanged();
String response = reporter.getPrometheusRegistry().scrape();
httpExchange.sendResponseHeaders(200, response.getBytes().length);
try (OutputStream os = httpExchange.getResponseBody()) {

7
dubbo-metrics/dubbo-metrics-registry/src/main/java/org/apache/dubbo/metrics/registry/collector/RegistryMetricsCollector.java
View File

@ -149,4 +149,11 @@ public class RegistryMetricsCollector extends CombMetricsCollector<RegistryEvent
this.stats.setServiceKey(metricsKey, serviceKey, num, attachments);
}
@Override
public boolean calSamplesChanged() {
// Should ensure that all the stat's samplesChanged have been compareAndSet, and cannot flip the `or` logic
boolean changed = stats.calSamplesChanged();
changed = internalStat.calSamplesChanged() || changed;
return changed;
}
}

22
dubbo-metrics/dubbo-metrics-registry/src/main/java/org/apache/dubbo/metrics/registry/collector/RegistryStatComposite.java
View File

@ -33,6 +33,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import static org.apache.dubbo.metrics.MetricsConstants.SELF_INCREMENT_SIZE;
@ -41,6 +42,8 @@ public class RegistryStatComposite extends AbstractMetricsExport {
private final Map<MetricsKey, Map<ApplicationMetric, AtomicLong>> appStats = new ConcurrentHashMap<>();
private final AtomicBoolean samplesChanged = new AtomicBoolean(true);
public RegistryStatComposite(ApplicationModel applicationModel) {
super(applicationModel);
init(RegistryMetricsConstants.REGISTER_LEVEL_KEYS);
@ -50,7 +53,10 @@ public class RegistryStatComposite extends AbstractMetricsExport {
if (CollectionUtils.isEmpty(appKeys)) {
return;
}
appKeys.forEach(appKey -> appStats.put(appKey, new ConcurrentHashMap<>()));
appKeys.forEach(appKey -> {
appStats.put(appKey, new ConcurrentHashMap<>());
});
samplesChanged.set(true);
}
@Override
@ -71,11 +77,23 @@ public class RegistryStatComposite extends AbstractMetricsExport {
}
ApplicationMetric applicationMetric = new ApplicationMetric(getApplicationModel());
applicationMetric.setExtraInfo(Collections.singletonMap(RegistryConstants.REGISTRY_CLUSTER_KEY.toLowerCase(), name));
appStats.get(metricsKey).computeIfAbsent(applicationMetric, k -> new AtomicLong(0L)).getAndAdd(SELF_INCREMENT_SIZE);
Map<ApplicationMetric, AtomicLong> stats = appStats.get(metricsKey);
AtomicLong metrics = stats.get(applicationMetric);
if (metrics == null) {
metrics = stats.computeIfAbsent(applicationMetric, k -> new AtomicLong(0L));
samplesChanged.set(true);
}
metrics.getAndAdd(SELF_INCREMENT_SIZE);
MetricsSupport.fillZero(appStats);
}
public Map<MetricsKey, Map<ApplicationMetric, AtomicLong>> getAppStats() {
return appStats;
}
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
return samplesChanged.compareAndSet(true, false);
}
}

2
dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/command/impl/DefaultMetricsReporterCmd.java
View File

@ -102,7 +102,7 @@ public class DefaultMetricsReporterCmd implements BaseCommand {
String response = "DefaultMetricsReporter not init";
MetricsReporter metricsReporter = applicationModel.getBeanFactory().getBean(DefaultMetricsReporter.class);
if (metricsReporter != null) {
metricsReporter.refreshData();
metricsReporter.resetIfSamplesChanged();
response = metricsReporter.getResponseWithName(metricsName);
}
return response;