OneSizeFitsQuorum commented on code in PR #9833:
URL: https://github.com/apache/iotdb/pull/9833#discussion_r1211180075
##########
server/src/main/java/org/apache/iotdb/db/service/metrics/ProcessMetrics.java:
##########
@@ -61,15 +65,27 @@ private void collectProcessCpuInfo(AbstractMetricService
metricService) {
Metric.PROCESS_CPU_LOAD.toString(),
MetricLevel.CORE,
sunOsMxBean,
- a -> (long) (sunOsMxBean.getProcessCpuLoad() * 100),
+ a -> {
+ if (System.currentTimeMillis() - lastUpdateTime > UPDATE_INTERVAL) {
+ lastUpdateTime = System.currentTimeMillis();
+ processCpuLoad = (long) (sunOsMxBean.getProcessCpuLoad() * 100);
+ }
+ return processCpuLoad;
+ },
Tag.NAME.toString(),
"process");
metricService.createAutoGauge(
Metric.PROCESS_CPU_TIME.toString(),
MetricLevel.CORE,
sunOsMxBean,
- com.sun.management.OperatingSystemMXBean::getProcessCpuTime,
+ bean -> {
Review Comment:
same
##########
server/src/main/java/org/apache/iotdb/db/service/metrics/ProcessMetrics.java:
##########
@@ -61,15 +65,27 @@ private void collectProcessCpuInfo(AbstractMetricService
metricService) {
Metric.PROCESS_CPU_LOAD.toString(),
MetricLevel.CORE,
sunOsMxBean,
- a -> (long) (sunOsMxBean.getProcessCpuLoad() * 100),
+ a -> {
Review Comment:
In this case, it looks like every time Prometheus pull the value of the
monitored item, one of processCpuLoad and processCpuTime will be updated, which
might cause an accuracy error? It feels like if there were only two monitors,
it would be fine to leave things as they are
##########
server/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java:
##########
@@ -31,23 +34,43 @@
import org.apache.iotdb.db.mpp.metric.QueryResourceMetricSet;
import org.apache.iotdb.db.mpp.metric.SeriesScanCostMetricSet;
import org.apache.iotdb.metrics.metricsets.UpTimeMetrics;
+import org.apache.iotdb.metrics.metricsets.cpu.CpuUsageMetrics;
import org.apache.iotdb.metrics.metricsets.disk.DiskMetrics;
import org.apache.iotdb.metrics.metricsets.jvm.JvmMetrics;
import org.apache.iotdb.metrics.metricsets.logback.LogbackMetrics;
import org.apache.iotdb.metrics.metricsets.net.NetMetrics;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
public class DataNodeMetricsHelper {
/** Bind predefined metric sets into DataNode. */
public static void bind() {
MetricService.getInstance().addMetricSet(new UpTimeMetrics());
MetricService.getInstance().addMetricSet(new JvmMetrics());
+ MetricService.getInstance().addMetricSet(ThreadPoolMetrics.getInstance());
MetricService.getInstance().addMetricSet(new LogbackMetrics());
MetricService.getInstance().addMetricSet(new FileMetrics());
MetricService.getInstance().addMetricSet(CompactionMetrics.getInstance());
MetricService.getInstance().addMetricSet(new ProcessMetrics());
MetricService.getInstance().addMetricSet(new SystemMetrics(true));
MetricService.getInstance().addMetricSet(new
DiskMetrics(IoTDBConstant.DN_ROLE));
MetricService.getInstance().addMetricSet(new
NetMetrics(IoTDBConstant.DN_ROLE));
+ List<String> threadModules = new ArrayList<>();
Review Comment:
Extract this code into a separate function to keep `bind` clear?
##########
server/src/main/java/org/apache/iotdb/db/service/metrics/DataNodeMetricsHelper.java:
##########
@@ -31,23 +34,43 @@
import org.apache.iotdb.db.mpp.metric.QueryResourceMetricSet;
import org.apache.iotdb.db.mpp.metric.SeriesScanCostMetricSet;
import org.apache.iotdb.metrics.metricsets.UpTimeMetrics;
+import org.apache.iotdb.metrics.metricsets.cpu.CpuUsageMetrics;
import org.apache.iotdb.metrics.metricsets.disk.DiskMetrics;
import org.apache.iotdb.metrics.metricsets.jvm.JvmMetrics;
import org.apache.iotdb.metrics.metricsets.logback.LogbackMetrics;
import org.apache.iotdb.metrics.metricsets.net.NetMetrics;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
public class DataNodeMetricsHelper {
/** Bind predefined metric sets into DataNode. */
public static void bind() {
MetricService.getInstance().addMetricSet(new UpTimeMetrics());
MetricService.getInstance().addMetricSet(new JvmMetrics());
+ MetricService.getInstance().addMetricSet(ThreadPoolMetrics.getInstance());
MetricService.getInstance().addMetricSet(new LogbackMetrics());
MetricService.getInstance().addMetricSet(new FileMetrics());
MetricService.getInstance().addMetricSet(CompactionMetrics.getInstance());
MetricService.getInstance().addMetricSet(new ProcessMetrics());
MetricService.getInstance().addMetricSet(new SystemMetrics(true));
MetricService.getInstance().addMetricSet(new
DiskMetrics(IoTDBConstant.DN_ROLE));
MetricService.getInstance().addMetricSet(new
NetMetrics(IoTDBConstant.DN_ROLE));
+ List<String> threadModules = new ArrayList<>();
+ Arrays.stream(DataNodeThreadModule.values()).forEach(x ->
threadModules.add(x.toString()));
+ List<String> pools = new ArrayList<>();
+ Arrays.stream(ThreadName.values()).forEach(x -> pools.add(x.name()));
+ MetricService.getInstance()
+ .addMetricSet(
+ new CpuUsageMetrics(
+ threadModules,
+ pools,
+ x -> ThreadName.getModuleTheThreadBelongs(x).toString(),
Review Comment:
Will this be `unknown`?
##########
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadScheduledExecutor.java:
##########
@@ -19,57 +19,70 @@
package org.apache.iotdb.commons.concurrent.threadpool;
+import org.apache.iotdb.commons.concurrent.ThreadPoolMetrics;
import org.apache.iotdb.commons.concurrent.WrappedCallable;
import org.apache.iotdb.commons.concurrent.WrappedRunnable;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.service.JMXService;
import java.util.Collection;
+import java.util.LinkedList;
import java.util.List;
+import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
public class WrappedSingleThreadScheduledExecutor
implements ScheduledExecutorService,
WrappedSingleThreadScheduledExecutorMBean {
private final String mbeanName;
ScheduledExecutorService service;
+ private final AtomicInteger taskCount = new AtomicInteger(0);
Review Comment:
we can directly copy WrappedScheduledExecutorService's design
##########
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadExecutorService.java:
##########
@@ -77,50 +84,120 @@ public boolean awaitTermination(long timeout, TimeUnit
unit) throws InterruptedE
@Override
public <T> Future<T> submit(Callable<T> task) {
- return service.submit(WrappedCallable.wrap(task));
+ taskCount.incrementAndGet();
+ return service.submit(WrappedCallable.wrapWithCount(task, runCount));
}
@Override
public <T> Future<T> submit(Runnable task, T result) {
- return service.submit(WrappedRunnable.wrap(task), result);
+ taskCount.incrementAndGet();
+ return service.submit(WrappedRunnable.wrapWithCount(task, runCount),
result);
}
@Override
public Future<?> submit(Runnable task) {
- return service.submit(WrappedRunnable.wrap(task));
+ taskCount.incrementAndGet();
+ return service.submit(WrappedRunnable.wrapWithCount(task, runCount));
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
+ taskCount.addAndGet(tasks.size());
return service.invokeAll(
-
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()));
+ tasks.stream()
+ .map(x -> WrappedCallable.wrapWithCount(x, runCount))
+ .collect(Collectors.toList()));
}
@Override
public <T> List<Future<T>> invokeAll(
Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
+ taskCount.addAndGet(tasks.size());
return service.invokeAll(
-
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()),
timeout, unit);
+ tasks.stream()
+ .map(x -> WrappedCallable.wrapWithCount(x, runCount))
+ .collect(Collectors.toList()),
+ timeout,
+ unit);
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
+ taskCount.addAndGet(tasks.size());
return service.invokeAny(
-
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()));
+ tasks.stream()
+ .map(x -> WrappedCallable.wrapWithCount(x, runCount))
+ .collect(Collectors.toList()));
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long
timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
+ taskCount.addAndGet(tasks.size());
return service.invokeAny(
-
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()),
timeout, unit);
+ tasks.stream()
+ .map(x -> WrappedCallable.wrapWithCount(x, runCount))
+ .collect(Collectors.toList()),
+ timeout,
+ unit);
}
@Override
public void execute(Runnable command) {
- service.execute(WrappedRunnable.wrap(command));
+ taskCount.incrementAndGet();
+ service.execute(WrappedRunnable.wrapWithCount(command, runCount));
+ }
+
+ @Override
+ public int getCorePoolSize() {
+ return 1;
+ }
+
+ @Override
+ public boolean prestartCoreThread() {
+ return false;
+ }
+
+ @Override
+ public int getMaximumPoolSize() {
+ return 1;
+ }
+
+ @Override
+ public Queue<Runnable> getQueue() {
+ return new LinkedList<>();
Review Comment:
seems weird. How about just copying `WrappedScheduledExecutorService`'s
design
##########
metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/cpu/CpuUsageMetrics.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.iotdb.metrics.metricsets.cpu;
+
+import org.apache.iotdb.metrics.AbstractMetricService;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.metrics.utils.MetricType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.UnaryOperator;
+
+public class CpuUsageMetrics implements IMetricSet {
+ private static final Logger log =
LoggerFactory.getLogger(CpuUsageMetrics.class);
+ private static final String MODULE_CPU_USAGE = "module_cpu_usage";
+ private static final String POOL_CPU_USAGE = "pool_cpu_usage";
+ private static final String POOL = "pool";
+ private static final String MODULE = "module";
+ private static final String MODULE_USER_TIME_PERCENTAGE =
"module_user_time_percentage";
+ private static final String POOL_USER_TIME_PERCENTAGE =
"user_time_percentage";
+ private final List<String> modules;
+ private final List<String> pools;
+ private static final long UPDATE_INTERVAL = 10_000L;
Review Comment:
Put Into a more basic class
##########
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedCallable.java:
##########
@@ -47,4 +48,20 @@ public V callMayThrow() throws Exception {
}
};
}
+
+ public static <V> Callable<V> wrapWithCount(Callable<V> callable,
AtomicInteger count) {
+ if (callable instanceof WrappedCallable) {
+ return callable;
+ }
+ return new WrappedCallable<V>() {
+ @Override
+ public V callMayThrow() throws Exception {
+ try {
+ return callable.call();
+ } finally {
+ count.incrementAndGet();
Review Comment:
Maybe we don't have to record it ourselves
##########
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java:
##########
@@ -84,4 +275,106 @@ public enum ThreadName {
public String getName() {
return name;
}
+
+ @SuppressWarnings("java:S6541")
+ public static DataNodeThreadModule getModuleTheThreadBelongs(String
givenThreadName) {
+ for (ThreadName threadName : queryThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.QUERY;
+ }
+ }
+ if
(givenThreadName.contains(MPP_COORDINATOR_SCHEDULED_EXECUTOR.getName())) {
+ return DataNodeThreadModule.MPP_SCHEDULE;
+ }
+ for (ThreadName threadName : compactionThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.COMPACTION;
+ }
+ }
+ for (ThreadName threadName : walThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.WAL;
+ }
+ }
+ if (givenThreadName.contains(MPP_COORDINATOR_WRITE_EXECUTOR.getName())) {
+ return DataNodeThreadModule.MPP_WRITE;
+ }
+ for (ThreadName threadName : flushThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.FLUSH;
+ }
+ }
+ for (ThreadName threadName : schemaEngineThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.SCHEMA_ENGINE;
+ }
+ }
+ for (ThreadName threadName : clientServiceThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.CLIENT_SERVICE;
+ }
+ }
+ for (ThreadName threadName : iotConsensusThrreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.IOT_CONSENSUS;
+ }
+ }
+ for (ThreadName threadName : ratisThreadNames) {
+ if (threadName.getName().contains("\\d")) {
+ if
(Pattern.compile(threadName.getName()).matcher(givenThreadName).find()) {
+ return DataNodeThreadModule.RATIS_CONSENSUS;
+ }
+ } else if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.RATIS_CONSENSUS;
+ }
+ }
+ for (ThreadName threadName : computeThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.COMPUTE;
+ }
+ }
+ for (ThreadName threadName : syncThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.SYNC;
+ }
+ }
+ for (ThreadName threadName : jvmThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.JVM;
+ }
+ }
+ if (givenThreadName.contains(LOG_BACK.getName())) {
+ return DataNodeThreadModule.LOG_BACK;
+ }
+ for (ThreadName threadName : metricsThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.METRICS;
+ }
+ }
+ for (ThreadName threadName : otherThreadNames) {
+ if (givenThreadName.contains(threadName.getName())) {
+ return DataNodeThreadModule.OTHER;
+ }
+ }
+ log.error("Unknown thread name {}", givenThreadName);
Review Comment:
Using debug log and printing the log only means that the metric module has
encountered a thread that is not known in advance when counting threads, and
the issue does not affect the system's operation
##########
server/src/main/java/org/apache/iotdb/db/engine/flush/pool/FlushSubTaskPoolManager.java:
##########
@@ -52,8 +51,7 @@ public String getName() {
@Override
public void start() {
if (pool == null) {
Review Comment:
when will this pool be null?
##########
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadExecutorService.java:
##########
@@ -19,33 +19,40 @@
package org.apache.iotdb.commons.concurrent.threadpool;
+import org.apache.iotdb.commons.concurrent.ThreadPoolMetrics;
import org.apache.iotdb.commons.concurrent.WrappedCallable;
import org.apache.iotdb.commons.concurrent.WrappedRunnable;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.service.JMXService;
import java.util.Collection;
+import java.util.LinkedList;
import java.util.List;
+import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
public class WrappedSingleThreadExecutorService
implements ExecutorService, WrappedSingleThreadExecutorServiceMBean {
private final String mbeanName;
ExecutorService service;
+ private final AtomicInteger taskCount = new AtomicInteger(0);
Review Comment:
we can directly copy `WrappedScheduledExecutorService`'s design
##########
server/src/main/java/org/apache/iotdb/db/engine/flush/pool/FlushTaskPoolManager.java:
##########
@@ -53,8 +53,7 @@ public String getName() {
public void start() {
if (pool == null) {
Review Comment:
when will this pool be null?
##########
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/threadpool/WrappedSingleThreadScheduledExecutor.java:
##########
@@ -101,50 +114,112 @@ public boolean awaitTermination(long timeout, TimeUnit
unit) throws InterruptedE
@Override
public <T> Future<T> submit(Callable<T> task) {
- return service.submit(WrappedCallable.wrap(task));
+ return service.submit(WrappedCallable.wrapWithCount(task, runCount));
}
@Override
public <T> Future<T> submit(Runnable task, T result) {
- return service.submit(WrappedRunnable.wrap(task), result);
+ return service.submit(WrappedRunnable.wrapWithCount(task, runCount),
result);
}
@Override
public Future<?> submit(Runnable task) {
- return service.submit(WrappedRunnable.wrap(task));
+ return service.submit(WrappedRunnable.wrapWithCount(task, runCount));
}
@Override
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {
return service.invokeAll(
-
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()));
+ tasks.stream()
+ .map(x -> WrappedCallable.wrapWithCount(x, runCount))
+ .collect(Collectors.toList()));
}
@Override
public <T> List<Future<T>> invokeAll(
Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException {
return service.invokeAll(
-
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()),
timeout, unit);
+ tasks.stream()
+ .map(x -> WrappedCallable.wrapWithCount(x, runCount))
+ .collect(Collectors.toList()),
+ timeout,
+ unit);
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {
return service.invokeAny(
-
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()));
+ tasks.stream()
+ .map(x -> WrappedCallable.wrapWithCount(x, runCount))
+ .collect(Collectors.toList()));
}
@Override
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long
timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
return service.invokeAny(
-
tasks.stream().map(WrappedCallable::wrap).collect(Collectors.toList()),
timeout, unit);
+ tasks.stream()
+ .map(x -> WrappedCallable.wrapWithCount(x, runCount))
+ .collect(Collectors.toList()),
+ timeout,
+ unit);
}
@Override
public void execute(Runnable command) {
- service.execute(WrappedRunnable.wrap(command));
+ service.execute(WrappedRunnable.wrapWithCount(command, runCount));
+ }
+
+ @Override
+ public int getCorePoolSize() {
+ return 1;
+ }
+
+ @Override
+ public boolean prestartCoreThread() {
+ return false;
+ }
+
+ @Override
+ public int getMaximumPoolSize() {
+ return 1;
+ }
+
+ @Override
+ public Queue<Runnable> getQueue() {
+ return new LinkedList<>();
Review Comment:
seems weird. How about just copying WrappedScheduledExecutorService's design
##########
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/WrappedRunnable.java:
##########
@@ -45,4 +47,20 @@ public void runMayThrow() {
}
};
}
+
+ public static Runnable wrapWithCount(Runnable runnable, AtomicInteger count)
{
+ if (runnable instanceof WrappedRunnable) {
+ return runnable;
+ }
+ return new WrappedRunnable() {
+ @Override
+ public void runMayThrow() {
+ try {
+ runnable.run();
+ } finally {
+ count.incrementAndGet();
Review Comment:
Maybe we don't have to record it ourselves
##########
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadPoolMetrics.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.iotdb.commons.concurrent;
+
+import org.apache.iotdb.commons.concurrent.threadpool.IThreadPoolMBean;
+import org.apache.iotdb.metrics.AbstractMetricService;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.metrics.utils.MetricType;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class ThreadPoolMetrics implements IMetricSet {
+ private static final String THREAD_POOL_ACTIVE_THREAD_COUNT =
"thread_pool_active_thread_count";
+ private static final String THREAD_POOL_DONE_TASK_COUNT =
"thread_pool_done_task_count";
+ private static final String THREAD_POOL_WAITING_TASK_COUNT =
"thread_pool_waiting_task_count";
+ private static final String THREAD_POOL_CORE_SIZE = "thread_pool_core_size";
+ private static final String POOL_NAME = "pool_name";
+ private AbstractMetricService metricService;
+ private Map<String, IThreadPoolMBean> notRegisteredPoolMap = new HashMap<>();
+ private Map<String, IThreadPoolMBean> registeredPoolMap = new HashMap<>();
+
+ public static ThreadPoolMetrics getInstance() {
+ return ThreadPoolMetricsHolder.INSTANCE;
+ }
+
+ private ThreadPoolMetrics() {}
+
+ public void registerThreadPool(IThreadPoolMBean pool, String name) {
+ synchronized (this) {
+ if (metricService == null) {
+ notRegisteredPoolMap.put(name, pool);
+ } else {
+ registeredPoolMap.put(name, pool);
+ metricService.createAutoGauge(
+ THREAD_POOL_ACTIVE_THREAD_COUNT,
+ MetricLevel.IMPORTANT,
+ registeredPoolMap,
+ map -> registeredPoolMap.get(name).getActiveCount(),
+ POOL_NAME,
+ name);
+ metricService.createAutoGauge(
+ THREAD_POOL_CORE_SIZE,
+ MetricLevel.IMPORTANT,
+ registeredPoolMap,
+ map -> registeredPoolMap.get(name).getCorePoolSize(),
Review Comment:
How about add `LargestPoolSize`
##########
node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadPoolMetrics.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.iotdb.commons.concurrent;
+
+import org.apache.iotdb.commons.concurrent.threadpool.IThreadPoolMBean;
+import org.apache.iotdb.metrics.AbstractMetricService;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.metrics.utils.MetricType;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class ThreadPoolMetrics implements IMetricSet {
+ private static final String THREAD_POOL_ACTIVE_THREAD_COUNT =
"thread_pool_active_thread_count";
+ private static final String THREAD_POOL_DONE_TASK_COUNT =
"thread_pool_done_task_count";
+ private static final String THREAD_POOL_WAITING_TASK_COUNT =
"thread_pool_waiting_task_count";
+ private static final String THREAD_POOL_CORE_SIZE = "thread_pool_core_size";
+ private static final String POOL_NAME = "pool_name";
+ private AbstractMetricService metricService;
+ private Map<String, IThreadPoolMBean> notRegisteredPoolMap = new HashMap<>();
+ private Map<String, IThreadPoolMBean> registeredPoolMap = new HashMap<>();
+
+ public static ThreadPoolMetrics getInstance() {
+ return ThreadPoolMetricsHolder.INSTANCE;
+ }
+
+ private ThreadPoolMetrics() {}
+
+ public void registerThreadPool(IThreadPoolMBean pool, String name) {
+ synchronized (this) {
+ if (metricService == null) {
+ notRegisteredPoolMap.put(name, pool);
+ } else {
+ registeredPoolMap.put(name, pool);
+ metricService.createAutoGauge(
+ THREAD_POOL_ACTIVE_THREAD_COUNT,
+ MetricLevel.IMPORTANT,
+ registeredPoolMap,
+ map -> registeredPoolMap.get(name).getActiveCount(),
+ POOL_NAME,
+ name);
+ metricService.createAutoGauge(
Review Comment:
How about add `MaximumPoolSize`
##########
metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/cpu/CpuUsageMetrics.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.iotdb.metrics.metricsets.cpu;
+
+import org.apache.iotdb.metrics.AbstractMetricService;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.metrics.utils.MetricType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.UnaryOperator;
+
+public class CpuUsageMetrics implements IMetricSet {
+ private static final Logger log =
LoggerFactory.getLogger(CpuUsageMetrics.class);
+ private static final String MODULE_CPU_USAGE = "module_cpu_usage";
+ private static final String POOL_CPU_USAGE = "pool_cpu_usage";
+ private static final String POOL = "pool";
+ private static final String MODULE = "module";
+ private static final String MODULE_USER_TIME_PERCENTAGE =
"module_user_time_percentage";
+ private static final String POOL_USER_TIME_PERCENTAGE =
"user_time_percentage";
+ private final List<String> modules;
+ private final List<String> pools;
+ private static final long UPDATE_INTERVAL = 10_000L;
+ protected AbstractMetricService metricService;
+ protected final UnaryOperator<String> threadNameToModule;
+ protected final UnaryOperator<String> threadNameToPool;
+ protected final Map<Long, String> threadIdToModuleCache = new HashMap<>();
+ protected final Map<Long, String> threadIdToPoolCache = new HashMap<>();
+ private final Map<String, Double> moduleCpuTimePercentageMap = new
HashMap<>();
+ private final Map<String, Double> moduleUserTimePercentageMap = new
HashMap<>();
+ private final Map<String, Double> poolCpuUsageMap = new HashMap<>();
+ private final Map<String, Double> poolUserTimePercentageMap = new
HashMap<>();
+ private final Map<Long, Long> lastThreadCpuTime = new HashMap<>();
+ private final Map<Long, Long> lastThreadUserTime = new HashMap<>();
+ AutoGauge processCpuLoadGauge = null;
+ private final ThreadMXBean threadMxBean =
ManagementFactory.getThreadMXBean();
+ private AtomicLong lastUpdateTime = new AtomicLong(0L);
+
+ public CpuUsageMetrics(
+ List<String> modules,
+ List<String> pools,
+ UnaryOperator<String> threadNameToModule,
+ UnaryOperator<String> threadNameToPool) {
+ this.modules = modules;
+ this.pools = pools;
+ this.threadNameToModule = threadNameToModule;
+ this.threadNameToPool = threadNameToPool;
+ }
+
+ @Override
+ public void bindTo(AbstractMetricService metricService) {
+ this.metricService = metricService;
+ for (String moduleName : modules) {
+ metricService.createAutoGauge(
+ MODULE_CPU_USAGE,
+ MetricLevel.IMPORTANT,
+ this,
+ x -> x.getModuleCpuUsage().getOrDefault(moduleName, 0.0),
+ MODULE,
+ moduleName);
+ metricService.createAutoGauge(
+ MODULE_USER_TIME_PERCENTAGE,
+ MetricLevel.IMPORTANT,
+ this,
+ x -> x.getModuleUserTimePercentage().getOrDefault(moduleName, 0.0),
+ MODULE,
+ moduleName);
+ }
+ for (String poolName : pools) {
+ metricService.createAutoGauge(
+ POOL_CPU_USAGE,
+ MetricLevel.IMPORTANT,
+ this,
+ x -> x.getPoolCpuUsage().getOrDefault(poolName, 0.0),
+ POOL,
+ poolName);
+ metricService.createAutoGauge(
+ POOL_USER_TIME_PERCENTAGE,
+ MetricLevel.IMPORTANT,
+ this,
+ x -> x.getPoolUserCpuPercentage().getOrDefault(poolName, 0.0),
+ POOL,
+ poolName);
+ }
+ }
+
+ @Override
+ public void unbindFrom(AbstractMetricService metricService) {
+ for (String moduleName : modules) {
+ metricService.remove(MetricType.AUTO_GAUGE, MODULE_CPU_USAGE, MODULE,
moduleName);
+ metricService.remove(MetricType.AUTO_GAUGE, MODULE_USER_TIME_PERCENTAGE,
MODULE, moduleName);
+ }
+ for (String poolName : pools) {
+ metricService.remove(MetricType.AUTO_GAUGE, POOL_CPU_USAGE, POOL,
poolName);
+ metricService.remove(MetricType.AUTO_GAUGE, POOL_USER_TIME_PERCENTAGE,
POOL, poolName);
+ }
+ }
+
+ public Map<String, Double> getModuleCpuUsage() {
+ checkAndMayUpdate();
+ return moduleCpuTimePercentageMap;
+ }
+
+ public Map<String, Double> getPoolCpuUsage() {
+ checkAndMayUpdate();
+ return poolCpuUsageMap;
+ }
+
+ public Map<String, Double> getPoolUserCpuPercentage() {
+ checkAndMayUpdate();
+ return poolUserTimePercentageMap;
+ }
+
+ public Map<String, Double> getModuleUserTimePercentage() {
+ checkAndMayUpdate();
+ return moduleUserTimePercentageMap;
+ }
+
+ private synchronized void checkAndMayUpdate() {
+ long currentTime = System.currentTimeMillis();
+ if (currentTime - lastUpdateTime.get() > UPDATE_INTERVAL) {
+ lastUpdateTime.set(currentTime);
+ updateCpuUsage();
+ }
+ }
+
+ private String getThreadModuleById(long id, ThreadInfo threadInfo) {
+ return threadIdToModuleCache.computeIfAbsent(
+ id, k -> threadNameToModule.apply(threadInfo.getThreadName()));
+ }
+
+ private String getThreadPoolById(long id, ThreadInfo threadInfo) {
+ return threadIdToPoolCache.computeIfAbsent(
+ id, k -> threadNameToPool.apply(threadInfo.getThreadName()));
+ }
+
+ private void updateCpuUsage() {
+ if (!checkCpuMonitorEnable()) {
+ return;
+ }
+ // update
+ long[] taskIds = threadMxBean.getAllThreadIds();
+ ThreadInfo[] threadInfos = threadMxBean.getThreadInfo(taskIds);
+
+ Map<Long, Long> currentThreadCpuTime = new HashMap<>(taskIds.length + 1,
1.0f);
+ Map<Long, Long> currentThreadUserTime = new HashMap<>(taskIds.length + 1,
1.0f);
+ collectThreadCpuInfo(currentThreadCpuTime, currentThreadUserTime,
threadInfos);
+
+ Map<String, Long> moduleIncrementCpuTimeMap = new HashMap<>(modules.size()
+ 1, 1.0f);
+ Map<String, Long> moduleIncrementUserTimeMap = new
HashMap<>(modules.size() + 1, 1.0f);
+ Map<String, Long> poolIncrementCpuTimeMap = new HashMap<>(pools.size() +
1, 1.0f);
+ Map<String, Long> poolIncrementUserTimeMap = new HashMap<>(pools.size() +
1, 1.0f);
+
+ long totalIncrementTime =
+ computeUsageInfoForModuleAndPool(
+ moduleIncrementCpuTimeMap,
+ moduleIncrementUserTimeMap,
+ poolIncrementCpuTimeMap,
+ poolIncrementUserTimeMap,
+ lastThreadCpuTime,
+ lastThreadUserTime,
+ currentThreadCpuTime,
+ currentThreadUserTime,
+ threadInfos);
+
+ if (totalIncrementTime == 0L) {
+ return;
+ }
+
+ updateUsageMap(
+ moduleIncrementCpuTimeMap,
+ moduleIncrementUserTimeMap,
+ poolIncrementCpuTimeMap,
+ poolIncrementUserTimeMap,
+ totalIncrementTime);
+ lastThreadCpuTime.clear();
+ lastThreadCpuTime.putAll(currentThreadCpuTime);
+ lastThreadUserTime.putAll(currentThreadUserTime);
+ }
+
+ private boolean checkCpuMonitorEnable() {
+ if (!threadMxBean.isThreadCpuTimeSupported()) {
+ return false;
+ }
+ if (!threadMxBean.isThreadCpuTimeEnabled()) {
+ threadMxBean.setThreadCpuTimeEnabled(true);
+ }
+ return true;
+ }
+
+ private void collectThreadCpuInfo(
+ Map<Long, Long> cpuTimeMap, Map<Long, Long> userTimeMap, ThreadInfo[]
threadInfos) {
+ Arrays.stream(threadInfos)
+ .forEach(
+ info -> {
+ cpuTimeMap.put(info.getThreadId(),
threadMxBean.getThreadCpuTime(info.getThreadId()));
+ userTimeMap.put(
+ info.getThreadId(),
threadMxBean.getThreadUserTime(info.getThreadId()));
+ });
+ }
+
+ @SuppressWarnings("java:S107")
+ private long computeUsageInfoForModuleAndPool(
+ Map<String, Long> moduleIncrementCpuTimeMap,
+ Map<String, Long> moduleIncrementUserTimeMap,
+ Map<String, Long> poolIncrementCpuTimeMap,
+ Map<String, Long> poolIncrementUserTimeMap,
+ Map<Long, Long> beforeThreadCpuTime,
+ Map<Long, Long> beforeThreadUserTime,
+ Map<Long, Long> afterThreadCpuTime,
+ Map<Long, Long> afterThreadUserTime,
+ ThreadInfo[] threadInfos) {
+ long totalIncrementTime = 0L;
+ for (ThreadInfo threadInfo : threadInfos) {
+ long id = threadInfo.getThreadId();
+ long beforeCpuTime = beforeThreadCpuTime.getOrDefault(id, 0L);
+ long afterCpuTime = afterThreadCpuTime.get(id);
+ long beforeUserTime = beforeThreadUserTime.getOrDefault(id, 0L);
+ long afterUserTime = afterThreadUserTime.get(id);
+ totalIncrementTime += afterCpuTime - beforeCpuTime;
+ String module = getThreadModuleById(id, threadInfo);
+ String pool = getThreadPoolById(id, threadInfo);
+ moduleIncrementCpuTimeMap.compute(
+ module,
+ (k, v) -> v == null ? afterCpuTime - beforeCpuTime : v +
afterCpuTime - beforeCpuTime);
+ moduleIncrementUserTimeMap.compute(
+ module,
+ (k, v) ->
+ v == null ? afterUserTime - beforeUserTime : v + afterUserTime -
beforeUserTime);
+ poolIncrementCpuTimeMap.compute(
+ pool,
+ (k, v) -> v == null ? afterCpuTime - beforeCpuTime : v +
afterCpuTime - beforeCpuTime);
+ poolIncrementUserTimeMap.compute(
+ pool,
+ (k, v) ->
+ v == null ? afterUserTime - beforeUserTime : v + afterUserTime -
beforeUserTime);
+ }
+ return totalIncrementTime;
+ }
+
+ private void updateUsageMap(
+ Map<String, Long> moduleIncrementCpuTimeMap,
+ Map<String, Long> moduleIncrementUserTimeMap,
+ Map<String, Long> poolIncrementCpuTimeMap,
+ Map<String, Long> poolIncrementUserTimeMap,
+ long totalIncrementTime) {
+ if (processCpuLoadGauge == null) {
+ processCpuLoadGauge =
+ metricService.getAutoGauge("process_cpu_load", MetricLevel.CORE,
"name", "process");
+ }
+ double processCpuLoad = processCpuLoadGauge.value();
+ for (Map.Entry<String, Long> entry : moduleIncrementCpuTimeMap.entrySet())
{
+ moduleCpuTimePercentageMap.put(
+ entry.getKey(), entry.getValue() * 1.0 / totalIncrementTime *
processCpuLoad);
+ if (entry.getValue() > 0.0) {
+ moduleUserTimePercentageMap.put(
Review Comment:
Do we need to look at the increment during each pull？
##########
metrics/interface/src/main/java/org/apache/iotdb/metrics/metricsets/cpu/CpuUsageMetrics.java:
##########
@@ -0,0 +1,303 @@
+/*
+ * 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.iotdb.metrics.metricsets.cpu;
+
+import org.apache.iotdb.metrics.AbstractMetricService;
+import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.type.AutoGauge;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.metrics.utils.MetricType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.UnaryOperator;
+
+public class CpuUsageMetrics implements IMetricSet {
+ private static final Logger log =
LoggerFactory.getLogger(CpuUsageMetrics.class);
+ private static final String MODULE_CPU_USAGE = "module_cpu_usage";
+ private static final String POOL_CPU_USAGE = "pool_cpu_usage";
+ private static final String POOL = "pool";
+ private static final String MODULE = "module";
+ private static final String MODULE_USER_TIME_PERCENTAGE =
"module_user_time_percentage";
+ private static final String POOL_USER_TIME_PERCENTAGE =
"user_time_percentage";
+ private final List<String> modules;
+ private final List<String> pools;
+ private static final long UPDATE_INTERVAL = 10_000L;
+ protected AbstractMetricService metricService;
+ protected final UnaryOperator<String> threadNameToModule;
+ protected final UnaryOperator<String> threadNameToPool;
+ protected final Map<Long, String> threadIdToModuleCache = new HashMap<>();
+ protected final Map<Long, String> threadIdToPoolCache = new HashMap<>();
+ private final Map<String, Double> moduleCpuTimePercentageMap = new
HashMap<>();
+ private final Map<String, Double> moduleUserTimePercentageMap = new
HashMap<>();
+ private final Map<String, Double> poolCpuUsageMap = new HashMap<>();
+ private final Map<String, Double> poolUserTimePercentageMap = new
HashMap<>();
+ private final Map<Long, Long> lastThreadCpuTime = new HashMap<>();
+ private final Map<Long, Long> lastThreadUserTime = new HashMap<>();
+ AutoGauge processCpuLoadGauge = null;
+ private final ThreadMXBean threadMxBean =
ManagementFactory.getThreadMXBean();
+ private AtomicLong lastUpdateTime = new AtomicLong(0L);
+
+ public CpuUsageMetrics(
+ List<String> modules,
+ List<String> pools,
+ UnaryOperator<String> threadNameToModule,
+ UnaryOperator<String> threadNameToPool) {
+ this.modules = modules;
+ this.pools = pools;
+ this.threadNameToModule = threadNameToModule;
+ this.threadNameToPool = threadNameToPool;
+ }
+
+ @Override
+ public void bindTo(AbstractMetricService metricService) {
+ this.metricService = metricService;
+ for (String moduleName : modules) {
+ metricService.createAutoGauge(
+ MODULE_CPU_USAGE,
+ MetricLevel.IMPORTANT,
+ this,
+ x -> x.getModuleCpuUsage().getOrDefault(moduleName, 0.0),
+ MODULE,
+ moduleName);
+ metricService.createAutoGauge(
+ MODULE_USER_TIME_PERCENTAGE,
+ MetricLevel.IMPORTANT,
+ this,
+ x -> x.getModuleUserTimePercentage().getOrDefault(moduleName, 0.0),
+ MODULE,
+ moduleName);
+ }
+ for (String poolName : pools) {
+ metricService.createAutoGauge(
+ POOL_CPU_USAGE,
+ MetricLevel.IMPORTANT,
+ this,
+ x -> x.getPoolCpuUsage().getOrDefault(poolName, 0.0),
+ POOL,
+ poolName);
+ metricService.createAutoGauge(
+ POOL_USER_TIME_PERCENTAGE,
+ MetricLevel.IMPORTANT,
+ this,
+ x -> x.getPoolUserCpuPercentage().getOrDefault(poolName, 0.0),
+ POOL,
+ poolName);
+ }
+ }
+
+ @Override
+ public void unbindFrom(AbstractMetricService metricService) {
+ for (String moduleName : modules) {
+ metricService.remove(MetricType.AUTO_GAUGE, MODULE_CPU_USAGE, MODULE,
moduleName);
+ metricService.remove(MetricType.AUTO_GAUGE, MODULE_USER_TIME_PERCENTAGE,
MODULE, moduleName);
+ }
+ for (String poolName : pools) {
+ metricService.remove(MetricType.AUTO_GAUGE, POOL_CPU_USAGE, POOL,
poolName);
+ metricService.remove(MetricType.AUTO_GAUGE, POOL_USER_TIME_PERCENTAGE,
POOL, poolName);
+ }
+ }
+
+ public Map<String, Double> getModuleCpuUsage() {
+ checkAndMayUpdate();
+ return moduleCpuTimePercentageMap;
+ }
+
+ public Map<String, Double> getPoolCpuUsage() {
+ checkAndMayUpdate();
+ return poolCpuUsageMap;
+ }
+
+ public Map<String, Double> getPoolUserCpuPercentage() {
+ checkAndMayUpdate();
+ return poolUserTimePercentageMap;
+ }
+
+ public Map<String, Double> getModuleUserTimePercentage() {
+ checkAndMayUpdate();
+ return moduleUserTimePercentageMap;
+ }
+
+ private synchronized void checkAndMayUpdate() {
+ long currentTime = System.currentTimeMillis();
+ if (currentTime - lastUpdateTime.get() > UPDATE_INTERVAL) {
+ lastUpdateTime.set(currentTime);
+ updateCpuUsage();
+ }
+ }
+
+ private String getThreadModuleById(long id, ThreadInfo threadInfo) {
+ return threadIdToModuleCache.computeIfAbsent(
+ id, k -> threadNameToModule.apply(threadInfo.getThreadName()));
+ }
+
+ private String getThreadPoolById(long id, ThreadInfo threadInfo) {
+ return threadIdToPoolCache.computeIfAbsent(
+ id, k -> threadNameToPool.apply(threadInfo.getThreadName()));
+ }
+
+ private void updateCpuUsage() {
+ if (!checkCpuMonitorEnable()) {
+ return;
+ }
+ // update
+ long[] taskIds = threadMxBean.getAllThreadIds();
+ ThreadInfo[] threadInfos = threadMxBean.getThreadInfo(taskIds);
+
+ Map<Long, Long> currentThreadCpuTime = new HashMap<>(taskIds.length + 1,
1.0f);
+ Map<Long, Long> currentThreadUserTime = new HashMap<>(taskIds.length + 1,
1.0f);
+ collectThreadCpuInfo(currentThreadCpuTime, currentThreadUserTime,
threadInfos);
+
+ Map<String, Long> moduleIncrementCpuTimeMap = new HashMap<>(modules.size()
+ 1, 1.0f);
+ Map<String, Long> moduleIncrementUserTimeMap = new
HashMap<>(modules.size() + 1, 1.0f);
+ Map<String, Long> poolIncrementCpuTimeMap = new HashMap<>(pools.size() +
1, 1.0f);
+ Map<String, Long> poolIncrementUserTimeMap = new HashMap<>(pools.size() +
1, 1.0f);
+
+ long totalIncrementTime =
+ computeUsageInfoForModuleAndPool(
+ moduleIncrementCpuTimeMap,
+ moduleIncrementUserTimeMap,
+ poolIncrementCpuTimeMap,
+ poolIncrementUserTimeMap,
+ lastThreadCpuTime,
+ lastThreadUserTime,
+ currentThreadCpuTime,
+ currentThreadUserTime,
+ threadInfos);
+
+ if (totalIncrementTime == 0L) {
+ return;
+ }
+
+ updateUsageMap(
+ moduleIncrementCpuTimeMap,
+ moduleIncrementUserTimeMap,
+ poolIncrementCpuTimeMap,
+ poolIncrementUserTimeMap,
+ totalIncrementTime);
+ lastThreadCpuTime.clear();
Review Comment:
lost `lastThreadUserTime.clear()`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
