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

Merge branch 'apache-3.2' into apache-3.3 

# Conflicts:
#	dubbo-dependencies-bom/pom.xml
#	dubbo-dependencies/dubbo-dependencies-zookeeper-curator5/pom.xml
#	dubbo-dependencies/dubbo-dependencies-zookeeper/pom.xml
#	dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/BaseStatComposite.java
#	dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/DefaultMetricsCollector.java
#	dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadPoolMetricsSampler.java
#	dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadRejectMetricsCountSampler.java
#	pom.xml
This commit is contained in:
Albumen Kevin 2023-10-08 14:46:26 +08:00
parent 0dab17f11d b2e081e513
commit 1848e400a0
35 changed files with 361 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
dubbo-dependencies-bom/pom.xml
View File

@ -96,15 +96,15 @@
<javassist_version>3.29.2-GA</javassist_version>
<bytebuddy.version>1.14.8</bytebuddy.version>
<netty_version>3.2.10.Final</netty_version>
<netty4_version>4.1.97.Final</netty4_version>
<netty4_version>4.1.99.Final</netty4_version>
<mina_version>2.2.1</mina_version>
<grizzly_version>2.4.4</grizzly_version>
<httpclient_version>4.5.14</httpclient_version>
<httpcore_version>4.4.16</httpcore_version>
<fastjson_version>1.2.83</fastjson_version>
<fastjson2_version>2.0.40</fastjson2_version>
<zookeeper_version>3.4.14</zookeeper_version>
<curator_version>4.3.0</curator_version>
<zookeeper_version>3.7.0</zookeeper_version>
<curator_version>5.1.0</curator_version>
<curator_test_version>2.12.0</curator_test_version>
<jedis_version>3.10.0</jedis_version>
<consul_version>1.4.5</consul_version>
@ -182,7 +182,7 @@
<hessian_lite_version>3.2.13</hessian_lite_version>
<swagger_version>1.6.11</swagger_version>
<snappy_java_version>1.1.10.4</snappy_java_version>
<snappy_java_version>1.1.10.5</snappy_java_version>
<bouncycastle-bcprov_version>1.70</bouncycastle-bcprov_version>
<metrics_version>2.0.6</metrics_version>
<sofa_registry_version>5.4.3</sofa_registry_version>
@ -983,16 +983,6 @@
</build>
<profiles>
<profile>
<id>curator5</id>
<activation>
<jdk>[17,)</jdk>
</activation>
<properties>
<zookeeper_version>3.7.0</zookeeper_version>
<curator_version>5.1.0</curator_version>
</properties>
</profile>
<profile>
<id>release</id>
<build>

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) {
@ -84,4 +90,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

@ -137,4 +137,14 @@ public abstract class BaseStatComposite implements MetricsExport {
public void setAppKey(MetricsKey metricsKey, Long num) {
applicationStatComposite.setAppKey(metricsKey, num);
}
@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

@ -78,6 +78,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) {
@ -97,12 +98,14 @@ public class DefaultMetricsCollector extends CombMetricsCollector<RequestEvent>
super.setEventMulticaster(new DefaultSubDispatcher(this));
this.samplers.add(applicationSampler);
this.samplers.add(threadPoolSampler);
this.samplesChanged.set(true);
this.errorCodeSampler = new ErrorCodeSampler(this);
this.applicationModel = applicationModel;
}
public void addSampler(MetricsSampler sampler) {
samplers.add(sampler);
samplesChanged.set(true);
}
public void setApplicationName(String applicationName) {
@ -211,5 +214,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();
}

17
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;
@ -64,7 +63,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
@ -154,5 +156,10 @@ public class ThreadPoolMetricsSampler implements MetricsSampler {
this.addExecutors("sharedScheduledExecutor", frameworkExecutorRepository.getSharedScheduledExecutor());
this.addExecutors("mappingRefreshingExecutor", frameworkExecutorRepository.getMappingRefreshingExecutor());
}
}
@Override
public boolean calSamplesChanged() {
// CAS to get and reset the flag in an atomic operation
return samplesChanged.compareAndSet(true, false);
}
}

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

@ -23,13 +23,18 @@ 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 static org.apache.dubbo.metrics.model.MetricsCategory.THREAD_POOL;
public class ThreadRejectMetricsCountSampler extends MetricsNameCountSampler<String, String, ThreadPoolRejectMetric> {
private final AtomicBoolean samplesChanged = new AtomicBoolean(true);
public ThreadRejectMetricsCountSampler(DefaultMetricsCollector collector) {
super(collector, THREAD_POOL,MetricsKey.THREAD_POOL_THREAD_REJECT_COUNT);
}
@ -49,4 +54,10 @@ public class ThreadRejectMetricsCountSampler extends MetricsNameCountSampler<Str
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) -> {

36
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/AbstractMetricsReporterFactory.java
View File

@ -1,36 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.metrics.report;
import org.apache.dubbo.rpc.model.ApplicationModel;
/**
* AbstractMetricsReporterFactory.
*/
public abstract class AbstractMetricsReporterFactory implements MetricsReporterFactory {
private final ApplicationModel applicationModel;
public AbstractMetricsReporterFactory(ApplicationModel applicationModel) {
this.applicationModel = applicationModel;
}
protected ApplicationModel getApplicationModel() {
return applicationModel;
}
}

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;

6
dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/AuthenticationExceptionTranslatorFilter.java
View File

@ -30,10 +30,14 @@ import org.springframework.security.core.AuthenticationException;
import static org.apache.dubbo.rpc.RpcException.AUTHORIZATION_EXCEPTION;
import static org.apache.dubbo.spring.security.utils.SecurityNames.CORE_JACKSON_2_MODULE_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.JAVA_TIME_MODULE_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.OBJECT_MAPPER_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.SECURITY_CONTEXT_HOLDER_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.SIMPLE_MODULE_CLASS_NAME;
@Activate(group = CommonConstants.PROVIDER, order = Integer.MAX_VALUE, onClass = {SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME})
@Activate(group = CommonConstants.PROVIDER, order = Integer.MAX_VALUE, onClass = {
SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME,
JAVA_TIME_MODULE_CLASS_NAME, SIMPLE_MODULE_CLASS_NAME})
public class AuthenticationExceptionTranslatorFilter implements Filter, Filter.Listener {

13
dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/ContextHolderAuthenticationPrepareFilter.java
View File

@ -18,6 +18,8 @@ package org.apache.dubbo.spring.security.filter;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@ -33,12 +35,17 @@ import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
import static org.apache.dubbo.spring.security.utils.SecurityNames.CORE_JACKSON_2_MODULE_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.JAVA_TIME_MODULE_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.OBJECT_MAPPER_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.SECURITY_CONTEXT_HOLDER_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.SIMPLE_MODULE_CLASS_NAME;
@Activate(group = CommonConstants.CONSUMER, order = -10000, onClass = {SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME})
@Activate(group = CommonConstants.CONSUMER, order = -10000, onClass = {
SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME,
JAVA_TIME_MODULE_CLASS_NAME, SIMPLE_MODULE_CLASS_NAME})
public class ContextHolderAuthenticationPrepareFilter implements ClusterFilter {
private final Logger logger = LoggerFactory.getLogger(getClass());
private final ObjectMapperCodec mapper;
public ContextHolderAuthenticationPrepareFilter(ApplicationModel applicationModel) {
@ -47,7 +54,9 @@ public class ContextHolderAuthenticationPrepareFilter implements ClusterFilter {
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
setSecurityContext(invocation);
if (this.mapper != null) {
setSecurityContext(invocation);
}
return invoker.invoke(invocation);
}

10
dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/ContextHolderAuthenticationResolverFilter.java
View File

@ -32,10 +32,14 @@ import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import static org.apache.dubbo.spring.security.utils.SecurityNames.CORE_JACKSON_2_MODULE_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.JAVA_TIME_MODULE_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.OBJECT_MAPPER_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.SECURITY_CONTEXT_HOLDER_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.SIMPLE_MODULE_CLASS_NAME;
@Activate(group = CommonConstants.PROVIDER, order = -10000, onClass = {SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME})
@Activate(group = CommonConstants.PROVIDER, order = -10000, onClass = {
SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME,
JAVA_TIME_MODULE_CLASS_NAME, SIMPLE_MODULE_CLASS_NAME})
public class ContextHolderAuthenticationResolverFilter implements Filter {
private final ObjectMapperCodec mapper;
@ -46,7 +50,9 @@ public class ContextHolderAuthenticationResolverFilter implements Filter {
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
getSecurityContext(invocation);
if (this.mapper != null) {
getSecurityContext(invocation);
}
return invoker.invoke(invocation);
}

23
dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/model/SecurityScopeModelInitializer.java
View File

@ -19,6 +19,8 @@ package org.apache.dubbo.spring.security.model;
import org.apache.dubbo.common.beans.factory.ScopeBeanFactory;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.model.ModuleModel;
@ -29,22 +31,33 @@ import org.apache.dubbo.spring.security.jackson.ObjectMapperCodecCustomer;
import java.util.Set;
import static org.apache.dubbo.spring.security.utils.SecurityNames.CORE_JACKSON_2_MODULE_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.JAVA_TIME_MODULE_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.OBJECT_MAPPER_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.SECURITY_CONTEXT_HOLDER_CLASS_NAME;
import static org.apache.dubbo.spring.security.utils.SecurityNames.SIMPLE_MODULE_CLASS_NAME;
@Activate(onClass = {SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME})
@Activate(onClass = {SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME,
JAVA_TIME_MODULE_CLASS_NAME, SIMPLE_MODULE_CLASS_NAME})
public class SecurityScopeModelInitializer implements ScopeModelInitializer {
private final Logger logger = LoggerFactory.getLogger(getClass());
@Override
public void initializeFrameworkModel(FrameworkModel frameworkModel) {
ScopeBeanFactory beanFactory = frameworkModel.getBeanFactory();
ObjectMapperCodec objectMapperCodec = beanFactory.getOrRegisterBean(ObjectMapperCodec.class);
try {
ObjectMapperCodec objectMapperCodec = new ObjectMapperCodec();
Set<ObjectMapperCodecCustomer> objectMapperCodecCustomerList = frameworkModel.getExtensionLoader(ObjectMapperCodecCustomer.class).getSupportedExtensionInstances();
Set<ObjectMapperCodecCustomer> objectMapperCodecCustomerList = frameworkModel.getExtensionLoader(ObjectMapperCodecCustomer.class).getSupportedExtensionInstances();
for (ObjectMapperCodecCustomer objectMapperCodecCustomer : objectMapperCodecCustomerList) {
objectMapperCodecCustomer.customize(objectMapperCodec);
for (ObjectMapperCodecCustomer objectMapperCodecCustomer : objectMapperCodecCustomerList) {
objectMapperCodecCustomer.customize(objectMapperCodec);
}
beanFactory.registerBean(objectMapperCodec);
} catch (Throwable t) {
logger.info("Failed to initialize ObjectMapperCodecCustomer and spring security related features are disabled.", t);
}
}

2
dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/utils/SecurityNames.java
View File

@ -24,6 +24,8 @@ final public class SecurityNames {
public static final String SECURITY_CONTEXT_HOLDER_CLASS_NAME = "org.springframework.security.core.context.SecurityContextHolder";
public static final String CORE_JACKSON_2_MODULE_CLASS_NAME = "org.springframework.security.jackson2.CoreJackson2Module";
public static final String OBJECT_MAPPER_CLASS_NAME = "com.fasterxml.jackson.databind.ObjectMapper";
public static final String JAVA_TIME_MODULE_CLASS_NAME = "com.fasterxml.jackson.datatype.jsr310.JavaTimeModule";
public static final String SIMPLE_MODULE_CLASS_NAME = "com.fasterxml.jackson.databind.module.SimpleModule";
private SecurityNames() {}

6
dubbo-registry/dubbo-registry-multiple/src/test/java/org/apache/dubbo/registry/multiple/MultipleRegistry2S2RTest.java
View File

@ -21,7 +21,7 @@ import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.Registry;
import org.apache.dubbo.registry.zookeeper.ZookeeperRegistry;
import org.apache.dubbo.remoting.zookeeper.ZookeeperClient;
import org.apache.dubbo.remoting.zookeeper.curator.CuratorZookeeperClient;
import org.apache.dubbo.remoting.zookeeper.curator5.Curator5ZookeeperClient;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Assumptions;
@ -61,9 +61,9 @@ class MultipleRegistry2S2RTest {
multipleRegistry = (MultipleRegistry) new MultipleRegistryFactory().createRegistry(url);
// for test validation
zookeeperClient = new CuratorZookeeperClient(URL.valueOf(zookeeperConnectionAddress1));
zookeeperClient = new Curator5ZookeeperClient(URL.valueOf(zookeeperConnectionAddress1));
zookeeperRegistry = MultipleRegistryTestUtil.getZookeeperRegistry(multipleRegistry.getServiceRegistries().values());
zookeeperClient2 = new CuratorZookeeperClient(URL.valueOf(zookeeperConnectionAddress2));
zookeeperClient2 = new Curator5ZookeeperClient(URL.valueOf(zookeeperConnectionAddress2));
zookeeperRegistry2 = MultipleRegistryTestUtil.getZookeeperRegistry(multipleRegistry.getServiceRegistries().values());
}

18
dubbo-registry/dubbo-registry-zookeeper/pom.xml
View File

@ -37,7 +37,7 @@
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-remoting-zookeeper</artifactId>
<artifactId>dubbo-remoting-zookeeper-curator5</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
@ -45,20 +45,4 @@
<artifactId>curator-x-discovery</artifactId>
</dependency>
</dependencies>
<profiles>
<profile>
<id>curator5</id>
<activation>
<jdk>[17,)</jdk>
</activation>
<dependencies>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-remoting-zookeeper-curator5</artifactId>
<version>${project.parent.version}</version>
</dependency>
</dependencies>
</profile>
</profiles>
</project>