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

add error type metrics (#11596)

This commit is contained in:
jojocodeX 2023-02-28 15:42:20 +08:00 committed by GitHub
parent 5483693119
commit 15f5b0e950
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 627 additions and 215 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
dubbo-kubernetes/src/test/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscoveryTest.java
View File

@ -16,10 +16,15 @@
*/
package org.apache.dubbo.registry.kubernetes;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import io.fabric8.kubernetes.api.model.Endpoints;
import io.fabric8.kubernetes.api.model.EndpointsBuilder;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodBuilder;
import io.fabric8.kubernetes.api.model.Service;
import io.fabric8.kubernetes.api.model.ServiceBuilder;
import io.fabric8.kubernetes.client.Config;
import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesServer;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.registry.client.DefaultServiceInstance;
@ -38,15 +43,9 @@ import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import io.fabric8.kubernetes.api.model.Endpoints;
import io.fabric8.kubernetes.api.model.EndpointsBuilder;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodBuilder;
import io.fabric8.kubernetes.api.model.Service;
import io.fabric8.kubernetes.api.model.ServiceBuilder;
import io.fabric8.kubernetes.client.Config;
import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesServer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.NAMESPACE;
import static org.awaitility.Awaitility.await;

5
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/event/MetricsEvent.java
View File

@ -54,7 +54,10 @@ public abstract class MetricsEvent {
PROCESSING,
UNKNOWN_FAILED,
TOTAL_FAILED,
APPLICATION_INFO
APPLICATION_INFO,
NETWORK_EXCEPTION,
SERVICE_UNAVAILABLE,
CODEC_EXCEPTION;
}
}

15
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/model/MethodMetric.java
View File

@ -26,15 +26,15 @@ import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_SIDE;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_CHAR_SEPARATOR;
import static org.apache.dubbo.common.constants.CommonConstants.PATH_SEPARATOR;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_IP;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_HOSTNAME;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER_SIDE;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_APPLICATION_NAME;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_INTERFACE_KEY;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_METHOD_KEY;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_GROUP_KEY;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_HOSTNAME;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_INTERFACE_KEY;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_IP;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_METHOD_KEY;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_VERSION_KEY;
import static org.apache.dubbo.common.utils.NetUtils.getLocalHost;
import static org.apache.dubbo.common.utils.NetUtils.getLocalHostName;
@ -51,9 +51,7 @@ public class MethodMetric implements Metric {
private String group;
private String version;
public MethodMetric() {
}
public MethodMetric() {}
public MethodMetric(String applicationName, Invocation invocation) {
this.applicationName = applicationName;
@ -97,7 +95,6 @@ public class MethodMetric implements Metric {
tags.put(TAG_IP, getLocalHost());
tags.put(TAG_HOSTNAME, getLocalHostName());
tags.put(TAG_APPLICATION_NAME, applicationName);
tags.put(TAG_INTERFACE_KEY, interfaceName);
tags.put(TAG_METHOD_KEY, methodName);
tags.put(TAG_GROUP_KEY, group);

23
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/model/MetricsKey.java
View File

@ -22,13 +22,17 @@ public enum MetricsKey {
// provider metrics key
METRIC_REQUESTS("dubbo.%s.requests.total", "Total Requests"),
METRIC_REQUESTS_SUCCEED("dubbo.%s.requests.succeed.total", "Succeed Requests"),
METRIC_REQUEST_BUSINESS_FAILED("dubbo.%s.requests.business.failed.total", "Failed Business Requests"),
METRIC_REQUESTS_SUCCEED("dubbo.%s.requests.succeed.total", "Total Succeed Requests"),
METRIC_REQUEST_BUSINESS_FAILED("dubbo.%s.requests.business.failed.total","Total Failed Business Requests"),
METRIC_REQUESTS_PROCESSING("dubbo.%s.requests.processing", "Processing Requests"),
METRIC_REQUESTS_TIMEOUT("dubbo.%s.requests.timeout.total", "Total Timeout Failed Requests"),
METRIC_REQUESTS_LIMIT("dubbo.%s.requests.limit.total", "Total Limit Failed Requests"),
METRIC_REQUESTS_FAILED("dubbo.%s.requests.unknown.failed.total", "Unknown Failed Requests"),
METRIC_REQUESTS_FAILED("dubbo.%s.requests.unknown.failed.total", "Total Unknown Failed Requests"),
METRIC_REQUESTS_TOTAL_FAILED("dubbo.%s.requests.failed.total", "Total Failed Requests"),
METRIC_REQUESTS_NETWORK_FAILED("dubbo.%s.requests.failed.network.total", "Total network Failed Requests"),
METRIC_REQUESTS_SERVICE_UNAVAILABLE_FAILED("dubbo.%s.requests.failed.service.unavailable.total", "Total Service Unavailable Failed Requests"),
METRIC_REQUESTS_CODEC_FAILED("dubbo.%s.requests.failed.codec.total", "Total Codec Failed Requests"),
METRIC_REQUESTS_TOTAL_AGG("dubbo.%s.requests.total.aggregate", "Aggregated Total Requests"),
METRIC_REQUESTS_SUCCEED_AGG("dubbo.%s.requests.succeed.aggregate", "Aggregated Succeed Requests"),
@ -37,6 +41,9 @@ public enum MetricsKey {
METRIC_REQUESTS_TIMEOUT_AGG("dubbo.%s.requests.timeout.failed.aggregate", "Aggregated timeout Failed Requests"),
METRIC_REQUESTS_LIMIT_AGG("dubbo.%s.requests.limit.aggregate", "Aggregated limit Requests"),
METRIC_REQUESTS_TOTAL_FAILED_AGG("dubbo.%s.requests.failed.total.aggregate", "Aggregated failed total Requests"),
METRIC_REQUESTS_TOTAL_NETWORK_FAILED_AGG("dubbo.%s.requests.failed.network.total.aggregate", "Aggregated failed network total Requests"),
METRIC_REQUESTS_TOTAL_CODEC_FAILED_AGG("dubbo.%s.requests.failed.codec.total.aggregate", "Aggregated failed codec total Requests"),
METRIC_REQUESTS_TOTAL_SERVICE_UNAVAILABLE_FAILED_AGG("dubbo.%s.requests.failed.service.unavailable.total.aggregate", "Aggregated failed codec total Requests"),
METRIC_QPS("dubbo.%s.qps.total", "Query Per Seconds"),
METRIC_RT_LAST("dubbo.%s.rt.milliseconds.last", "Last Response Time"),
@ -88,8 +95,8 @@ public enum MetricsKey {
// consumer metrics key
;
private final String name;
private final String description;
private String name;
private String description;
public final String getName() {
return this.name;
@ -99,6 +106,12 @@ public enum MetricsKey {
return String.format(name, type);
}
public final MetricsKey formatName(String type) {
this.name = String.format(name, type);
return this;
}
public final String getDescription() {
return this.description;
}

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

@ -17,11 +17,6 @@
package org.apache.dubbo.metrics.collector;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.config.MetricsConfig;
import org.apache.dubbo.config.context.ConfigManager;
@ -37,7 +32,11 @@ import org.apache.dubbo.metrics.model.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;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.metrics.model.MetricsCategory.QPS;
import static org.apache.dubbo.metrics.model.MetricsCategory.REQUESTS;
import static org.apache.dubbo.metrics.model.MetricsCategory.RT;
@ -49,24 +48,17 @@ import static org.apache.dubbo.metrics.model.MetricsCategory.RT;
public class AggregateMetricsCollector implements MetricsCollector, MetricsListener {
private int bucketNum;
private int timeWindowSeconds;
private final ConcurrentMap<MethodMetric, TimeWindowCounter> totalRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> succeedRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> unknownFailedRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> businessFailedRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> timeoutRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> limitRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> totalFailedRequests = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowCounter> qps = new ConcurrentHashMap<>();
private final Map<MetricsEvent.Type, ConcurrentHashMap<MethodMetric, TimeWindowCounter>> methodTypeCounter = new ConcurrentHashMap<>();
private final ConcurrentMap<MethodMetric, TimeWindowQuantile> rt = new ConcurrentHashMap<>();
private final ConcurrentHashMap<MethodMetric, TimeWindowCounter> qps = new ConcurrentHashMap<>();
private final ApplicationModel applicationModel;
private static final Integer DEFAULT_COMPRESSION = 100;
private static final Integer DEFAULT_BUCKET_NUM = 10;
private static final Integer DEFAULT_TIME_WINDOW_SECONDS = 120;
public AggregateMetricsCollector(ApplicationModel applicationModel) {
this.registryerEventTypeHandler();
this.applicationModel = applicationModel;
ConfigManager configManager = applicationModel.getApplicationConfigManager();
MetricsConfig config = configManager.getMetrics().orElse(null);
@ -80,10 +72,6 @@ public class AggregateMetricsCollector implements MetricsCollector, MetricsListe
}
}
private void registerListener() {
applicationModel.getBeanFactory().getBean(DefaultMetricsCollector.class).addListener(this);
}
@Override
public void onEvent(MetricsEvent event) {
if (event instanceof RTEvent) {
@ -100,45 +88,24 @@ public class AggregateMetricsCollector implements MetricsCollector, MetricsListe
quantile.add(responseTime);
}
private void onRequestEvent(RequestEvent event) {
MethodMetric metric = (MethodMetric) event.getSource();
RequestEvent.Type type = event.getType();
TimeWindowCounter counter = null;
switch (type) {
case TOTAL:
counter = ConcurrentHashMapUtils.computeIfAbsent(totalRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
TimeWindowCounter qpsCounter = ConcurrentHashMapUtils.computeIfAbsent(qps, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
qpsCounter.increment();
break;
case SUCCEED:
counter = ConcurrentHashMapUtils.computeIfAbsent(succeedRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case UNKNOWN_FAILED:
counter = ConcurrentHashMapUtils.computeIfAbsent(unknownFailedRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case BUSINESS_FAILED:
counter = ConcurrentHashMapUtils.computeIfAbsent(businessFailedRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
case REQUEST_TIMEOUT:
counter = ConcurrentHashMapUtils.computeIfAbsent(timeoutRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
MetricsEvent.Type type = event.getType();
case REQUEST_LIMIT:
counter = ConcurrentHashMapUtils.computeIfAbsent(limitRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
ConcurrentMap<MethodMetric, TimeWindowCounter> counter = methodTypeCounter.get(type);
case TOTAL_FAILED:
counter = ConcurrentHashMapUtils.computeIfAbsent(totalFailedRequests, metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
break;
default:
break;
if (counter == null) {
return;
}
TimeWindowCounter windowCounter = ConcurrentHashMapUtils.computeIfAbsent(counter, metric, methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
if (counter != null) {
counter.increment();
if (type == MetricsEvent.Type.TOTAL) {
TimeWindowCounter qpsCounter = ConcurrentHashMapUtils.computeIfAbsent(qps, metric, methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
qpsCounter.increment();
}
windowCounter.increment();
}
@Override
@ -152,39 +119,51 @@ public class AggregateMetricsCollector implements MetricsCollector, MetricsListe
}
private void collectRequests(List<MetricSample> list) {
totalRequests.forEach((k, v) ->
list.add(getGaugeMetricSample(MetricsKey.METRIC_REQUESTS_TOTAL_AGG, k, v)));
succeedRequests.forEach((k, v) ->
list.add(getGaugeMetricSample(MetricsKey.METRIC_REQUESTS_SUCCEED_AGG, k, v)));
unknownFailedRequests.forEach((k, v) ->
list.add(getGaugeMetricSample(MetricsKey.METRIC_REQUESTS_FAILED_AGG, k, v)));
businessFailedRequests.forEach((k, v) ->
list.add(getGaugeMetricSample(MetricsKey.METRIC_REQUESTS_BUSINESS_FAILED_AGG, k, v)));
timeoutRequests.forEach((k, v) ->
list.add(getGaugeMetricSample(MetricsKey.METRIC_REQUESTS_TIMEOUT_AGG, k, v)));
limitRequests.forEach((k, v) ->
list.add(getGaugeMetricSample(MetricsKey.METRIC_REQUESTS_LIMIT_AGG, k, v)));
totalFailedRequests.forEach((k, v) ->
list.add(getGaugeMetricSample(MetricsKey.METRIC_REQUESTS_TOTAL_FAILED_AGG, k, v)));
collectMethod(list, MetricsEvent.Type.TOTAL, MetricsKey.METRIC_REQUESTS_TOTAL_AGG);
collectMethod(list, MetricsEvent.Type.SUCCEED, MetricsKey.METRIC_REQUESTS_SUCCEED_AGG);
collectMethod(list, MetricsEvent.Type.UNKNOWN_FAILED, MetricsKey.METRIC_REQUESTS_FAILED_AGG);
collectMethod(list, MetricsEvent.Type.BUSINESS_FAILED, MetricsKey.METRIC_REQUESTS_BUSINESS_FAILED_AGG);
collectMethod(list, MetricsEvent.Type.REQUEST_TIMEOUT, MetricsKey.METRIC_REQUESTS_TIMEOUT_AGG);
collectMethod(list, MetricsEvent.Type.REQUEST_LIMIT, MetricsKey.METRIC_REQUESTS_LIMIT_AGG);
collectMethod(list, MetricsEvent.Type.TOTAL_FAILED, MetricsKey.METRIC_REQUESTS_TOTAL_FAILED_AGG);
collectMethod(list, MetricsEvent.Type.NETWORK_EXCEPTION, MetricsKey.METRIC_REQUESTS_TOTAL_NETWORK_FAILED_AGG);
collectMethod(list, MetricsEvent.Type.CODEC_EXCEPTION, MetricsKey.METRIC_REQUESTS_TOTAL_CODEC_FAILED_AGG);
collectMethod(list, MetricsEvent.Type.SERVICE_UNAVAILABLE, MetricsKey.METRIC_REQUESTS_TOTAL_SERVICE_UNAVAILABLE_FAILED_AGG);
}
private GaugeMetricSample getGaugeMetricSample(MetricsKey metricRequestsTotalAgg, MethodMetric k, TimeWindowCounter v) {
return new GaugeMetricSample(metricRequestsTotalAgg.getNameByType(k.getSide()),
metricRequestsTotalAgg.getDescription(), k.getTags(), REQUESTS, v::get);
private void collectMethod(List<MetricSample> list, MetricsEvent.Type eventType, MetricsKey metricsKey) {
ConcurrentHashMap<MethodMetric, TimeWindowCounter> windowCounter = methodTypeCounter.get(eventType);
if (windowCounter != null) {
windowCounter.forEach((k, v) -> list.add(new GaugeMetricSample(metricsKey.formatName(k.getSide()), k.getTags(), REQUESTS, v::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.get() / v.bucketLivedSeconds())));
qps.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_QPS.formatName(k.getSide()), k.getTags(), QPS, () -> v.get() / v.bucketLivedSeconds())));
}
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.quantile(0.99)));
list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_P95.getNameByType(k.getSide()),
MetricsKey.METRIC_RT_P95.getDescription(), k.getTags(), RT, () -> v.quantile(0.95)));
list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_P99.formatName(k.getSide()), k.getTags(), RT, () -> v.quantile(0.99)));
list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_P95.formatName(k.getSide()), k.getTags(), RT, () -> v.quantile(0.95)));
});
}
private void registryerEventTypeHandler() {
methodTypeCounter.put(MetricsEvent.Type.TOTAL, new ConcurrentHashMap<>());
methodTypeCounter.put(MetricsEvent.Type.SUCCEED, new ConcurrentHashMap<>());
methodTypeCounter.put(MetricsEvent.Type.UNKNOWN_FAILED, new ConcurrentHashMap<>());
methodTypeCounter.put(MetricsEvent.Type.BUSINESS_FAILED, new ConcurrentHashMap<>());
methodTypeCounter.put(MetricsEvent.Type.REQUEST_TIMEOUT, new ConcurrentHashMap<>());
methodTypeCounter.put(MetricsEvent.Type.REQUEST_LIMIT, new ConcurrentHashMap<>());
methodTypeCounter.put(MetricsEvent.Type.TOTAL_FAILED, new ConcurrentHashMap<>());
methodTypeCounter.put(MetricsEvent.Type.SERVICE_UNAVAILABLE, new ConcurrentHashMap<>());
methodTypeCounter.put(MetricsEvent.Type.NETWORK_EXCEPTION, new ConcurrentHashMap<>());
methodTypeCounter.put(MetricsEvent.Type.CODEC_EXCEPTION, new ConcurrentHashMap<>());
}
private void registerListener() {
applicationModel.getBeanFactory().getBean(DefaultMetricsCollector.class).addListener(this);
}
}

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

@ -29,10 +29,8 @@ import org.apache.dubbo.metrics.model.ApplicationMetric;
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 static org.apache.dubbo.metrics.model.MetricsCategory.APPLICATION;
import static org.apache.dubbo.metrics.model.MetricsKey.APPLICATION_METRIC_INFO;
@ -84,6 +82,10 @@ public class DefaultMetricsCollector implements MetricsCollector {
return this.methodSampler;
}
public ThreadPoolMetricsSampler getThreadPoolSampler() {
return this.threadPoolSampler;
}
public void collectApplication(ApplicationModel applicationModel) {
this.setApplicationName(applicationModel.getApplicationName());
this.applicationModel = applicationModel;

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

@ -28,11 +28,9 @@ import org.apache.dubbo.metrics.model.MetricsKey;
import org.apache.dubbo.metrics.model.sample.GaugeMetricSample;
import org.apache.dubbo.metrics.model.sample.MetricSample;
import org.apache.dubbo.rpc.Invocation;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Supplier;
import static org.apache.dubbo.metrics.model.MetricsCategory.REQUESTS;
import static org.apache.dubbo.metrics.model.MetricsCategory.RT;
@ -63,7 +61,8 @@ public class MethodMetricsSampler extends SimpleMetricsCountSampler<Invocation,
List<MetricSample> metricSamples = new ArrayList<>();
collect(metricSamples);
metricSamples.addAll(collectRT((key, metric, count) -> getGaugeMetricSample(key, metric, RT, () -> count)));
metricSamples.addAll(
this.collectRT((key, metric, count) -> getGaugeMetricSample(key, metric, RT, () -> count)));
return metricSamples;
}
@ -77,8 +76,12 @@ public class MethodMetricsSampler extends SimpleMetricsCountSampler<Invocation,
count(list, MetricsEvent.Type.REQUEST_TIMEOUT, MetricsKey.METRIC_REQUESTS_TIMEOUT);
count(list, MetricsEvent.Type.REQUEST_LIMIT, MetricsKey.METRIC_REQUESTS_LIMIT);
count(list, MetricsEvent.Type.TOTAL_FAILED, MetricsKey.METRIC_REQUESTS_TOTAL_FAILED);
count(list, MetricsEvent.Type.NETWORK_EXCEPTION, MetricsKey.METRIC_REQUESTS_NETWORK_FAILED);
count(list, MetricsEvent.Type.SERVICE_UNAVAILABLE, MetricsKey.METRIC_REQUESTS_SERVICE_UNAVAILABLE_FAILED);
count(list,MetricsEvent.Type.CODEC_EXCEPTION,MetricsKey.METRIC_REQUESTS_CODEC_FAILED);
}
private GaugeMetricSample getGaugeMetricSample(MetricsKey metricsKey, MethodMetric methodMetric,
MetricsCategory metricsCategory, Supplier<Number> get) {
return new GaugeMetricSample(metricsKey.getNameByType(methodMetric.getSide()), metricsKey.getDescription(),

1
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MetricsCountSampleConfigurer.java
View File

@ -18,7 +18,6 @@
package org.apache.dubbo.metrics.collector.sample;
import org.apache.dubbo.metrics.model.Metric;
import java.util.function.Consumer;
import java.util.function.Function;

4
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MetricsCountSampler.java
View File

@ -38,10 +38,14 @@ public interface MetricsCountSampler<S, K, M extends Metric> extends MetricsSamp
void addRT(S source, Long rt);
void addRT(S source, K metricName, Long rt);
Optional<ConcurrentMap<M, AtomicLong>> getCount(K metricName);
<R extends MetricSample> List<R> collectRT(MetricSampleFactory<M, R> factory);
<R extends MetricSample> List<R> collectRT(MetricSampleFactory<M, R> factory,K metricName);
interface MetricSampleFactory<M, R extends MetricSample> {
R newInstance(MetricsKey key, M metric, Long count);
}

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

@ -18,7 +18,6 @@
package org.apache.dubbo.metrics.collector.sample;
import org.apache.dubbo.metrics.model.sample.MetricSample;
import java.util.List;
public interface MetricsSampler {

96
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/SimpleMetricsCountSampler.java
View File

@ -44,14 +44,15 @@ public abstract class SimpleMetricsCountSampler<S, K, M extends Metric>
implements MetricsCountSampler<S, K, M> {
private final ConcurrentMap<M, AtomicLong> EMPTY_COUNT = new ConcurrentHashMap<>();
private Map<K, ConcurrentMap<M, AtomicLong>> metricCounter = new ConcurrentHashMap<>();
// lastRT, totalRT, rtCount, avgRT share a container, can utilize the system cache line
private final ConcurrentMap<M, AtomicLongArray> rtSample = new ConcurrentHashMap<>();
private final ConcurrentMap<M, LongAccumulator> minRT = new ConcurrentHashMap<>();
private final ConcurrentMap<M, LongAccumulator> maxRT = new ConcurrentHashMap<>();
// lastRT, totalRT, rtCount, avgRT share a container, can utilize the system cache line
private final ConcurrentMap<M, AtomicLongArray> rtArray = new ConcurrentHashMap<>();
private final Map<K, ConcurrentMap<M, AtomicLong>> metricCounter = new ConcurrentHashMap<>();
private final ConcurrentMap<K, ConcurrentMap<M,AtomicLongArray>> rtGroupSample = new ConcurrentHashMap<>();
private final ConcurrentMap<K, ConcurrentMap<M,LongAccumulator>> groupMinRT = new ConcurrentHashMap<>();
private final ConcurrentMap<K, ConcurrentMap<M,LongAccumulator>> groupMaxRT = new ConcurrentHashMap<>();
@Override
public void inc(S source, K metricName) {
@ -84,7 +85,6 @@ public abstract class SimpleMetricsCountSampler<S, K, M extends Metric>
return true;
});
}
@Override
public void addRT(S source, Long rt) {
MetricsCountSampleConfigurer<S, K, M> sampleConfigure = new MetricsCountSampleConfigurer<>();
@ -94,16 +94,16 @@ public abstract class SimpleMetricsCountSampler<S, K, M extends Metric>
M metric = sampleConfigure.getMetric();
AtomicLongArray rtArray = ConcurrentHashMapUtils.computeIfAbsent(this.rtArray, metric, k -> new AtomicLongArray(4));
AtomicLongArray rtCalculator = ConcurrentHashMapUtils.computeIfAbsent(this.rtSample, metric, k -> new AtomicLongArray(4));
// set lastRT
rtArray.set(0, rt);
rtCalculator.set(0, rt);
// add to totalRT
rtArray.addAndGet(1, rt);
rtCalculator.addAndGet(1, rt);
// add to rtCount
rtArray.incrementAndGet(2);
rtCalculator.incrementAndGet(2);
// calc avgRT. In order to reduce the amount of calculation, calculated when collect
//rtArray.set(3, Math.floorDiv(rtArray.get(1), rtArray.get(2)));
@ -119,6 +119,49 @@ public abstract class SimpleMetricsCountSampler<S, K, M extends Metric>
sampleConfigure.getFireEventHandler().accept(sampleConfigure);
}
@Override
public void addRT(S source, K metricName, Long rt) {
MetricsCountSampleConfigurer<S,K,M> sampleConfigure = new MetricsCountSampleConfigurer<>();
sampleConfigure.setSource(source);
sampleConfigure.setMetricsName(metricName);
this.rtConfigure(sampleConfigure);
M metric = sampleConfigure.getMetric();
ConcurrentMap<M, AtomicLongArray> nameToCalculator = rtGroupSample.get(metricName);
if (nameToCalculator == null) {
ConcurrentHashMap<M, AtomicLongArray> calculator = new ConcurrentHashMap<>();
calculator.put(metric, new AtomicLongArray(4));
rtGroupSample.put(metricName,calculator);
nameToCalculator = rtGroupSample.get(metricName);
}
AtomicLongArray calculator = nameToCalculator.get(metric);
// set lastRT
calculator.set(0, rt);
// add to totalRT
calculator.addAndGet(1, rt);
// add to rtCount
calculator.incrementAndGet(2);
ConcurrentMap<M, LongAccumulator> minRT = ConcurrentHashMapUtils.computeIfAbsent(groupMinRT, metricName, k -> new ConcurrentHashMap<>());
LongAccumulator min = ConcurrentHashMapUtils.computeIfAbsent(minRT, metric, k -> new LongAccumulator(Long::min, Long.MAX_VALUE));
min.accumulate(rt);
ConcurrentMap<M, LongAccumulator> maxRT = ConcurrentHashMapUtils.computeIfAbsent(groupMaxRT, metricName, k -> new ConcurrentHashMap<>());
LongAccumulator max = ConcurrentHashMapUtils.computeIfAbsent(maxRT, metric, k -> new LongAccumulator(Long::max, Long.MIN_VALUE));
max.accumulate(rt);
sampleConfigure.setRt(rt);
sampleConfigure.getFireEventHandler().accept(sampleConfigure);
}
@Override
public Optional<ConcurrentMap<M, AtomicLong>> getCount(K metricName) {
return Optional.ofNullable(metricCounter.get(metricName) == null ?
@ -128,29 +171,40 @@ public abstract class SimpleMetricsCountSampler<S, K, M extends Metric>
@Override
public <R extends MetricSample> List<R> collectRT(MetricSampleFactory<M, R> factory) {
final List<R> rtMetricSamples = new ArrayList<>();
rtArray.forEach((k, v) -> {
return collect(factory, rtSample, this.minRT, this.maxRT);
}
public <R extends MetricSample> List<R> collectRT(MetricSampleFactory<M, R> factory,K metricName){
return collect(factory, rtGroupSample.get(metricName), groupMinRT.get(metricName),
groupMaxRT.get(metricName));
}
private <R extends MetricSample> List<R> collect(MetricSampleFactory<M, R> factory,
ConcurrentMap<M, AtomicLongArray> rtSample,
ConcurrentMap<M, LongAccumulator> min,
ConcurrentMap<M, LongAccumulator> max){
final List<R> result = new ArrayList<>();
rtSample.forEach((k, v) -> {
// lastRT
rtMetricSamples.add(factory.newInstance(MetricsKey.METRIC_RT_LAST, k, v.get(0)));
result.add(factory.newInstance(MetricsKey.METRIC_RT_LAST, k, v.get(0)));
// totalRT
long totalRT = v.get(1);
long rtCount = v.get(2);
rtMetricSamples.add(factory.newInstance(MetricsKey.METRIC_RT_SUM, k, totalRT));
result.add(factory.newInstance(MetricsKey.METRIC_RT_SUM, k, totalRT));
// avgRT
rtMetricSamples.add(factory.newInstance(MetricsKey.METRIC_RT_AVG, k, Math.floorDiv(totalRT, rtCount)));
result.add(factory.newInstance(MetricsKey.METRIC_RT_AVG, k, Math.floorDiv(totalRT, rtCount)));
});
this.minRT.forEach((k, v) ->
rtMetricSamples.add(factory.newInstance(MetricsKey.METRIC_RT_MIN, k, v.get())));
min.forEach((k, v) ->
result.add(factory.newInstance(MetricsKey.METRIC_RT_MIN, k, v.get())));
this.maxRT.forEach((k, v) ->
rtMetricSamples.add(factory.newInstance(MetricsKey.METRIC_RT_MAX, k, v.get())));
max.forEach((k, v) ->
result.add(factory.newInstance(MetricsKey.METRIC_RT_MAX, k, v.get())));
return rtMetricSamples;
return result;
}
protected void rtConfigure(MetricsCountSampleConfigurer<S, K, M> configure) {
}
protected abstract void countConfigure(MetricsCountSampleConfigurer<S, K, M> sampleConfigure);

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

@ -15,70 +15,84 @@
* limitations under the License.
*/
package org.apache.dubbo.metrics.collector.sample;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.metrics.collector.DefaultMetricsCollector;
import org.apache.dubbo.metrics.model.MetricsKey;
import org.apache.dubbo.metrics.model.ThreadPoolMetric;
import org.apache.dubbo.metrics.model.sample.GaugeMetricSample;
import org.apache.dubbo.metrics.model.sample.MetricSample;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_METRICS_COLLECTOR_EXCEPTION;
import static org.apache.dubbo.metrics.model.MetricsCategory.THREAD_POOL;
public class ThreadPoolMetricsSampler implements MetricsSampler {
private final DefaultMetricsCollector collector;
private final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(ThreadPoolMetricsSampler.class);
private DefaultMetricsCollector collector;
private FrameworkExecutorRepository frameworkExecutorRepository;
private final Set<ThreadPoolMetric> threadPoolMetricSet = new HashSet<>();
private Map<String, ThreadPoolExecutor> sampleThreadPoolExecutor = new ConcurrentHashMap<>();
public ThreadPoolMetricsSampler(DefaultMetricsCollector collector) {
this.collector = collector;
this.registryDefaultSampleThreadPoolExecutor();
}
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));
}
@Override
public List<MetricSample> sample() {
collect();
List<MetricSample> metricSamples = new ArrayList<>();
threadPoolMetricSet.forEach(e -> metricSamples.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_CORE_SIZE, e.getTags(), THREAD_POOL, e::getCorePoolSize)));
threadPoolMetricSet.forEach(e -> metricSamples.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_LARGEST_SIZE, e.getTags(), THREAD_POOL, e::getLargestPoolSize)));
threadPoolMetricSet.forEach(e -> metricSamples.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_MAX_SIZE, e.getTags(), THREAD_POOL, e::getMaximumPoolSize)));
threadPoolMetricSet.forEach(e -> metricSamples.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_ACTIVE_SIZE, e.getTags(), THREAD_POOL, e::getActiveCount)));
threadPoolMetricSet.forEach(e -> metricSamples.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_THREAD_COUNT, e.getTags(), THREAD_POOL, e::getPoolSize)));
threadPoolMetricSet.forEach(e -> metricSamples.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_QUEUE_SIZE, e.getTags(), THREAD_POOL, e::getQueueSize)));
sampleThreadPoolExecutor.forEach((name, executor)->{
metricSamples.addAll(createMetricsSample(name,executor));
});
return metricSamples;
}
private void collect() {
private List<MetricSample> createMetricsSample(String name,ThreadPoolExecutor executor) {
List<MetricSample> list = new ArrayList<>();
ThreadPoolMetric poolMetrics = new ThreadPoolMetric(collector.getApplicationName(), name, executor);
list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_CORE_SIZE, poolMetrics.getTags(), THREAD_POOL, poolMetrics::getCorePoolSize));
list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_LARGEST_SIZE, poolMetrics.getTags(), THREAD_POOL, poolMetrics::getLargestPoolSize));
list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_MAX_SIZE, poolMetrics.getTags(), THREAD_POOL, poolMetrics::getMaximumPoolSize));
list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_ACTIVE_SIZE, poolMetrics.getTags(), THREAD_POOL, poolMetrics::getActiveCount));
list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_THREAD_COUNT, poolMetrics.getTags(), THREAD_POOL, poolMetrics::getPoolSize));
list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_QUEUE_SIZE, poolMetrics.getTags(), THREAD_POOL, poolMetrics::getQueueSize));
return list;
}
private void registryDefaultSampleThreadPoolExecutor() {
try {
if (this.frameworkExecutorRepository == null) {
this.frameworkExecutorRepository = collector.getApplicationModel().getFrameworkModel().getBeanFactory().getBean(FrameworkExecutorRepository.class);
this.frameworkExecutorRepository = collector.getApplicationModel()
.getFrameworkModel().getBeanFactory()
.getBean(FrameworkExecutorRepository.class);
}
} catch (Exception ignored) {
} catch (Exception ex) {
logger.warn(COMMON_METRICS_COLLECTOR_EXCEPTION, "", "", "ThreadPoolMetricsSampler! frameworkExecutorRepository non-init");
}
if (frameworkExecutorRepository != null) {
addThread("SharedExecutor", frameworkExecutorRepository.getSharedExecutor());
addThread("MappingRefreshingExecutor", frameworkExecutorRepository.getMappingRefreshingExecutor());
addThread("PoolRouterExecutor", frameworkExecutorRepository.getPoolRouterExecutor());
if (this.frameworkExecutorRepository != null) {
this.addExecutors("sharedExecutor", frameworkExecutorRepository.getSharedExecutor());
this.addExecutors("mappingRefreshingExecutor", frameworkExecutorRepository.getMappingRefreshingExecutor());
this.addExecutors("poolRouterExecutor", frameworkExecutorRepository.getPoolRouterExecutor());
}
}
private void addThread(String threadPoolName, ExecutorService executorService) {
Optional<ExecutorService> executorOptional = Optional.ofNullable(executorService);
if (executorOptional.isPresent() && executorOptional.get() instanceof ThreadPoolExecutor) {
threadPoolMetricSet.add(
new ThreadPoolMetric(collector.getApplicationName(), threadPoolName,
(ThreadPoolExecutor) executorOptional.get()));
}
}
}

77
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/filter/MethodMetricsInterceptor.java
View File

@ -16,15 +16,11 @@
*/
package org.apache.dubbo.metrics.filter;
import org.apache.dubbo.metrics.collector.sample.MethodMetricsSampler;
import org.apache.dubbo.metrics.event.MetricsEvent;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
import java.util.function.Supplier;
import static org.apache.dubbo.common.constants.MetricsConstants.METRIC_FILTER_START_TIME;
public class MethodMetricsInterceptor {
@ -35,60 +31,61 @@ public class MethodMetricsInterceptor {
this.sampler = sampler;
}
public void beforeExecute(Invocation invocation) {
public void beforeMethod(Invocation invocation) {
sampler.incOnEvent(invocation, MetricsEvent.Type.TOTAL);
sampler.incOnEvent(invocation,MetricsEvent.Type.PROCESSING);
invocation.put(METRIC_FILTER_START_TIME, System.currentTimeMillis());
}
public void postExecute(Invocation invocation, Result result) {
public void afterMethod(Invocation invocation, Result result) {
if (result.hasException()) {
throwExecute(invocation, result.getException());
return;
handleMethodException(invocation, result.getException());
}else{
sampler.incOnEvent(invocation,MetricsEvent.Type.SUCCEED);
onCompleted(invocation);
}
sampler.incOnEvent(invocation,MetricsEvent.Type.SUCCEED);
endExecute(invocation);
}
public void throwExecute(Invocation invocation, Throwable throwable) {
public void handleMethodException(Invocation invocation, Throwable throwable) {
if (throwable == null) {
return;
}
if (throwable instanceof RpcException) {
RpcException rpcException = (RpcException) throwable;
switch (rpcException.getCode()) {
RpcException e = (RpcException) throwable;
case RpcException.TIMEOUT_EXCEPTION:
sampler.incOnEvent(invocation,MetricsEvent.Type.REQUEST_TIMEOUT);
break;
MetricsEvent.Type eventType = MetricsEvent.Type.UNKNOWN_FAILED;
case RpcException.LIMIT_EXCEEDED_EXCEPTION:
sampler.incOnEvent(invocation,MetricsEvent.Type.REQUEST_LIMIT);
break;
case RpcException.BIZ_EXCEPTION:
sampler.incOnEvent(invocation,MetricsEvent.Type.BUSINESS_FAILED);
break;
default:
sampler.incOnEvent(invocation,MetricsEvent.Type.UNKNOWN_FAILED);
if (e.isTimeout()) {
eventType = MetricsEvent.Type.REQUEST_TIMEOUT;
}
if (e.isLimitExceed()) {
eventType = MetricsEvent.Type.REQUEST_LIMIT;
}
if (e.isBiz()) {
eventType = MetricsEvent.Type.BUSINESS_FAILED;
}
if (e.isSerialization()) {
eventType = MetricsEvent.Type.CODEC_EXCEPTION;
}
if (e.isNetwork()) {
eventType = MetricsEvent.Type.NETWORK_EXCEPTION;
}
sampler.incOnEvent(invocation,eventType);
}
if (throwable instanceof RpcException && ((RpcException) throwable).isBiz()) {
onCompleted(invocation);
}
sampler.incOnEvent(invocation,MetricsEvent.Type.TOTAL_FAILED);
endExecute(invocation, () -> throwable instanceof RpcException && ((RpcException) throwable).isBiz());
}
private void endExecute(Invocation invocation) {
endExecute(invocation, () -> true);
}
private void endExecute(Invocation invocation, Supplier<Boolean> rtStat) {
if (rtStat.get()) {
Long endTime = System.currentTimeMillis();
Long beginTime = (Long) invocation.get(METRIC_FILTER_START_TIME);
Long rt = endTime - beginTime;
sampler.addRT(invocation, rt);
}
private void onCompleted(Invocation invocation) {
Long endTime = System.currentTimeMillis();
Long beginTime = (Long) invocation.get(METRIC_FILTER_START_TIME);
Long rt = endTime - beginTime;
sampler.addRT(invocation, rt);
sampler.dec(invocation,MetricsEvent.Type.PROCESSING);
}
}

68
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/filter/MetricsClusterFilter.java Normal file
View File

@ -0,0 +1,68 @@
/*
* 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.filter;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.metrics.collector.DefaultMetricsCollector;
import org.apache.dubbo.metrics.event.MetricsEvent;
import org.apache.dubbo.rpc.BaseFilter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.filter.ClusterFilter;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ScopeModelAware;
import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER;
@Activate(group = CONSUMER)
public class MetricsClusterFilter implements ClusterFilter, BaseFilter.Listener, ScopeModelAware {
private DefaultMetricsCollector collector;
@Override
public void setApplicationModel(ApplicationModel applicationModel) {
this.collector = applicationModel.getBeanFactory().getBean(DefaultMetricsCollector.class);
}
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
return invoker.invoke(invocation);
}
@Override
public void onResponse(Result result, Invoker<?> invoker, Invocation invocation) {
handleMethodException(result.getException(), invocation);
}
@Override
public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
handleMethodException(t, invocation);
}
private void handleMethodException(Throwable t, Invocation invocation) {
if (collector == null || !collector.isCollectEnabled()) {
return;
}
if (t != null && t instanceof RpcException) {
RpcException e = (RpcException) t;
if (e.isForbidden()) {
collector.getMethodSampler().incOnEvent(invocation, MetricsEvent.Type.SERVICE_UNAVAILABLE);
}
}
}
}

10
dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/filter/MetricsFilter.java
View File

@ -34,10 +34,8 @@ import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER;
public class MetricsFilter implements Filter, BaseFilter.Listener, ScopeModelAware {
private DefaultMetricsCollector collector = null;
private MethodMetricsInterceptor metricsInterceptor;
@Override
public void setApplicationModel(ApplicationModel applicationModel) {
collector = applicationModel.getBeanFactory().getBean(DefaultMetricsCollector.class);
@ -45,7 +43,6 @@ public class MetricsFilter implements Filter, BaseFilter.Listener, ScopeModelAwa
if (collector != null) {
metricsInterceptor = new MethodMetricsInterceptor(collector.getMethodSampler());
}
}
@Override
@ -54,10 +51,9 @@ public class MetricsFilter implements Filter, BaseFilter.Listener, ScopeModelAwa
return invoker.invoke(invocation);
}
metricsInterceptor.beforeExecute(invocation);
metricsInterceptor.beforeMethod(invocation);
return invoker.invoke(invocation);
}
@Override
@ -65,7 +61,7 @@ public class MetricsFilter implements Filter, BaseFilter.Listener, ScopeModelAwa
if (collector == null || !collector.isCollectEnabled()) {
return;
}
metricsInterceptor.postExecute(invocation, result);
metricsInterceptor.afterMethod(invocation, result);
}
@Override
@ -73,7 +69,7 @@ public class MetricsFilter implements Filter, BaseFilter.Listener, ScopeModelAwa
if (collector == null || !collector.isCollectEnabled()) {
return;
}
metricsInterceptor.throwExecute(invocation, t);
metricsInterceptor.handleMethodException(invocation, t);
}
}

1
dubbo-metrics/dubbo-metrics-default/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.filter.ClusterFilter
View File

@ -1 +1,2 @@
observationsender=org.apache.dubbo.metrics.observation.ObservationSenderFilter
metricsClusterFilter=org.apache.dubbo.metrics.filter.MetricsClusterFilter

16
dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollectorTest.java
View File

@ -105,10 +105,14 @@ class AggregateMetricsCollectorTest {
defaultCollector.setApplicationName(applicationName);
MethodMetricsSampler methodMetricsCountSampler = defaultCollector.getMethodSampler();
methodMetricsCountSampler.incOnEvent(invocation, MetricsEvent.Type.TOTAL);
methodMetricsCountSampler.incOnEvent(invocation, MetricsEvent.Type.SUCCEED);
methodMetricsCountSampler.incOnEvent(invocation, MetricsEvent.Type.UNKNOWN_FAILED);
methodMetricsCountSampler.incOnEvent(invocation, MetricsEvent.Type.BUSINESS_FAILED);
methodMetricsCountSampler.incOnEvent(invocation,MetricsEvent.Type.TOTAL);
methodMetricsCountSampler.incOnEvent(invocation,MetricsEvent.Type.SUCCEED);
methodMetricsCountSampler.incOnEvent(invocation,MetricsEvent.Type.UNKNOWN_FAILED);
methodMetricsCountSampler.incOnEvent(invocation,MetricsEvent.Type.BUSINESS_FAILED);
methodMetricsCountSampler.incOnEvent(invocation,MetricsEvent.Type.NETWORK_EXCEPTION);
methodMetricsCountSampler.incOnEvent(invocation,MetricsEvent.Type.SERVICE_UNAVAILABLE);
methodMetricsCountSampler.incOnEvent(invocation,MetricsEvent.Type.CODEC_EXCEPTION);
List<MetricSample> samples = collector.collect();
for (MetricSample sample : samples) {
@ -131,6 +135,10 @@ class AggregateMetricsCollectorTest {
Assertions.assertEquals(sampleMap.get(MetricsKey.METRIC_REQUESTS_FAILED_AGG.getNameByType(side)), 1L);
Assertions.assertEquals(sampleMap.get(MetricsKey.METRIC_REQUESTS_BUSINESS_FAILED_AGG.getNameByType(side)), 1L);
Assertions.assertEquals(sampleMap.get(MetricsKey.METRIC_REQUESTS_TOTAL_NETWORK_FAILED_AGG.getNameByType(side)), 1L);
Assertions.assertEquals(sampleMap.get(MetricsKey.METRIC_REQUESTS_TOTAL_CODEC_FAILED_AGG.getNameByType(side)), 1L);
Assertions.assertEquals(sampleMap.get(MetricsKey.METRIC_REQUESTS_TOTAL_SERVICE_UNAVAILABLE_FAILED_AGG.getNameByType(side)), 1L);
Assertions.assertTrue(sampleMap.containsKey(MetricsKey.METRIC_QPS.getNameByType(side)));
}

81
dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/filter/MetricsFilterTest.java
View File

@ -44,6 +44,11 @@ import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE;
import static org.apache.dubbo.common.constants.CommonConstants.GENERIC_PARAMETER_DESC;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_GROUP_KEY;
@ -57,6 +62,7 @@ class MetricsFilterTest {
private ApplicationModel applicationModel;
private MetricsFilter filter;
private MetricsClusterFilter metricsClusterFilter;
private DefaultMetricsCollector collector;
private RpcInvocation invocation;
private final Invoker<?> invoker = mock(Invoker.class);
@ -86,6 +92,8 @@ class MetricsFilterTest {
invocation.setInvoker(new TestMetricsInvoker(side));
RpcContext.getServiceContext().setUrl(URL.valueOf("test://test:11/test?accesslog=true&group=dubbo&version=1.1&side=" + side));
metricsClusterFilter = new MetricsClusterFilter();
metricsClusterFilter.setApplicationModel(applicationModel);
}
@AfterEach
@ -128,7 +136,6 @@ class MetricsFilterTest {
Assertions.assertEquals(tags.get(TAG_VERSION_KEY), VERSION);
}
@Test
void testBusinessFailedRequests() {
collector.setCollectEnabled(true);
@ -257,6 +264,78 @@ class MetricsFilterTest {
Assertions.assertEquals(tags.get(TAG_VERSION_KEY), VERSION);
}
@Test
public void testErrors(){
testFilterError(RpcException.SERIALIZATION_EXCEPTION, MetricsKey.METRIC_REQUESTS_CODEC_FAILED.formatName(side));
testFilterError(RpcException.NETWORK_EXCEPTION, MetricsKey.METRIC_REQUESTS_NETWORK_FAILED.formatName(side));
}
@Test
public void testNoProvider(){
testClusterFilterError(RpcException.FORBIDDEN_EXCEPTION,
MetricsKey.METRIC_REQUESTS_SERVICE_UNAVAILABLE_FAILED.formatName(CommonConstants.CONSUMER));
}
private void testClusterFilterError(int errorCode,MetricsKey metricsKey){
setup();
collector.setCollectEnabled(true);
given(invoker.invoke(invocation)).willThrow(new RpcException(errorCode));
initParam();
Long count = 1L;
for (int i = 0; i < count; i++) {
try {
metricsClusterFilter.invoke(invoker, invocation);
} catch (Exception e) {
Assertions.assertTrue(e instanceof RpcException);
metricsClusterFilter.onError(e, invoker, invocation);
}
}
Map<String, MetricSample> metricsMap = getMetricsMap();
Assertions.assertTrue(metricsMap.containsKey(metricsKey.getName()));
MetricSample sample = metricsMap.get(metricsKey.getName());
Assertions.assertSame(((GaugeMetricSample) sample).getSupplier().get().longValue(), count);
teardown();
}
private void testFilterError(int errorCode,MetricsKey metricsKey){
setup();
collector.setCollectEnabled(true);
given(invoker.invoke(invocation)).willThrow(new RpcException(errorCode));
initParam();
Long count = 1L;
for (int i = 0; i < count; i++) {
try {
filter.invoke(invoker, invocation);
} catch (Exception e) {
Assertions.assertTrue(e instanceof RpcException);
filter.onError(e, invoker, invocation);
}
}
Map<String, MetricSample> metricsMap = getMetricsMap();
Assertions.assertTrue(metricsMap.containsKey(metricsKey.getName()));
MetricSample sample = metricsMap.get(metricsKey.getName());
Assertions.assertSame(((GaugeMetricSample) sample).getSupplier().get().longValue(), count);
Assertions.assertTrue(metricsMap.containsKey(metricsKey.getName()));
Map<String, String> tags = sample.getTags();
Assertions.assertEquals(tags.get(TAG_INTERFACE_KEY), INTERFACE_NAME);
Assertions.assertEquals(tags.get(TAG_METHOD_KEY), METHOD_NAME);
Assertions.assertEquals(tags.get(TAG_GROUP_KEY), GROUP);
Assertions.assertEquals(tags.get(TAG_VERSION_KEY), VERSION);
teardown();
}
@Test
void testMissingVersion() {
collector.setCollectEnabled(true);

197
dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/sampler/CountSamplerTest.java Normal file
View File

@ -0,0 +1,197 @@
/*
* 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.sampler;
import org.apache.dubbo.metrics.collector.sample.MetricsCountSampleConfigurer;
import org.apache.dubbo.metrics.collector.sample.SimpleMetricsCountSampler;
import org.apache.dubbo.metrics.model.MethodMetric;
import org.apache.dubbo.metrics.model.Metric;
import org.apache.dubbo.metrics.model.MetricsCategory;
import org.apache.dubbo.metrics.model.MetricsKey;
import org.apache.dubbo.metrics.model.sample.GaugeMetricSample;
import org.apache.dubbo.metrics.model.sample.MetricSample;
import org.jetbrains.annotations.NotNull;
import org.junit.Test;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import static org.apache.dubbo.metrics.model.MetricsCategory.RT;
public class CountSamplerTest {
public RequestMetricsCountSampler sampler = new RequestMetricsCountSampler();
@BeforeEach
public void before() {
sampler = new RequestMetricsCountSampler();
}
@Test
public void rtTest() {
String applicationName = "test";
sampler.addRT(applicationName, RTType.METHOD_REQUEST, 2L);
Map<String, GaugeMetricSample> collect = getCollect(RTType.METHOD_REQUEST);
Assertions.assertNotNull(collect);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_LAST.getName()) && collect.get(
MetricsKey.METRIC_RT_LAST.getName()).getSupplier().get().longValue() == 2);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_MIN.getName()) && collect.get(
MetricsKey.METRIC_RT_MIN.getName()).getSupplier().get().longValue() == 2);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_MAX.getName()) && collect.get(
MetricsKey.METRIC_RT_MAX.getName()).getSupplier().get().longValue() == 2);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_AVG.getName()) && collect.get(
MetricsKey.METRIC_RT_AVG.getName()).getSupplier().get().longValue() == 2);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_SUM.getName()) && collect.get(
MetricsKey.METRIC_RT_SUM.getName()).getSupplier().get().longValue() == 2);
sampler.addRT(applicationName, RTType.METHOD_REQUEST, 1L);
collect = getCollect(RTType.METHOD_REQUEST);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_LAST.getName()) && collect.get(
MetricsKey.METRIC_RT_LAST.getName()).getSupplier().get().longValue() == 1);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_MIN.getName()) && collect.get(
MetricsKey.METRIC_RT_MIN.getName()).getSupplier().get().longValue() == 1);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_MAX.getName()) && collect.get(
MetricsKey.METRIC_RT_MAX.getName()).getSupplier().get().longValue() == 2);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_AVG.getName()) && collect.get(
MetricsKey.METRIC_RT_AVG.getName()).getSupplier().get().longValue() == 1);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_SUM.getName()) && collect.get(
MetricsKey.METRIC_RT_SUM.getName()).getSupplier().get().longValue() == 3);
sampler.addRT(applicationName, RTType.APPLICATION, 4L);
collect = getCollect(RTType.APPLICATION);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_LAST.getName()) && collect.get(
MetricsKey.METRIC_RT_LAST.getName()).getSupplier().get().longValue() == 4);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_MIN.getName()) && collect.get(
MetricsKey.METRIC_RT_MIN.getName()).getSupplier().get().longValue() == 4);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_MAX.getName()) && collect.get(
MetricsKey.METRIC_RT_MAX.getName()).getSupplier().get().longValue() == 4);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_AVG.getName()) && collect.get(
MetricsKey.METRIC_RT_AVG.getName()).getSupplier().get().longValue() == 4);
Assertions.assertTrue(
null != collect.get(MetricsKey.METRIC_RT_SUM.getName()) && collect.get(
MetricsKey.METRIC_RT_SUM.getName()).getSupplier().get().longValue() == 4);
}
@NotNull
private Map<String, GaugeMetricSample> getCollect(RTType rtType) {
List<GaugeMetricSample> metricSamples = sampler.collectRT((key, metric, count) -> new GaugeMetricSample(key.formatName("consumer"), metric.getTags(), RT, () -> count),rtType);
Map<String, GaugeMetricSample> collect = metricSamples.stream()
.collect(Collectors.toMap(MetricSample::getName, v -> v));
return collect;
}
private GaugeMetricSample getGaugeMetricSample(MetricsKey metricsKey, MethodMetric methodMetric,
MetricsCategory metricsCategory, Supplier<Number> get) {
return new GaugeMetricSample(metricsKey.getNameByType(methodMetric.getSide()), metricsKey.getDescription(),
methodMetric.getTags(), metricsCategory, get);
}
public class RequestMetricsCountSampler
extends SimpleMetricsCountSampler<String, RTType, RequestMethodMetrics> {
@Override
public List<MetricSample> sample() {
return null;
}
@Override
protected void countConfigure(
MetricsCountSampleConfigurer<String, RTType, RequestMethodMetrics> sampleConfigure) {
sampleConfigure.configureMetrics(
configure -> new RequestMethodMetrics(configure.getSource()));
sampleConfigure.configureEventHandler(configure -> {
System.out.println("generic event");
});
}
@Override
public void rtConfigure(
MetricsCountSampleConfigurer<String, RTType, RequestMethodMetrics> sampleConfigure) {
sampleConfigure.configureMetrics(configure -> new RequestMethodMetrics(configure.getSource()));
sampleConfigure.configureEventHandler(configure -> {
System.out.println("rt event");
});
}
}
static enum RTType{
METHOD_REQUEST,
APPLICATION
}
static class RequestMethodMetrics implements Metric {
private String applicationName;
public RequestMethodMetrics(String applicationName) {
this.applicationName=applicationName;
}
@Override
public Map<String, String> getTags() {
Map<String,String> tags = new HashMap<>();
tags.put("serviceName", "test");
tags.put("version", "1.0.0");
tags.put("uptime", "20220202");
return tags;
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (!(o instanceof RequestMethodMetrics))
return false;
RequestMethodMetrics that = (RequestMethodMetrics) o;
return Objects.equals(applicationName, that.applicationName);
}
@Override
public int hashCode() {
return Objects.hash(applicationName);
}
}
}