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

Merge branch 'apache-3.2' into 3.2.7-release

This commit is contained in:
Albumen Kevin 2023-09-30 10:12:18 +08:00
parent 4655d783e6 70600eb632
commit 53f1391c92
39 changed files with 468 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dubbo-common/src/main/java/org/apache/dubbo/common/constants/MetricsConstants.java
View File

@ -51,6 +51,8 @@ public interface MetricsConstants {
String ENABLE_COLLECTOR_SYNC_KEY = "enable.collector.sync";
String COLLECTOR_SYNC_PERIOD_KEY = "collector.sync.period";
String AGGREGATION_COLLECTOR_KEY = "aggregation";
String AGGREGATION_ENABLED_KEY = "aggregation.enabled";

13
dubbo-common/src/main/java/org/apache/dubbo/config/MetricsConfig.java
View File

@ -71,6 +71,11 @@ public class MetricsConfig extends AbstractConfig {
*/
private Boolean enableCollectorSync;
/**
* Collector sync period.
*/
private Integer collectorSyncPeriod;
/**
* @deprecated After metrics config is refactored.
* This parameter should no longer use and will be deleted in the future.
@ -248,6 +253,14 @@ public class MetricsConfig extends AbstractConfig {
this.enableCollectorSync = enableCollectorSync;
}
public Integer getCollectorSyncPeriod() {
return collectorSyncPeriod;
}
public void setCollectorSyncPeriod(Integer collectorSyncPeriod) {
this.collectorSyncPeriod = collectorSyncPeriod;
}
public Boolean getUseGlobalRegistry() {
return useGlobalRegistry;
}

13
dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ReflectionServiceDescriptor.java
View File

@ -75,7 +75,18 @@ public class ReflectionServiceDescriptor implements ServiceDescriptor {
methods.forEach((methodName, methodList) -> {
Map<String, MethodDescriptor> descMap = descToMethods.computeIfAbsent(methodName, k -> new HashMap<>());
methodList.forEach(methodModel -> descMap.put(methodModel.getParamDesc(), methodModel));
// not support BI_STREAM and SERVER_STREAM at the same time, for example,
// void foo(Request, StreamObserver<Response>) ---> SERVER_STREAM
// StreamObserver<Response> foo(StreamObserver<Request>) ---> BI_STREAM
long streamMethodCount = methodList.stream()
.peek(methodModel -> descMap.put(methodModel.getParamDesc(), methodModel))
.map(MethodDescriptor::getRpcType)
.filter(rpcType -> rpcType == MethodDescriptor.RpcType.SERVER_STREAM
|| rpcType == MethodDescriptor.RpcType.BI_STREAM)
.count();
if (streamMethodCount > 1L)
throw new IllegalStateException("Stream method could not be overloaded.There are " + streamMethodCount
+" stream method signatures. method(" + methodName + ")");
});
}

11
dubbo-common/src/test/java/org/apache/dubbo/rpc/model/ReflectionServiceDescriptorTest.java
View File

@ -21,6 +21,7 @@ import org.apache.dubbo.common.utils.ReflectUtils;
import org.apache.dubbo.metadata.definition.TypeDefinitionBuilder;
import org.apache.dubbo.rpc.support.DemoService;
import org.apache.dubbo.rpc.support.DemoService1;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
@ -42,6 +43,16 @@ class ReflectionServiceDescriptorTest {
Assertions.assertEquals(1, service2.getMethods("sayHello2").size());
}
@Test
void testStreamRpcTypeException() {
try {
new ReflectionServiceDescriptor(DemoService1.class);
} catch (IllegalStateException e) {
Assertions.assertTrue(e.getMessage()
.contains("Stream method could not be overloaded."));
}
}
@Test
void getFullServiceDefinition() {
TypeDefinitionBuilder.initBuilders(new FrameworkModel());

25
dubbo-common/src/test/java/org/apache/dubbo/rpc/support/DemoService1.java Normal file
View File

@ -0,0 +1,25 @@
/*
* 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.rpc.support;
import org.apache.dubbo.common.stream.StreamObserver;
public interface DemoService1 {
StreamObserver<String> sayHello(StreamObserver<String> request);
void sayHello(String msg, StreamObserver<String> request);
}

34
dubbo-common/src/test/java/org/apache/dubbo/rpc/support/DemoService1Impl.java Normal file
View File

@ -0,0 +1,34 @@
/*
* 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.rpc.support;
import org.apache.dubbo.common.stream.StreamObserver;
public class DemoService1Impl implements DemoService1{
@Override
public StreamObserver<String> sayHello(StreamObserver<String> request) {
request.onNext("BI_STREAM");
return request;
}
@Override
public void sayHello(String msg, StreamObserver<String> request) {
request.onNext(msg);
request.onNext("SERVER_STREAM");
request.onCompleted();
}
}

5
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/InternalServiceConfigBuilder.java
View File

@ -262,8 +262,11 @@ public class InternalServiceConfigBuilder<T> {
serviceConfig.setScopeModel(applicationModel.getInternalModule());
serviceConfig.setApplication(applicationConfig);
RegistryConfig registryConfig = new RegistryConfig("N/A");
RegistryConfig registryConfig = new RegistryConfig();
registryConfig.refresh();
registryConfig.setNeedRefresh(false);
registryConfig.setId(this.registryId);
registryConfig.setAddress("N/A");
registryConfig.setScopeModel(this.applicationModel);
serviceConfig.setRegistry(registryConfig);

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);
}
}

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

@ -53,6 +53,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_METRICS_COLLECTOR_EXCEPTION;
import static org.apache.dubbo.common.constants.MetricsConstants.COLLECTOR_SYNC_PERIOD_KEY;
import static org.apache.dubbo.common.constants.MetricsConstants.ENABLE_COLLECTOR_SYNC_KEY;
import static org.apache.dubbo.common.constants.MetricsConstants.ENABLE_JVM_METRICS_KEY;
@ -77,7 +78,7 @@ public abstract class AbstractMetricsReporter implements MetricsReporter {
private ScheduledExecutorService collectorSyncJobExecutor = null;
private static final int DEFAULT_SCHEDULE_INITIAL_DELAY = 5;
private static final int DEFAULT_SCHEDULE_PERIOD = 3;
private static final int DEFAULT_SCHEDULE_PERIOD = 60;
protected AbstractMetricsReporter(URL url, ApplicationModel applicationModel) {
this.url = url;
@ -143,40 +144,26 @@ public abstract class AbstractMetricsReporter implements MetricsReporter {
private void scheduleMetricsCollectorSyncJob() {
boolean enableCollectorSync = url.getParameter(ENABLE_COLLECTOR_SYNC_KEY, true);
if (enableCollectorSync) {
int collectSyncPeriod = url.getParameter(COLLECTOR_SYNC_PERIOD_KEY, DEFAULT_SCHEDULE_PERIOD);
NamedThreadFactory threadFactory = new NamedThreadFactory("metrics-collector-sync-job", true);
collectorSyncJobExecutor = Executors.newScheduledThreadPool(1, threadFactory);
collectorSyncJobExecutor.scheduleWithFixedDelay(this::refreshData, DEFAULT_SCHEDULE_INITIAL_DELAY, DEFAULT_SCHEDULE_PERIOD, 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);
}
@ -184,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;

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() {}

3
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java
View File

@ -59,6 +59,9 @@ public class XdsServiceDiscovery extends ReflectionBasedServiceDiscovery {
@Override
public void doDestroy() {
try {
if (exchanger == null) {
return;
}
exchanger.destroy();
} catch (Throwable t) {
logger.error(REGISTRY_ERROR_INITIALIZE_XDS, "", "", t.getMessage(), t);