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

add thread pool metric (#11461) 

* add thread pool metric,and run successfully

* add more thread pool metric,and run successfully

* delete comment

* avoid start import

* avoid start import and unused import

* avoid check

* ut pass

* Resolve unit test failures
The previous unit test run failed due to the introduction of thread pool metrics,
Unit tests now filter thread pool metrics

* Spin-off the applicationModel from the DefaultMetric Collector

---------

Co-authored-by: dongdong.yang <dongdong.yang@ly.com>
This commit is contained in:
Webster 2023-02-12 08:57:35 +08:00 committed by GitHub
parent 62e96dbcd0
commit cc32ebefa1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 209 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -22,6 +22,8 @@ public interface MetricsConstants {
String TAG_IP = "ip";
String TAG_PID = "pid";
String TAG_HOSTNAME = "hostname";
String TAG_APPLICATION_NAME = "application.name";
@ -77,4 +79,6 @@ public interface MetricsConstants {
String PROMETHEUS_DEFAULT_JOB_NAME = "default_dubbo_job";
String METRIC_FILTER_START_TIME = "metric_filter_start_time";
String TAG_THREAD_NAME = "thread.pool.name";
}

1
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java
View File

@ -366,6 +366,7 @@ public class DefaultApplicationDeployer extends AbstractDeployer<ApplicationMode
if (metricsConfig != null && PROTOCOL_PROMETHEUS.equals(metricsConfig.getProtocol())) {
collector.setCollectEnabled(true);
collector.addApplicationInfo(applicationModel.getApplicationName(), Version.getVersion());
collector.addThreadPool(applicationModel.getFrameworkModel(), applicationModel.getApplicationName());
String protocol = metricsConfig.getProtocol();
if (StringUtils.isNotEmpty(protocol)) {
MetricsReporterFactory metricsReporterFactory = getExtensionLoader(MetricsReporterFactory.class).getAdaptiveExtension();

1
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/model/MetricsCategory.java
View File

@ -24,5 +24,6 @@ public enum MetricsCategory {
RT,
QPS,
REQUESTS,
THREAD_POOL,
APPLICATION
}

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

@ -48,6 +48,13 @@ public enum MetricsKey {
PROVIDER_METRIC_RT_P99("dubbo.provider.rt.seconds.p99", "Response Time P99"),
PROVIDER_METRIC_RT_P95("dubbo.provider.rt.seconds.p95", "Response Time P95"),
THREAD_POOL_CORE_SIZE("dubbo.thread.pool.core.size","Thread Pool Core Size"),
THREAD_POOL_LARGEST_SIZE("dubbo.thread.pool.largest.size","Thread Pool Largest Size"),
THREAD_POOL_MAX_SIZE("dubbo.thread.pool.max.size","Thread Pool Max Size"),
THREAD_POOL_ACTIVE_SIZE("dubbo.thread.pool.active.size","Thread Pool Active Size"),
THREAD_POOL_THREAD_COUNT("dubbo.thread.pool.thread.count","Thread Pool Thread Count"),
THREAD_POOL_QUEUE_SIZE("dubbo.thread.pool.queue.size","Thread Pool Queue Size"),
// consumer metrics key
;

123
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/model/ThreadPoolMetric.java Normal file
View File

@ -0,0 +1,123 @@
/*
* 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.model;
import org.apache.dubbo.common.utils.ConfigUtils;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ThreadPoolExecutor;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_IP;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_PID;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_HOSTNAME;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_APPLICATION_NAME;
import static org.apache.dubbo.common.constants.MetricsConstants.TAG_THREAD_NAME;
import static org.apache.dubbo.common.utils.NetUtils.getLocalHost;
import static org.apache.dubbo.common.utils.NetUtils.getLocalHostName;
public class ThreadPoolMetric {
private String applicationName;
private String threadPoolName;
private ThreadPoolExecutor threadPoolExecutor;
public ThreadPoolMetric(String applicationName, String threadPoolName, ThreadPoolExecutor threadPoolExecutor) {
this.applicationName = applicationName;
this.threadPoolExecutor = threadPoolExecutor;
this.threadPoolName = threadPoolName;
}
public String getThreadPoolName() {
return threadPoolName;
}
public void setThreadPoolName(String threadPoolName) {
this.threadPoolName = threadPoolName;
}
public ThreadPoolExecutor getThreadPoolExecutor() {
return threadPoolExecutor;
}
public void setThreadPoolExecutor(ThreadPoolExecutor threadPoolExecutor) {
this.threadPoolExecutor = threadPoolExecutor;
}
public String getApplicationName() {
return applicationName;
}
public void setApplicationName(String applicationName) {
this.applicationName = applicationName;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
ThreadPoolMetric that = (ThreadPoolMetric) o;
return Objects.equals(applicationName, that.applicationName) &&
Objects.equals(threadPoolName, that.threadPoolName);
}
@Override
public int hashCode() {
return Objects.hash(applicationName, threadPoolName);
}
public Map<String, String> getTags() {
Map<String, String> tags = new HashMap<>();
tags.put(TAG_IP, getLocalHost());
tags.put(TAG_PID, ConfigUtils.getPid()+"");
tags.put(TAG_HOSTNAME, getLocalHostName());
tags.put(TAG_APPLICATION_NAME, applicationName);
tags.put(TAG_THREAD_NAME, threadPoolName);
return tags;
}
public double getCorePoolSize() {
return threadPoolExecutor.getCorePoolSize();
}
public double getLargestPoolSize() {
return threadPoolExecutor.getLargestPoolSize();
}
public double getMaximumPoolSize() {
return threadPoolExecutor.getMaximumPoolSize();
}
public double getActiveCount() {
return threadPoolExecutor.getActiveCount();
}
public double getPoolSize(){
return threadPoolExecutor.getPoolSize();
}
public double getQueueSize(){
return threadPoolExecutor.getQueue().size();
}
}

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

@ -17,6 +17,7 @@
package org.apache.dubbo.metrics.collector;
import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.metrics.collector.stat.MetricsStatComposite;
import org.apache.dubbo.metrics.collector.stat.MetricsStatHandler;
import org.apache.dubbo.metrics.event.EmptyEvent;
@ -24,17 +25,25 @@ import org.apache.dubbo.metrics.event.MetricsEvent;
import org.apache.dubbo.metrics.event.SimpleMetricsEventMulticaster;
import org.apache.dubbo.metrics.listener.MetricsListener;
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 org.apache.dubbo.rpc.model.FrameworkModel;
import org.apache.dubbo.rpc.Invocation;
import java.util.ArrayList;
import java.util.List;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import static org.apache.dubbo.metrics.model.MetricsCategory.THREAD_POOL;
import static org.apache.dubbo.metrics.model.MetricsCategory.APPLICATION;
import static org.apache.dubbo.metrics.model.MetricsCategory.REQUESTS;
import static org.apache.dubbo.metrics.model.MetricsCategory.RT;
@ -46,11 +55,12 @@ import static org.apache.dubbo.metrics.model.MetricsCategory.RT;
public class DefaultMetricsCollector implements MetricsCollector {
private AtomicBoolean collectEnabled = new AtomicBoolean(false);
private final Set<ThreadPoolMetric> threadPoolMetricSet = new HashSet<ThreadPoolMetric>();
private final MetricsStatComposite stats;
private final SimpleMetricsEventMulticaster eventMulticaster;
public DefaultMetricsCollector() {
this.stats = new MetricsStatComposite(this);
this.stats = new MetricsStatComposite( this);
this.eventMulticaster = SimpleMetricsEventMulticaster.getInstance();
}
@ -112,22 +122,46 @@ public class DefaultMetricsCollector implements MetricsCollector {
public void addApplicationInfo(String applicationName, String version) {
doExecute(MetricsEvent.Type.APPLICATION_INFO, statHandler -> statHandler.addApplication(applicationName,version));
}
public void addThreadPool(FrameworkModel frameworkModel, String applicationName) {
FrameworkExecutorRepository frameworkExecutorRepository =
frameworkModel.getBeanFactory().getBean(FrameworkExecutorRepository.class);
addThreadPoolExecutor(applicationName, "SharedExecutor", frameworkExecutorRepository.getSharedExecutor());
addThreadPoolExecutor(applicationName, "MappingRefreshingExecutor", frameworkExecutorRepository.getMappingRefreshingExecutor());
addThreadPoolExecutor(applicationName, "PoolRouterExecutor", frameworkExecutorRepository.getPoolRouterExecutor());
}
private void addThreadPoolExecutor(String applicationName, String threadPoolName, ExecutorService executorService) {
Optional<ExecutorService> executorOptional = Optional.ofNullable(executorService);
if (executorOptional.isPresent() && executorOptional.get() instanceof ThreadPoolExecutor ) {
threadPoolMetricSet.add(new ThreadPoolMetric(applicationName, threadPoolName,
(ThreadPoolExecutor) executorOptional.get()));
}
}
@Override
public List<MetricSample> collect() {
List<MetricSample> list = new ArrayList<>();
collectApplication(list);
collectRequests(list);
collectRT(list);
collectThreadPool(list);
return list;
}
private void collectThreadPool(List<MetricSample> list) {
threadPoolMetricSet.forEach(e -> list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_CORE_SIZE, e.getTags(), THREAD_POOL, e::getCorePoolSize)));
threadPoolMetricSet.forEach(e -> list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_LARGEST_SIZE, e.getTags(), THREAD_POOL, e::getLargestPoolSize)));
threadPoolMetricSet.forEach(e -> list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_MAX_SIZE, e.getTags(), THREAD_POOL, e::getMaximumPoolSize)));
threadPoolMetricSet.forEach(e -> list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_ACTIVE_SIZE, e.getTags(), THREAD_POOL, e::getActiveCount)));
threadPoolMetricSet.forEach(e -> list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_THREAD_COUNT, e.getTags(), THREAD_POOL, e::getPoolSize)));
threadPoolMetricSet.forEach(e -> list.add(new GaugeMetricSample(MetricsKey.THREAD_POOL_QUEUE_SIZE, e.getTags(), THREAD_POOL, e::getQueueSize)));
}
private void collectApplication(List<MetricSample> list) {
doCollect(MetricsEvent.Type.APPLICATION_INFO, MetricsStatHandler::get).filter(e -> !e.isEmpty())
.ifPresent(map -> map.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.APPLICATION_METRIC_INFO, k.getTags(),
APPLICATION, v::get))));
}
private void collectRequests(List<MetricSample> list) {

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

@ -17,10 +17,13 @@
package org.apache.dubbo.metrics.filter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.dubbo.common.constants.MetricsConstants;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.metrics.collector.DefaultMetricsCollector;
import org.apache.dubbo.metrics.model.MetricsKey;
@ -322,6 +325,13 @@ class MetricsFilterTest {
private Map<String, MetricSample> getMetricsMap() {
List<MetricSample> samples = collector.collect();
return samples.stream().collect(Collectors.toMap(MetricSample::getName, Function.identity()));
List<MetricSample> samples1 = new ArrayList<>();
for (MetricSample sample : samples) {
if (sample.getName().contains("dubbo.thread.pool")) {
continue;
}
samples1.add(sample);
}
return samples1.stream().collect(Collectors.toMap(MetricSample::getName, Function.identity()));
}
}

27
dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/metrics/collector/DefaultMetricsCollectorTest.java
View File

@ -34,6 +34,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
@ -55,6 +56,7 @@ class DefaultMetricsCollectorTest {
private String group;
private String version;
private RpcInvocation invocation;
public static final String DUBBO_THREAD_METRIC_MARK = "dubbo.thread.pool";
@BeforeEach
public void setup() {
@ -93,6 +95,9 @@ class DefaultMetricsCollectorTest {
List<MetricSample> samples = collector.collect();
for (MetricSample sample : samples) {
if (sample.getName().contains(DUBBO_THREAD_METRIC_MARK)) {
continue;
}
Assertions.assertTrue(sample instanceof GaugeMetricSample);
GaugeMetricSample gaugeSample = (GaugeMetricSample) sample;
Map<String, String> tags = gaugeSample.getTags();
@ -107,7 +112,14 @@ class DefaultMetricsCollectorTest {
collector.decreaseProcessingRequests(applicationName, invocation);
samples = collector.collect();
Map<String, Long> sampleMap = samples.stream().collect(Collectors.toMap(MetricSample::getName, k -> {
List<MetricSample> samples1 = new ArrayList<>();
for (MetricSample sample : samples) {
if (sample.getName().contains(DUBBO_THREAD_METRIC_MARK)) {
continue;
}
samples1.add(sample);
}
Map<String, Long> sampleMap = samples1.stream().collect(Collectors.toMap(MetricSample::getName, k -> {
Number number = ((GaugeMetricSample) k).getSupplier().get();
return number.longValue();
}));
@ -125,6 +137,9 @@ class DefaultMetricsCollectorTest {
List<MetricSample> samples = collector.collect();
for (MetricSample sample : samples) {
if (sample.getName().contains(DUBBO_THREAD_METRIC_MARK)) {
continue;
}
Map<String, String> tags = sample.getTags();
Assertions.assertEquals(tags.get(TAG_INTERFACE_KEY), interfaceName);
@ -132,8 +147,14 @@ class DefaultMetricsCollectorTest {
Assertions.assertEquals(tags.get(TAG_GROUP_KEY), group);
Assertions.assertEquals(tags.get(TAG_VERSION_KEY), version);
}
Map<String, Long> sampleMap = samples.stream().collect(Collectors.toMap(MetricSample::getName, k -> {
List<MetricSample> samples1 = new ArrayList<>();
for (MetricSample sample : samples) {
if (sample.getName().contains(DUBBO_THREAD_METRIC_MARK)) {
continue;
}
samples1.add(sample);
}
Map<String, Long> sampleMap = samples1.stream().collect(Collectors.toMap(MetricSample::getName, k -> {
Number number = ((GaugeMetricSample) k).getSupplier().get();
return number.longValue();
}));