Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

✨ add framework metrics #12576

Merged
merged 5 commits into from
Jun 28, 2023
Merged

✨ add framework metrics #12576

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
17ee8a1
add framework metric
songxiaosheng Jun 20, 2023
55bc0aa
:sparkles: add framework metrics
songxiaosheng Jun 20, 2023
d981ff3
Merge branch '3.2' into feature/add_framework_executor_metric
songxiaosheng Jun 20, 2023
4343526
:sparkles: add framework metrics
songxiaosheng Jun 25, 2023
6f97f83
Merge branch 'feature/add_framework_executor_metric' of github.com:so…
songxiaosheng Jun 25, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 38 additions & 20 deletions ...ult/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadPoolMetricsSampler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ public class ThreadPoolMetricsSampler implements MetricsSampler {
private DataStore dataStore;
private final Map<String, ThreadPoolExecutor> sampleThreadPoolExecutor = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, ThreadPoolMetric> threadPoolMetricMap = new ConcurrentHashMap<>();

public ThreadPoolMetricsSampler(DefaultMetricsCollector collector) {
this.collector = collector;
}
Expand Down Expand Up @@ -87,23 +88,29 @@ private List<MetricSample> createMetricsSample(String name, ThreadPoolExecutor e
list.add(new GaugeMetricSample<>(MetricsKey.THREAD_POOL_ACTIVE_SIZE, threadPoolMetric.getTags(), THREAD_POOL, threadPoolMetric, ThreadPoolMetric::getActiveCount));
list.add(new GaugeMetricSample<>(MetricsKey.THREAD_POOL_THREAD_COUNT, threadPoolMetric.getTags(), THREAD_POOL, threadPoolMetric, ThreadPoolMetric::getPoolSize));
list.add(new GaugeMetricSample<>(MetricsKey.THREAD_POOL_QUEUE_SIZE, threadPoolMetric.getTags(), THREAD_POOL, threadPoolMetric, ThreadPoolMetric::getQueueSize));

return list;
}

public void registryDefaultSampleThreadPoolExecutor() {
ApplicationModel applicationModel = collector.getApplicationModel();
if (applicationModel == null) {
return;
}
try {
if (this.frameworkExecutorRepository == null) {
this.frameworkExecutorRepository = collector.getApplicationModel().getBeanFactory()
.getBean(FrameworkExecutorRepository.class);
addRpcExecutors();
addFrameworkExecutors();
addExecutorRejectMetrics();
}

private void addExecutorRejectMetrics() {
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);
((AbortPolicyWithReport) entry.getValue().getRejectedExecutionHandler()).addThreadPoolExhaustedEventListener(metricThreadPoolExhaustedListener);
}
} catch (Exception ex) {
logger.warn(COMMON_METRICS_COLLECTOR_EXCEPTION, "", "", "ThreadPoolMetricsSampler! frameworkExecutorRepository non-init");
}
});
}

private void addRpcExecutors() {
if (this.dataStore == null) {
this.dataStore = collector.getApplicationModel().getExtensionLoader(DataStore.class).getDefaultExtension();
}
Expand All @@ -113,7 +120,7 @@ public void registryDefaultSampleThreadPoolExecutor() {
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);
Expand All @@ -123,18 +130,29 @@ public void registryDefaultSampleThreadPoolExecutor() {
this.addExecutors(CLIENT_THREAD_POOL_NAME + "-" + entry.getKey(), executor);
}
}
}
}

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);
((AbortPolicyWithReport) entry.getValue().getRejectedExecutionHandler()).addThreadPoolExhaustedEventListener(metricThreadPoolExhaustedListener);
}
});
private void addFrameworkExecutors() {
try {
if (this.frameworkExecutorRepository == null) {
this.frameworkExecutorRepository = collector.getApplicationModel().getBeanFactory()
.getBean(FrameworkExecutorRepository.class);
}
} catch (Exception ex) {
logger.warn(COMMON_METRICS_COLLECTOR_EXCEPTION, "", "", "ThreadPoolMetricsSampler! frameworkExecutorRepository non-init");
}
if (this.frameworkExecutorRepository != null) {
this.addExecutors("sharedExecutor", frameworkExecutorRepository.getSharedExecutor());
if (this.frameworkExecutorRepository == null) {
return;
}
this.addExecutors("poolRouterExecutor", frameworkExecutorRepository.getPoolRouterExecutor());
this.addExecutors("metadataRetryExecutor", frameworkExecutorRepository.getMetadataRetryExecutor());
this.addExecutors("internalServiceExecutor", frameworkExecutorRepository.getInternalServiceExecutor());
this.addExecutors("connectivityScheduledExecutor", frameworkExecutorRepository.getConnectivityScheduledExecutor());
this.addExecutors("cacheRefreshingScheduledExecutor", frameworkExecutorRepository.getCacheRefreshingScheduledExecutor());
this.addExecutors("sharedExecutor", frameworkExecutorRepository.getSharedExecutor());
this.addExecutors("sharedScheduledExecutor", frameworkExecutorRepository.getSharedScheduledExecutor());
this.addExecutors("mappingRefreshingExecutor", frameworkExecutorRepository.getMappingRefreshingExecutor());
}

}

2 changes: 1 addition & 1 deletion ...src/test/java/org/apache/dubbo/metrics/collector/sample/ThreadPoolMetricsSamplerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,7 +169,7 @@ public void testRegistryDefaultSampleThreadPoolExecutor() throws NoSuchFieldExce
f.setAccessible(true);
Map<String,ThreadPoolExecutor> executors = (Map<String, ThreadPoolExecutor>) f.get(sampler2);

Assertions.assertEquals(3, executors.size());
Assertions.assertEquals(8, executors.size());
Assertions.assertTrue(executors.containsKey("DubboServerHandler-server1"));
Assertions.assertTrue(executors.containsKey("DubboClientHandler-client1"));
Assertions.assertTrue(executors.containsKey("sharedExecutor"));
Expand Down