From 163890123ca5055d02e116e0697caa0acb92a0bd Mon Sep 17 00:00:00 2001 From: yaoyinglong <906271196@qq.com> Date: Fri, 17 Jul 2026 17:13:17 +0800 Subject: [PATCH 1/5] feat: Add executor thread pool metrics instrumentation --- instrumentation/executors/README.md | 1 + .../executors/bootstrap/build.gradle.kts | 4 + .../executors/JvmExecutorMetrics.java | 142 +++++ .../executors/ThreadPoolExecutorMetrics.java | 342 +++++++++++ .../executors/javaagent/build.gradle.kts | 23 + .../ExecutorsInstrumentationModule.java | 4 + ...PerTaskExecutorMetricsInstrumentation.java | 65 +++ ...eadPoolExecutorMetricsInstrumentation.java | 103 ++++ .../ThreadPoolExecutorMetricsTest.java | 533 ++++++++++++++++++ .../ThreadPerTaskExecutorMetricsTest.java | 205 +++++++ instrumentation/executors/metadata.yaml | 10 + 11 files changed, 1432 insertions(+) create mode 100644 instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/JvmExecutorMetrics.java create mode 100644 instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/ThreadPoolExecutorMetrics.java create mode 100644 instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPerTaskExecutorMetricsInstrumentation.java create mode 100644 instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPoolExecutorMetricsInstrumentation.java create mode 100644 instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java create mode 100644 instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java diff --git a/instrumentation/executors/README.md b/instrumentation/executors/README.md index deb0e9ed46cc..ff6fc8f10e28 100644 --- a/instrumentation/executors/README.md +++ b/instrumentation/executors/README.md @@ -4,3 +4,4 @@ | -------------------------------------------- | ------- | ------- | -------------------------------------------------------------------------- | | `otel.instrumentation.executors.include` | List | Empty | List of `Executor` subclasses to be instrumented. | | `otel.instrumentation.executors.include-all` | Boolean | `false` | Whether to instrument all classes that implement the `Executor` interface. | +| `otel.instrumentation.executors.name-normalization` | String | `all` | Replaces all consecutive digits in executor thread names with `*` for `all`; `trailing` replaces only trailing digits; other values use `all`. | diff --git a/instrumentation/executors/bootstrap/build.gradle.kts b/instrumentation/executors/bootstrap/build.gradle.kts index 072a96df450f..bfbb14f92c87 100644 --- a/instrumentation/executors/bootstrap/build.gradle.kts +++ b/instrumentation/executors/bootstrap/build.gradle.kts @@ -1,3 +1,7 @@ plugins { id("otel.javaagent-bootstrap") } + +dependencies { + compileOnly(project(":instrumentation-api-incubator")) +} diff --git a/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/JvmExecutorMetrics.java b/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/JvmExecutorMetrics.java new file mode 100644 index 000000000000..861dfe347adf --- /dev/null +++ b/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/JvmExecutorMetrics.java @@ -0,0 +1,142 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.bootstrap.executors; + +import static io.opentelemetry.api.common.AttributeKey.stringKey; + +import io.opentelemetry.api.OpenTelemetry; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.BatchCallback; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.Meter; +import io.opentelemetry.api.metrics.MeterBuilder; +import io.opentelemetry.api.metrics.ObservableLongMeasurement; +import io.opentelemetry.api.metrics.ObservableMeasurement; +import io.opentelemetry.instrumentation.api.internal.EmbeddedInstrumentationProperties; + +public final class JvmExecutorMetrics { + + private static final AttributeKey EXECUTOR_NAME = stringKey("jvm.executor.name"); + private static final AttributeKey EXECUTOR_OWNER_NAME_ATTRIBUTE = + stringKey("jvm.executor.owner.name"); + private static final AttributeKey EXECUTOR_TYPE = stringKey("jvm.executor.type"); + private static final AttributeKey STATE = stringKey("jvm.executor.state"); + + private static final String ACTIVE_STATE = "active"; + private static final String IDLE_STATE = "idle"; + + static JvmExecutorMetrics create( + OpenTelemetry openTelemetry, + String instrumentationName, + String executorName, + String executorOwnerName, + String executorType) { + MeterBuilder meterBuilder = openTelemetry.getMeterProvider().meterBuilder(instrumentationName); + String instrumentationVersion = + EmbeddedInstrumentationProperties.findVersion(instrumentationName); + if (instrumentationVersion != null) { + meterBuilder.setInstrumentationVersion(instrumentationVersion); + } + + return new JvmExecutorMetrics( + meterBuilder.build(), + Attributes.of( + EXECUTOR_NAME, + executorName, + EXECUTOR_OWNER_NAME_ATTRIBUTE, + executorOwnerName, + EXECUTOR_TYPE, + executorType)); + } + + private final Meter meter; + private final Attributes attributes; + private final Attributes activeThreadAttributes; + private final Attributes idleThreadAttributes; + + private JvmExecutorMetrics(Meter meter, Attributes attributes) { + this.meter = meter; + this.attributes = attributes; + activeThreadAttributes = attributes.toBuilder().put(STATE, ACTIVE_STATE).build(); + idleThreadAttributes = attributes.toBuilder().put(STATE, IDLE_STATE).build(); + } + + ObservableLongMeasurement threadCount() { + return meter + .upDownCounterBuilder("jvm.executor.thread.count") + .setUnit("{thread}") + .setDescription("The number of executor threads that are currently in the described state.") + .buildObserver(); + } + + ObservableLongMeasurement coreThreads() { + return meter + .upDownCounterBuilder("jvm.executor.thread.core") + .setUnit("{thread}") + .setDescription("The core number of threads configured for the executor.") + .buildObserver(); + } + + ObservableLongMeasurement maxThreads() { + return meter + .upDownCounterBuilder("jvm.executor.thread.max") + .setUnit("{thread}") + .setDescription("The maximum number of threads allowed for the executor.") + .buildObserver(); + } + + ObservableLongMeasurement queueSize() { + return meter + .upDownCounterBuilder("jvm.executor.queue.size") + .setUnit("{task}") + .setDescription("The number of tasks currently queued for execution.") + .buildObserver(); + } + + ObservableLongMeasurement queueRemaining() { + return meter + .upDownCounterBuilder("jvm.executor.queue.remaining") + .setUnit("{task}") + .setDescription("The remaining task capacity of the executor queue.") + .buildObserver(); + } + + ObservableLongMeasurement completedTasks() { + return meter + .counterBuilder("jvm.executor.task.completed") + .setUnit("{task}") + .setDescription("The number of tasks completed by the executor.") + .buildObserver(); + } + + LongCounter rejectedTasks() { + return meter + .counterBuilder("jvm.executor.task.rejected") + .setUnit("{task}") + .setDescription("The number of tasks rejected by the executor.") + .build(); + } + + BatchCallback batchCallback( + Runnable callback, + ObservableMeasurement observableMeasurement, + ObservableMeasurement... additionalMeasurements) { + return meter.batchCallback(callback, observableMeasurement, additionalMeasurements); + } + + Attributes attributes() { + return attributes; + } + + Attributes activeThreadAttributes() { + return activeThreadAttributes; + } + + Attributes idleThreadAttributes() { + return idleThreadAttributes; + } +} diff --git a/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/ThreadPoolExecutorMetrics.java b/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/ThreadPoolExecutorMetrics.java new file mode 100644 index 000000000000..cbc086c22b11 --- /dev/null +++ b/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/ThreadPoolExecutorMetrics.java @@ -0,0 +1,342 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.bootstrap.executors; + +import static java.util.Collections.emptySet; + +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.BatchCallback; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.metrics.ObservableLongMeasurement; +import io.opentelemetry.instrumentation.api.incubator.config.internal.DeclarativeConfigUtil; +import io.opentelemetry.instrumentation.api.internal.cache.Cache; +import java.lang.ref.WeakReference; +import java.util.Set; +import java.util.concurrent.Executor; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.atomic.AtomicReference; +import java.util.regex.Pattern; +import javax.annotation.Nullable; + +public final class ThreadPoolExecutorMetrics { + + private static final String INSTRUMENTATION_NAME = "io.opentelemetry.executors"; + private static final String UNKNOWN = "unknown"; + private static final String THREAD_NAME_NORMALIZATION = + DeclarativeConfigUtil.getInstrumentationConfig(GlobalOpenTelemetry.get(), "executors") + .getString("name_normalization", "all"); + private static final Pattern TRAILING_DIGITS_PATTERN = Pattern.compile("\\d+$"); + private static final Pattern ALL_DIGITS_PATTERN = Pattern.compile("\\d+"); + + private static final Cache registrations = Cache.weak(); + + public static void register(ThreadPoolExecutor executor, ThreadFactory threadFactory) { + registerMetrics(executor, threadFactory, emptySet()); + } + + public static void register(Executor executor, ThreadFactory threadFactory, Set threads) { + registerMetrics(executor, threadFactory, threads); + } + + private static void registerMetrics( + Executor executor, ThreadFactory threadFactory, Set threads) { + registrations.computeIfAbsent( + executor, + ignored -> { + String threadName = threadName(threadFactory); + return createRegistration( + executor, + threads, + UNKNOWN, + threadName, + executorName(threadName, THREAD_NAME_NORMALIZATION)); + }); + } + + public static void reregister( + Executor executor, @Nullable String ownerName, String threadNameNormalization) { + String executorOwnerName = ownerName == null ? UNKNOWN : ownerName; + MetricsRegistration registration = registrations.get(executor); + if (registration != null) { + updateRegistration( + executor, + registration, + registration.reregister( + executor, + executorOwnerName, + executorName(registration.threadName, threadNameNormalization))); + } + } + + public static void reregister(ThreadPoolExecutor executor, ThreadFactory threadFactory) { + MetricsRegistration registration = registrations.get(executor); + if (registration != null) { + updateRegistration(executor, registration, registration.reregister(executor, threadFactory)); + } + } + + public static void recordRejectedTask(Executor executor) { + MetricsRegistration registration = registrations.get(executor); + if (registration != null && registration.rejectedTasks != null) { + registration.rejectedTasks.add(1, registration.attributes); + } + } + + public static void unregister(Executor executor) { + MetricsRegistration registration; + synchronized (registrations) { + registration = registrations.get(executor); + if (registration != null) { + registrations.remove(executor); + } + } + if (registration != null) { + registration.close(); + } + } + + private static void updateRegistration( + Executor executor, + MetricsRegistration registration, + @Nullable MetricsRegistration replacement) { + boolean closeRegistration = false; + @Nullable MetricsRegistration discardedReplacement = null; + synchronized (registrations) { + if (registrations.get(executor) != registration) { + if (replacement != registration) { + discardedReplacement = replacement; + } + } else if (replacement == null) { + registrations.remove(executor); + closeRegistration = true; + } else if (replacement != registration) { + registrations.put(executor, replacement); + closeRegistration = true; + } + } + if (closeRegistration) { + registration.close(); + } + if (discardedReplacement != null) { + discardedReplacement.close(); + } + } + + private static MetricsRegistration createRegistration( + Executor executor, + Set threads, + String ownerName, + String threadName, + String executorName) { + JvmExecutorMetrics metrics = + JvmExecutorMetrics.create( + GlobalOpenTelemetry.get(), + INSTRUMENTATION_NAME, + executorName, + ownerName, + executor.getClass().getName()); + + AtomicReference callback; + @Nullable LongCounter rejectedTasks; + if (executor instanceof ThreadPoolExecutor) { + callback = createBatchCallback(metrics, (ThreadPoolExecutor) executor); + rejectedTasks = metrics.rejectedTasks(); + } else { + callback = createBatchCallback(metrics, threads); + rejectedTasks = null; + } + + WeakReference> threadsRef = new WeakReference<>(threads); + + return new MetricsRegistration( + callback, + rejectedTasks, + metrics.attributes(), + threadName, + executorName, + ownerName, + (newExecutor, newOwnerName, newExecutorName) -> { + Set activeThreads = threadsRef.get(); + return activeThreads == null + ? null + : createRegistration( + newExecutor, activeThreads, newOwnerName, threadName, newExecutorName); + }); + } + + private static AtomicReference createBatchCallback( + JvmExecutorMetrics metrics, ThreadPoolExecutor executor) { + ObservableLongMeasurement threadCount = metrics.threadCount(); + ObservableLongMeasurement coreThreads = metrics.coreThreads(); + ObservableLongMeasurement maxThreads = metrics.maxThreads(); + ObservableLongMeasurement queueSize = metrics.queueSize(); + ObservableLongMeasurement queueRemaining = metrics.queueRemaining(); + ObservableLongMeasurement completedTasks = metrics.completedTasks(); + + WeakReference executorRef = new WeakReference<>(executor); + AtomicReference callbackRef = new AtomicReference<>(); + callbackRef.set( + metrics.batchCallback( + () -> { + ThreadPoolExecutor threadPoolExecutor = executorRef.get(); + if (threadPoolExecutor == null) { + closeCallback(callbackRef); + return; + } + + long active = threadPoolExecutor.getActiveCount(); + threadCount.record(active, metrics.activeThreadAttributes()); + threadCount.record( + Math.max(threadPoolExecutor.getPoolSize() - active, 0), + metrics.idleThreadAttributes()); + coreThreads.record(threadPoolExecutor.getCorePoolSize(), metrics.attributes()); + maxThreads.record(threadPoolExecutor.getMaximumPoolSize(), metrics.attributes()); + queueSize.record(threadPoolExecutor.getQueue().size(), metrics.attributes()); + queueRemaining.record( + threadPoolExecutor.getQueue().remainingCapacity(), metrics.attributes()); + completedTasks.record( + threadPoolExecutor.getCompletedTaskCount(), metrics.attributes()); + }, + threadCount, + coreThreads, + maxThreads, + queueSize, + queueRemaining, + completedTasks)); + return callbackRef; + } + + private static AtomicReference createBatchCallback( + JvmExecutorMetrics metrics, Set threads) { + ObservableLongMeasurement threadCount = metrics.threadCount(); + WeakReference> threadsRef = new WeakReference<>(threads); + AtomicReference callbackRef = new AtomicReference<>(); + callbackRef.set( + metrics.batchCallback( + () -> { + Set activeThreads = threadsRef.get(); + if (activeThreads == null) { + closeCallback(callbackRef); + return; + } + + threadCount.record(activeThreads.size(), metrics.activeThreadAttributes()); + }, + threadCount)); + return callbackRef; + } + + private static void closeCallback(AtomicReference callbackRef) { + BatchCallback callback = callbackRef.getAndSet(null); + if (callback != null) { + callback.close(); + } + } + + private static String executorName(@Nullable String threadName, String threadNameNormalization) { + if (threadName == null) { + return UNKNOWN; + } + + threadName = threadName.trim(); + if (threadName.isEmpty()) { + return UNKNOWN; + } + + return ("trailing".equals(threadNameNormalization) + ? TRAILING_DIGITS_PATTERN + : ALL_DIGITS_PATTERN) + .matcher(threadName) + .replaceAll("*"); + } + + private static String threadName(@Nullable ThreadFactory threadFactory) { + try { + if (threadFactory != null) { + Thread thread = threadFactory.newThread(ThreadPoolExecutorMetrics::noopRunnable); + if (thread != null) { + String threadName = thread.getName(); + if (threadName != null) { + threadName = threadName.trim(); + if (!threadName.isEmpty()) { + return threadName; + } + } + } + } + } catch (Throwable ignored) { + // Use the fallback name when the thread factory cannot be safely probed. + } + return UNKNOWN; + } + + private static void noopRunnable() {} + + private interface RegistrationFactory { + @Nullable + MetricsRegistration create(Executor executor, String ownerName, String executorName); + } + + private static final class MetricsRegistration { + private final AtomicReference callback; + @Nullable private final LongCounter rejectedTasks; + private final Attributes attributes; + private final String threadName; + private final String executorName; + private final String ownerName; + private final RegistrationFactory registrationFactory; + + private MetricsRegistration( + AtomicReference callback, + @Nullable LongCounter rejectedTasks, + Attributes attributes, + String threadName, + String executorName, + String ownerName, + RegistrationFactory registrationFactory) { + this.callback = callback; + this.rejectedTasks = rejectedTasks; + this.attributes = attributes; + this.threadName = threadName; + this.executorName = executorName; + this.ownerName = ownerName; + this.registrationFactory = registrationFactory; + } + + @Nullable + private MetricsRegistration reregister( + Executor executor, String ownerName, String newExecutorName) { + if (this.ownerName.equals(ownerName) && executorName.equals(newExecutorName)) { + return this; + } + + MetricsRegistration replacement = + registrationFactory.create(executor, ownerName, newExecutorName); + return replacement; + } + + private MetricsRegistration reregister( + ThreadPoolExecutor executor, ThreadFactory threadFactory) { + String newThreadName = ThreadPoolExecutorMetrics.threadName(threadFactory); + String newExecutorName = executorName(newThreadName, THREAD_NAME_NORMALIZATION); + if (threadName.equals(newThreadName) && executorName.equals(newExecutorName)) { + return this; + } + + MetricsRegistration replacement = + createRegistration(executor, emptySet(), ownerName, newThreadName, newExecutorName); + return replacement; + } + + private void close() { + closeCallback(callback); + } + } + + private ThreadPoolExecutorMetrics() {} +} diff --git a/instrumentation/executors/javaagent/build.gradle.kts b/instrumentation/executors/javaagent/build.gradle.kts index 1b3b676c5dec..2344d0ccafd5 100644 --- a/instrumentation/executors/javaagent/build.gradle.kts +++ b/instrumentation/executors/javaagent/build.gradle.kts @@ -42,6 +42,29 @@ testing { } } } + + register("testTrailingThreadNameNormalization") { + sources { + java { + setSrcDirs(listOf("src/test/java")) + } + } + + dependencies { + implementation(project(":instrumentation:executors:testing")) + compileOnly(project(":instrumentation:executors:bootstrap")) + compileOnly(project(":javaagent-bootstrap")) + } + + targets { + all { + testTask.configure { + systemProperty("otel.instrumentation.executors.name-normalization", "trailing") + systemProperty("test.name-normalization.expected", "trailing") + } + } + } + } } } diff --git a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/ExecutorsInstrumentationModule.java b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/ExecutorsInstrumentationModule.java index b7e57eeee455..86ba74e51c27 100644 --- a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/ExecutorsInstrumentationModule.java +++ b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/ExecutorsInstrumentationModule.java @@ -11,6 +11,8 @@ import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule; import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; import io.opentelemetry.javaagent.extension.instrumentation.internal.EarlyInstrumentationModule; +import io.opentelemetry.javaagent.instrumentation.executors.metrics.ThreadPerTaskExecutorMetricsInstrumentation; +import io.opentelemetry.javaagent.instrumentation.executors.metrics.ThreadPoolExecutorMetricsInstrumentation; import java.util.List; @AutoService({InstrumentationModule.class, EarlyInstrumentationModule.class}) @@ -30,6 +32,8 @@ public List typeInstrumentations() { new JavaForkJoinTaskInstrumentation(), new RunnableInstrumentation(), new ThreadPoolExtendingExecutorInstrumentation(), + new ThreadPoolExecutorMetricsInstrumentation(), + new ThreadPerTaskExecutorMetricsInstrumentation(), new VirtualThreadInstrumentation(), new StructuredTaskScopeInstrumentation()); } diff --git a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPerTaskExecutorMetricsInstrumentation.java b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPerTaskExecutorMetricsInstrumentation.java new file mode 100644 index 000000000000..e24e14ea5bf4 --- /dev/null +++ b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPerTaskExecutorMetricsInstrumentation.java @@ -0,0 +1,65 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.executors.metrics; + +import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.methodIsDeclaredByType; +import static net.bytebuddy.matcher.ElementMatchers.isConstructor; +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.namedOneOf; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; + +import io.opentelemetry.javaagent.bootstrap.executors.ThreadPoolExecutorMetrics; +import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; +import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer; +import java.util.Set; +import java.util.concurrent.Executor; +import java.util.concurrent.ThreadFactory; +import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.matcher.ElementMatcher; + +public class ThreadPerTaskExecutorMetricsInstrumentation implements TypeInstrumentation { + + private static final String THREAD_PER_TASK_EXECUTOR = + "java.util.concurrent.ThreadPerTaskExecutor"; + + @Override + public ElementMatcher typeMatcher() { + return named(THREAD_PER_TASK_EXECUTOR); + } + + @Override + public void transform(TypeTransformer transformer) { + transformer.applyAdviceToMethod( + isConstructor().and(takesArguments(1)), getClass().getName() + "$ConstructorAdvice"); + transformer.applyAdviceToMethod( + namedOneOf("shutdown", "shutdownNow", "close") + .and(takesArguments(0)) + .and(methodIsDeclaredByType(named(THREAD_PER_TASK_EXECUTOR))), + getClass().getName() + "$ShutdownAdvice"); + } + + @SuppressWarnings("unused") + public static class ConstructorAdvice { + + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onExit( + @Advice.This Executor executor, + @Advice.Argument(0) ThreadFactory threadFactory, + @Advice.FieldValue("threads") Set threads) { + ThreadPoolExecutorMetrics.register(executor, threadFactory, threads); + } + } + + @SuppressWarnings("unused") + public static class ShutdownAdvice { + + @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) + public static void onExit(@Advice.This Executor executor) { + ThreadPoolExecutorMetrics.unregister(executor); + } + } +} diff --git a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPoolExecutorMetricsInstrumentation.java b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPoolExecutorMetricsInstrumentation.java new file mode 100644 index 000000000000..1ddc4a4fea8d --- /dev/null +++ b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPoolExecutorMetricsInstrumentation.java @@ -0,0 +1,103 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.executors.metrics; + +import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.extendsClass; +import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.methodIsDeclaredByType; +import static net.bytebuddy.matcher.ElementMatchers.isConstructor; +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.namedOneOf; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; + +import io.opentelemetry.javaagent.bootstrap.executors.ThreadPoolExecutorMetrics; +import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; +import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer; +import java.util.concurrent.Executor; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.matcher.ElementMatcher; + +public class ThreadPoolExecutorMetricsInstrumentation implements TypeInstrumentation { + + @Override + public ElementMatcher typeMatcher() { + return extendsClass(named(ThreadPoolExecutor.class.getName())); + } + + @Override + public void transform(TypeTransformer transformer) { + transformer.applyAdviceToMethod( + isConstructor() + .and(takesArguments(7)) + .and(methodIsDeclaredByType(named(ThreadPoolExecutor.class.getName()))), + getClass().getName() + "$ConstructorAdvice"); + transformer.applyAdviceToMethod( + named("setThreadFactory") + .and(takesArgument(0, ThreadFactory.class)) + .and(takesArguments(1)) + .and(methodIsDeclaredByType(named(ThreadPoolExecutor.class.getName()))), + getClass().getName() + "$SetThreadFactoryAdvice"); + transformer.applyAdviceToMethod( + named("reject") + .and(takesArgument(0, Runnable.class)) + .and(takesArguments(1)) + .and(methodIsDeclaredByType(named(ThreadPoolExecutor.class.getName()))), + getClass().getName() + "$RejectAdvice"); + transformer.applyAdviceToMethod( + namedOneOf("shutdown", "shutdownNow").and(takesArguments(0)), + getClass().getName() + "$ShutdownAdvice"); + } + + @SuppressWarnings("unused") + public static class ConstructorAdvice { + + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onExit( + @Advice.This ThreadPoolExecutor executor, @Advice.Argument(5) ThreadFactory threadFactory) { + if (!(executor instanceof ScheduledThreadPoolExecutor)) { + ThreadPoolExecutorMetrics.register(executor, threadFactory); + } + } + } + + @SuppressWarnings("unused") + public static class SetThreadFactoryAdvice { + + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onExit( + @Advice.This ThreadPoolExecutor executor, @Advice.Argument(0) ThreadFactory threadFactory) { + if (!(executor instanceof ScheduledThreadPoolExecutor)) { + ThreadPoolExecutorMetrics.reregister(executor, threadFactory); + } + } + } + + @SuppressWarnings("unused") + public static class RejectAdvice { + + @Advice.OnMethodEnter(suppress = Throwable.class) + public static void onEnter(@Advice.This ThreadPoolExecutor executor) { + if (!(executor instanceof ScheduledThreadPoolExecutor)) { + ThreadPoolExecutorMetrics.recordRejectedTask(executor); + } + } + } + + @SuppressWarnings("unused") + public static class ShutdownAdvice { + + @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) + public static void onExit(@Advice.This Executor executor) { + if (!(executor instanceof ScheduledThreadPoolExecutor)) { + ThreadPoolExecutorMetrics.unregister(executor); + } + } + } +} diff --git a/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java b/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java new file mode 100644 index 000000000000..b53a4f2871dd --- /dev/null +++ b/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java @@ -0,0 +1,533 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.executors; + +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.instrumentation.test.utils.GcUtils; +import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension; +import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; +import io.opentelemetry.javaagent.bootstrap.executors.ThreadPoolExecutorMetrics; +import io.opentelemetry.sdk.metrics.data.LongPointData; +import java.lang.ref.WeakReference; +import java.time.Duration; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +class ThreadPoolExecutorMetricsTest { + + private static final String INSTRUMENTATION_NAME = "io.opentelemetry.executors"; + private static final String DEFAULT_OWNER_NAME = "unknown"; + private static final String EXPECTED_THREAD_NAME_NORMALIZATION = + "test.name-normalization.expected"; + + private static final AttributeKey EXECUTOR_NAME = stringKey("jvm.executor.name"); + private static final AttributeKey EXECUTOR_OWNER_NAME = + stringKey("jvm.executor.owner.name"); + private static final AttributeKey EXECUTOR_TYPE = stringKey("jvm.executor.type"); + private static final AttributeKey EXECUTOR_STATE = stringKey("jvm.executor.state"); + + @RegisterExtension + static final InstrumentationExtension testing = AgentInstrumentationExtension.create(); + + @Test + void recordsThreadPoolMetricsAndUnregistersOnShutdown() throws Exception { + ThreadPoolExecutor executor = + new ThreadPoolExecutor( + 1, + 1, + 0, + MILLISECONDS, + new ArrayBlockingQueue<>(1), + new NamedThreadFactory("metrics-pool")); + + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + + try { + executor.execute( + () -> { + started.countDown(); + try { + release.await(10, SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + }); + assertThat(started.await(10, SECONDS)).isTrue(); + + executor.execute(() -> {}); + assertThatThrownBy(() -> executor.execute(() -> {})) + .isInstanceOf(RejectedExecutionException.class); + + assertThreadCountMetric("metrics-pool-*", DEFAULT_OWNER_NAME, "active", 1); + assertThreadCountMetric("metrics-pool-*", DEFAULT_OWNER_NAME, "idle", 0); + assertExecutorMetric( + "jvm.executor.thread.core", + "{thread}", + "The core number of threads configured for the executor.", + "metrics-pool-*", + DEFAULT_OWNER_NAME, + 1); + assertExecutorMetric( + "jvm.executor.thread.max", + "{thread}", + "The maximum number of threads allowed for the executor.", + "metrics-pool-*", + DEFAULT_OWNER_NAME, + 1); + assertExecutorMetric( + "jvm.executor.queue.size", + "{task}", + "The number of tasks currently queued for execution.", + "metrics-pool-*", + DEFAULT_OWNER_NAME, + 1); + assertExecutorMetric( + "jvm.executor.queue.remaining", + "{task}", + "The remaining task capacity of the executor queue.", + "metrics-pool-*", + DEFAULT_OWNER_NAME, + 0); + assertExecutorMetric( + "jvm.executor.task.completed", + "{task}", + "The number of tasks completed by the executor.", + "metrics-pool-*", + DEFAULT_OWNER_NAME, + 0); + assertExecutorMetric( + "jvm.executor.task.rejected", + "{task}", + "The number of tasks rejected by the executor.", + "metrics-pool-*", + DEFAULT_OWNER_NAME, + 1); + } finally { + release.countDown(); + executor.shutdown(); + assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); + } + + testing.clearData(); + assertNoExecutorMetric("jvm.executor.thread.count", "metrics-pool-*"); + } + + @Test + void skipsScheduledThreadPoolExecutor() { + ScheduledThreadPoolExecutor executor = + new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("scheduled-pool")); + + try { + assertNoExecutorMetric("jvm.executor.thread.count", "scheduled-pool-*"); + } finally { + executor.shutdownNow(); + } + } + + @Test + void unregistersUnlistedThreadPoolExecutorOnOverriddenShutdown() { + UnlistedThreadPoolExecutor executor = + new UnlistedThreadPoolExecutor(new NamedThreadFactory("unlisted-pool")); + + try { + testing.waitAndAssertMetrics( + INSTRUMENTATION_NAME, + "jvm.executor.thread.core", + metrics -> + metrics.anySatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .anySatisfy( + point -> + assertThat(point.getAttributes().get(EXECUTOR_TYPE)) + .isEqualTo(UnlistedThreadPoolExecutor.class.getName())))); + + executor.shutdown(); + testing.clearData(); + + assertNoExecutorMetric( + "jvm.executor.thread.core", + "unlisted-pool-*", + UnlistedThreadPoolExecutor.class.getName()); + } finally { + executor.shutdownNow(); + } + } + + @Test + void doesNotRegisterMetricsWhenConstructorFails() { + NamedThreadFactory threadFactory = new NamedThreadFactory("failed-pool"); + + assertThatThrownBy( + () -> + new ThreadPoolExecutor( + 2, 1, 0, MILLISECONDS, new ArrayBlockingQueue<>(1), threadFactory)) + .isInstanceOf(IllegalArgumentException.class); + + assertThat(threadFactory.createdThreadCount()).isZero(); + assertNoExecutorMetric("jvm.executor.thread.count", "failed-pool-*"); + } + + @Test + void getsThreadNameFromStartedThread() throws Exception { + ThreadPoolExecutor executor = + new ThreadPoolExecutor( + 0, + 1, + 1, + SECONDS, + new ArrayBlockingQueue<>(1), + new StartedThreadFactory("started-pool")); + + try { + assertExecutorMetric( + "jvm.executor.thread.core", + "{thread}", + "The core number of threads configured for the executor.", + "started-pool-*", + DEFAULT_OWNER_NAME, + 0); + } finally { + executor.shutdown(); + assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); + } + } + + @Test + void doesNotReregisterAfterUnregister() throws Exception { + ThreadPoolExecutor executor = + new ThreadPoolExecutor( + 0, + 1, + 1, + SECONDS, + new ArrayBlockingQueue<>(1), + new NamedThreadFactory("concurrent-reregister-pool")); + BlockingThreadFactory threadFactory = new BlockingThreadFactory(); + Thread reregisterThread = + new Thread(() -> ThreadPoolExecutorMetrics.reregister(executor, threadFactory)); + + try { + reregisterThread.start(); + assertThat(threadFactory.started.await(10, SECONDS)).isTrue(); + + ThreadPoolExecutorMetrics.unregister(executor); + threadFactory.release.countDown(); + reregisterThread.join(10_000); + assertThat(reregisterThread.isAlive()).isFalse(); + + testing.clearData(); + assertNoExecutorMetric("jvm.executor.thread.core", "reregister-pool"); + } finally { + threadFactory.release.countDown(); + executor.shutdown(); + assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); + } + } + + @Test + void doesNotRecordMetricsWhenUnclosedExecutorIsCollected() throws Exception { + WeakReference executorRef = + new WeakReference<>( + new ThreadPoolExecutor( + 0, + 1, + 1, + SECONDS, + new ArrayBlockingQueue<>(1), + new NamedThreadFactory("collected-pool"))); + + GcUtils.awaitGc(executorRef, Duration.ofSeconds(10)); + + testing.clearData(); + assertNoExecutorMetric("jvm.executor.thread.core", "collected-pool-*"); + } + + @Test + void normalizesExecutorThreadName() throws Exception { + ThreadPoolExecutor executor = + new ThreadPoolExecutor( + 1, + 1, + 0, + MILLISECONDS, + new ArrayBlockingQueue<>(1), + new NamedThreadFactory("name-normalization-1-test")); + + try { + String expectedExecutorName = + "trailing".equals(System.getProperty(EXPECTED_THREAD_NAME_NORMALIZATION, "all")) + ? "name-normalization-1-test-*" + : "name-normalization-*-test-*"; + + assertExecutorMetric( + "jvm.executor.thread.core", + "{thread}", + "The core number of threads configured for the executor.", + expectedExecutorName, + DEFAULT_OWNER_NAME, + 1); + } finally { + executor.shutdown(); + assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); + } + } + + @Test + void reregistersOwnerAndThreadFactory() throws Exception { + NamedThreadFactory originalThreadFactory = new NamedThreadFactory("original-pool-42-worker"); + ThreadPoolExecutor executor = + new ThreadPoolExecutor( + 1, 1, 0, MILLISECONDS, new ArrayBlockingQueue<>(1), originalThreadFactory); + + try { + String originalExecutorName = + "trailing".equals(System.getProperty(EXPECTED_THREAD_NAME_NORMALIZATION, "all")) + ? "original-pool-42-worker-*" + : "original-pool-*-worker-*"; + assertExecutorMetric( + "jvm.executor.thread.core", + "{thread}", + "The core number of threads configured for the executor.", + originalExecutorName, + DEFAULT_OWNER_NAME, + 1); + assertThat(originalThreadFactory.createdThreadCount()).isEqualTo(1); + + NamedThreadFactory replacementThreadFactory = + new NamedThreadFactory("original-pool-43-worker"); + executor.setThreadFactory(replacementThreadFactory); + assertThat(replacementThreadFactory.createdThreadCount()).isEqualTo(1); + + ThreadPoolExecutorMetrics.reregister(executor, "tomcat", "trailing"); + testing.clearData(); + + assertExecutorMetricAttributes( + "jvm.executor.thread.core", + "{thread}", + "The core number of threads configured for the executor.", + "original-pool-43-worker-*", + "tomcat"); + assertNoExecutorMetric("jvm.executor.thread.core", originalExecutorName); + assertThat(replacementThreadFactory.createdThreadCount()).isEqualTo(1); + } finally { + executor.shutdown(); + assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); + } + } + + @Test + void doesNotAddSuffixWhenExecutorNamesCollide() throws Exception { + ThreadPoolExecutor first = + new ThreadPoolExecutor( + 1, + 1, + 0, + MILLISECONDS, + new ArrayBlockingQueue<>(1), + new NamedThreadFactory("shared-pool")); + ThreadPoolExecutor second = + new ThreadPoolExecutor( + 1, + 1, + 0, + MILLISECONDS, + new ArrayBlockingQueue<>(1), + new NamedThreadFactory("shared-pool")); + + try { + testing.waitAndAssertMetrics( + INSTRUMENTATION_NAME, + "jvm.executor.thread.core", + metrics -> + metrics.anySatisfy( + metric -> { + long sharedExecutorCount = + metric.getLongSumData().getPoints().stream() + .filter( + point -> + "shared-pool-*" + .equals(point.getAttributes().get(EXECUTOR_NAME))) + .count(); + assertThat(sharedExecutorCount).isGreaterThan(0); + assertThat(metric.getLongSumData().getPoints()) + .noneMatch( + point -> + "shared-pool-*-2".equals(point.getAttributes().get(EXECUTOR_NAME))); + })); + } finally { + first.shutdown(); + second.shutdown(); + assertThat(first.awaitTermination(10, SECONDS)).isTrue(); + assertThat(second.awaitTermination(10, SECONDS)).isTrue(); + } + } + + private static void assertThreadCountMetric( + String executorName, String ownerName, String state, long expectedValue) { + testing.waitAndAssertMetrics( + INSTRUMENTATION_NAME, + "jvm.executor.thread.count", + metrics -> + metrics.anySatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .anySatisfy( + point -> { + assertThat(metric.getUnit()).isEqualTo("{thread}"); + assertThat(metric.getDescription()) + .isEqualTo( + "The number of executor threads that are currently in the described state."); + assertExecutorAttributes(point, executorName, ownerName); + assertThat(point.getAttributes().get(EXECUTOR_STATE)) + .isEqualTo(state); + assertThat(point.getValue()).isEqualTo(expectedValue); + }))); + } + + private static void assertExecutorMetric( + String metricName, + String unit, + String description, + String executorName, + String ownerName, + long expectedValue) { + testing.waitAndAssertMetrics( + INSTRUMENTATION_NAME, + metricName, + metrics -> + metrics.anySatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .anySatisfy( + point -> { + assertThat(metric.getUnit()).isEqualTo(unit); + assertThat(metric.getDescription()).isEqualTo(description); + assertExecutorAttributes(point, executorName, ownerName); + assertThat(point.getValue()).isEqualTo(expectedValue); + }))); + } + + private static void assertNoExecutorMetric(String metricName, String executorName) { + assertNoExecutorMetric(metricName, executorName, ThreadPoolExecutor.class.getName()); + } + + private static void assertNoExecutorMetric( + String metricName, String executorName, String executorType) { + testing.waitAndAssertMetrics( + INSTRUMENTATION_NAME, + metricName, + metrics -> + metrics.allSatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .noneMatch( + point -> + executorName.equals(point.getAttributes().get(EXECUTOR_NAME)) + && executorType.equals( + point.getAttributes().get(EXECUTOR_TYPE))))); + } + + private static void assertExecutorMetricAttributes( + String metricName, String unit, String description, String executorName, String ownerName) { + testing.waitAndAssertMetrics( + INSTRUMENTATION_NAME, + metricName, + metrics -> + metrics.anySatisfy( + metric -> { + assertThat(metric.getUnit()).isEqualTo(unit); + assertThat(metric.getDescription()).isEqualTo(description); + assertThat(metric.getLongSumData().getPoints()) + .anySatisfy( + point -> assertExecutorAttributes(point, executorName, ownerName)); + })); + } + + private static void assertExecutorAttributes( + LongPointData point, String executorName, String ownerName) { + assertThat(point.getAttributes().get(EXECUTOR_NAME)).isEqualTo(executorName); + assertThat(point.getAttributes().get(EXECUTOR_OWNER_NAME)).isEqualTo(ownerName); + assertThat(point.getAttributes().get(EXECUTOR_TYPE)) + .isEqualTo(ThreadPoolExecutor.class.getName()); + } + + private static final class NamedThreadFactory implements ThreadFactory { + private final String namePrefix; + private final AtomicInteger sequence = new AtomicInteger(); + + private NamedThreadFactory(String namePrefix) { + this.namePrefix = namePrefix; + } + + @Override + public Thread newThread(Runnable runnable) { + return new Thread(runnable, namePrefix + "-" + sequence.incrementAndGet()); + } + + private int createdThreadCount() { + return sequence.get(); + } + } + + private static final class StartedThreadFactory implements ThreadFactory { + private final String namePrefix; + private final AtomicInteger sequence = new AtomicInteger(); + + private StartedThreadFactory(String namePrefix) { + this.namePrefix = namePrefix; + } + + @Override + public Thread newThread(Runnable runnable) { + Thread thread = new Thread(runnable, namePrefix + "-" + sequence.incrementAndGet()); + thread.start(); + return thread; + } + } + + private static final class BlockingThreadFactory implements ThreadFactory { + private final CountDownLatch started = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + @Override + public Thread newThread(Runnable runnable) { + started.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + return new Thread(runnable, "reregister-pool"); + } + } + + private static class UnlistedThreadPoolExecutor extends ThreadPoolExecutor { + + private UnlistedThreadPoolExecutor(ThreadFactory threadFactory) { + super(1, 1, 0, MILLISECONDS, new ArrayBlockingQueue<>(1), threadFactory); + } + + @Override + public void shutdown() {} + } +} diff --git a/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java b/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java new file mode 100644 index 000000000000..0c857d435d91 --- /dev/null +++ b/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java @@ -0,0 +1,205 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.executors; + +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.assertj.core.api.Assertions.assertThat; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.instrumentation.test.utils.GcUtils; +import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension; +import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; +import io.opentelemetry.javaagent.bootstrap.executors.ThreadPoolExecutorMetrics; +import java.lang.ref.WeakReference; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +class ThreadPerTaskExecutorMetricsTest { + + private static final String INSTRUMENTATION_NAME = "io.opentelemetry.executors"; + private static final String EXECUTOR_NAME = "thread-per-task-*"; + private static final String EXECUTOR_TYPE = "java.util.concurrent.ThreadPerTaskExecutor"; + + private static final AttributeKey EXECUTOR_NAME_KEY = stringKey("jvm.executor.name"); + private static final AttributeKey EXECUTOR_OWNER_NAME_KEY = + stringKey("jvm.executor.owner.name"); + private static final AttributeKey EXECUTOR_TYPE_KEY = stringKey("jvm.executor.type"); + private static final AttributeKey EXECUTOR_STATE_KEY = stringKey("jvm.executor.state"); + + @RegisterExtension + static final InstrumentationExtension testing = AgentInstrumentationExtension.create(); + + @Test + void recordsActiveThreadCountAndUnregistersOnShutdown() throws Exception { + ExecutorService executor = + Executors.newThreadPerTaskExecutor(new NamedThreadFactory("thread-per-task")); + CountDownLatch started = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + + try { + Future future = + executor.submit( + () -> { + started.countDown(); + try { + release.await(10, SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + }); + + assertThat(started.await(10, SECONDS)).isTrue(); + assertActiveThreadCount(1); + assertNoUnsupportedMetrics(); + + release.countDown(); + future.get(10, SECONDS); + assertActiveThreadCount(0); + } finally { + release.countDown(); + executor.shutdown(); + assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); + } + + testing.clearData(); + assertNoThreadPerTaskMetric("jvm.executor.thread.count"); + } + + @Test + void unregistersOnClose() { + ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor(); + + try { + testing.waitAndAssertMetrics( + INSTRUMENTATION_NAME, + "jvm.executor.thread.count", + metrics -> + metrics.anySatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .anySatisfy( + point -> + assertThat(point.getAttributes().get(EXECUTOR_TYPE_KEY)) + .isEqualTo(EXECUTOR_TYPE)))); + executor.close(); + } finally { + executor.shutdown(); + } + + testing.clearData(); + assertNoThreadPerTaskMetric("jvm.executor.thread.count"); + } + + @Test + void doesNotRecordMetricsWhenUnclosedExecutorIsCollected() throws Exception { + WeakReference executorRef = + new WeakReference<>( + Executors.newThreadPerTaskExecutor( + new NamedThreadFactory("collected-thread-per-task"))); + + GcUtils.awaitGc(executorRef, Duration.ofSeconds(10)); + + testing.clearData(); + assertNoThreadPerTaskMetric("jvm.executor.thread.count"); + } + + @Test + void reregistersOwnerAndThreadNameNormalization() { + NamedThreadFactory threadFactory = new NamedThreadFactory("thread-per-task-42-worker"); + ExecutorService executor = Executors.newThreadPerTaskExecutor(threadFactory); + + try { + assertActiveThreadCount(0, "thread-per-task-*-worker-*", "unknown"); + assertThat(threadFactory.createdThreadCount()).isEqualTo(1); + + ThreadPoolExecutorMetrics.reregister(executor, "tomcat", "trailing"); + testing.clearData(); + + assertActiveThreadCount(0, "thread-per-task-42-worker-*", "tomcat"); + assertThat(threadFactory.createdThreadCount()).isEqualTo(1); + } finally { + executor.shutdown(); + } + } + + private static void assertActiveThreadCount(long expectedValue) { + assertActiveThreadCount(expectedValue, EXECUTOR_NAME, "unknown"); + } + + private static void assertActiveThreadCount( + long expectedValue, String executorName, String ownerName) { + testing.waitAndAssertMetrics( + INSTRUMENTATION_NAME, + "jvm.executor.thread.count", + metrics -> + metrics.anySatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .anySatisfy( + point -> { + assertThat(metric.getUnit()).isEqualTo("{thread}"); + assertThat(point.getAttributes().get(EXECUTOR_NAME_KEY)) + .isEqualTo(executorName); + assertThat(point.getAttributes().get(EXECUTOR_OWNER_NAME_KEY)) + .isEqualTo(ownerName); + assertThat(point.getAttributes().get(EXECUTOR_TYPE_KEY)) + .isEqualTo(EXECUTOR_TYPE); + assertThat(point.getAttributes().get(EXECUTOR_STATE_KEY)) + .isEqualTo("active"); + assertThat(point.getValue()).isEqualTo(expectedValue); + }))); + } + + private static void assertNoUnsupportedMetrics() { + assertNoThreadPerTaskMetric("jvm.executor.thread.core"); + assertNoThreadPerTaskMetric("jvm.executor.thread.max"); + assertNoThreadPerTaskMetric("jvm.executor.queue.size"); + assertNoThreadPerTaskMetric("jvm.executor.queue.remaining"); + assertNoThreadPerTaskMetric("jvm.executor.task.completed"); + assertNoThreadPerTaskMetric("jvm.executor.task.rejected"); + } + + private static void assertNoThreadPerTaskMetric(String metricName) { + testing.waitAndAssertMetrics( + INSTRUMENTATION_NAME, + metricName, + metrics -> + metrics.allSatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .noneMatch( + point -> + EXECUTOR_TYPE.equals( + point.getAttributes().get(EXECUTOR_TYPE_KEY))))); + } + + private static final class NamedThreadFactory implements ThreadFactory { + private final String namePrefix; + private final AtomicInteger sequence = new AtomicInteger(); + + private NamedThreadFactory(String namePrefix) { + this.namePrefix = namePrefix; + } + + @Override + public Thread newThread(Runnable runnable) { + return new Thread(runnable, namePrefix + "-" + sequence.incrementAndGet()); + } + + private int createdThreadCount() { + return sequence.get(); + } + } +} diff --git a/instrumentation/executors/metadata.yaml b/instrumentation/executors/metadata.yaml index 4cf24859bce8..4b0d33c33e60 100644 --- a/instrumentation/executors/metadata.yaml +++ b/instrumentation/executors/metadata.yaml @@ -22,3 +22,13 @@ configurations: description: Whether to instrument all classes that implement the Executor interface. type: boolean default: false + - name: otel.instrumentation.executors.name-normalization + declarative_name: java.executors.thread_name_normalization + description: > + Determines how executor thread names are normalized. all replaces every group of digits with + *; trailing replaces only the final group; other values act as all. + type: string + default: all + examples: + - all + - trailing From f1a7be22d6638974c72a3a9f0a9a3386fe4dbf4f Mon Sep 17 00:00:00 2001 From: yaoyinglong <906271196@qq.com> Date: Mon, 20 Jul 2026 15:54:31 +0800 Subject: [PATCH 2/5] fix: resolve CR issues and failed test cases --- .../executors/ThreadPoolExecutorMetrics.java | 298 +++++------- ...PerTaskExecutorMetricsInstrumentation.java | 27 +- ...eadPoolExecutorMetricsInstrumentation.java | 32 +- .../ThreadPoolExecutorMetricsTest.java | 449 +++++++----------- .../ThreadPerTaskExecutorMetricsTest.java | 141 +++--- .../JvmExecutorMetricsAssertions.java | 310 ++++++++++++ 6 files changed, 723 insertions(+), 534 deletions(-) create mode 100644 instrumentation/executors/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/JvmExecutorMetricsAssertions.java diff --git a/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/ThreadPoolExecutorMetrics.java b/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/ThreadPoolExecutorMetrics.java index cbc086c22b11..152c77709860 100644 --- a/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/ThreadPoolExecutorMetrics.java +++ b/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/ThreadPoolExecutorMetrics.java @@ -17,7 +17,6 @@ import java.lang.ref.WeakReference; import java.util.Set; import java.util.concurrent.Executor; -import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.atomic.AtomicReference; import java.util.regex.Pattern; @@ -33,62 +32,48 @@ public final class ThreadPoolExecutorMetrics { private static final Pattern TRAILING_DIGITS_PATTERN = Pattern.compile("\\d+$"); private static final Pattern ALL_DIGITS_PATTERN = Pattern.compile("\\d+"); - private static final Cache registrations = Cache.weak(); + private static final Cache registrations = Cache.weak(); - public static void register(ThreadPoolExecutor executor, ThreadFactory threadFactory) { - registerMetrics(executor, threadFactory, emptySet()); + public static void preRegister(ThreadPoolExecutor executor) { + preRegister(executor, emptySet()); } - public static void register(Executor executor, ThreadFactory threadFactory, Set threads) { - registerMetrics(executor, threadFactory, threads); - } - - private static void registerMetrics( - Executor executor, ThreadFactory threadFactory, Set threads) { - registrations.computeIfAbsent( - executor, - ignored -> { - String threadName = threadName(threadFactory); - return createRegistration( - executor, - threads, - UNKNOWN, - threadName, - executorName(threadName, THREAD_NAME_NORMALIZATION)); - }); + public static void preRegister(Executor executor, Set threads) { + registrations.computeIfAbsent(executor, ignored -> new Registration(threads)); } public static void reregister( Executor executor, @Nullable String ownerName, String threadNameNormalization) { String executorOwnerName = ownerName == null ? UNKNOWN : ownerName; - MetricsRegistration registration = registrations.get(executor); + Registration registration = registrations.get(executor); if (registration != null) { - updateRegistration( - executor, - registration, - registration.reregister( - executor, - executorOwnerName, - executorName(registration.threadName, threadNameNormalization))); + registration.reregister(executor, executorOwnerName, threadNameNormalization); } } - public static void reregister(ThreadPoolExecutor executor, ThreadFactory threadFactory) { - MetricsRegistration registration = registrations.get(executor); + public static void onThreadFactoryChanged(Executor executor) { + Registration registration = registrations.get(executor); if (registration != null) { - updateRegistration(executor, registration, registration.reregister(executor, threadFactory)); + registration.awaitNextWorkerThread(); + } + } + + public static void onWorkerThreadStarted(Executor executor, Thread thread) { + Registration registration = registrations.get(executor); + if (registration != null) { + registration.onWorkerThreadStarted(executor, thread); } } public static void recordRejectedTask(Executor executor) { - MetricsRegistration registration = registrations.get(executor); - if (registration != null && registration.rejectedTasks != null) { - registration.rejectedTasks.add(1, registration.attributes); + Registration registration = registrations.get(executor); + if (registration != null) { + registration.recordRejectedTask(); } } public static void unregister(Executor executor) { - MetricsRegistration registration; + Registration registration; synchronized (registrations) { registration = registrations.get(executor); if (registration != null) { @@ -100,75 +85,6 @@ public static void unregister(Executor executor) { } } - private static void updateRegistration( - Executor executor, - MetricsRegistration registration, - @Nullable MetricsRegistration replacement) { - boolean closeRegistration = false; - @Nullable MetricsRegistration discardedReplacement = null; - synchronized (registrations) { - if (registrations.get(executor) != registration) { - if (replacement != registration) { - discardedReplacement = replacement; - } - } else if (replacement == null) { - registrations.remove(executor); - closeRegistration = true; - } else if (replacement != registration) { - registrations.put(executor, replacement); - closeRegistration = true; - } - } - if (closeRegistration) { - registration.close(); - } - if (discardedReplacement != null) { - discardedReplacement.close(); - } - } - - private static MetricsRegistration createRegistration( - Executor executor, - Set threads, - String ownerName, - String threadName, - String executorName) { - JvmExecutorMetrics metrics = - JvmExecutorMetrics.create( - GlobalOpenTelemetry.get(), - INSTRUMENTATION_NAME, - executorName, - ownerName, - executor.getClass().getName()); - - AtomicReference callback; - @Nullable LongCounter rejectedTasks; - if (executor instanceof ThreadPoolExecutor) { - callback = createBatchCallback(metrics, (ThreadPoolExecutor) executor); - rejectedTasks = metrics.rejectedTasks(); - } else { - callback = createBatchCallback(metrics, threads); - rejectedTasks = null; - } - - WeakReference> threadsRef = new WeakReference<>(threads); - - return new MetricsRegistration( - callback, - rejectedTasks, - metrics.attributes(), - threadName, - executorName, - ownerName, - (newExecutor, newOwnerName, newExecutorName) -> { - Set activeThreads = threadsRef.get(); - return activeThreads == null - ? null - : createRegistration( - newExecutor, activeThreads, newOwnerName, threadName, newExecutorName); - }); - } - private static AtomicReference createBatchCallback( JvmExecutorMetrics metrics, ThreadPoolExecutor executor) { ObservableLongMeasurement threadCount = metrics.threadCount(); @@ -255,86 +171,128 @@ private static String executorName(@Nullable String threadName, String threadNam .replaceAll("*"); } - private static String threadName(@Nullable ThreadFactory threadFactory) { - try { - if (threadFactory != null) { - Thread thread = threadFactory.newThread(ThreadPoolExecutorMetrics::noopRunnable); - if (thread != null) { - String threadName = thread.getName(); - if (threadName != null) { - threadName = threadName.trim(); - if (!threadName.isEmpty()) { - return threadName; - } - } - } + private static final class Registration { + private final WeakReference> threadsRef; + private String ownerName = UNKNOWN; + private String threadNameNormalization = THREAD_NAME_NORMALIZATION; + @Nullable private AtomicReference callback; + @Nullable private LongCounter rejectedTasks; + private Attributes attributes = Attributes.empty(); + private String threadName = UNKNOWN; + private String executorName = UNKNOWN; + private volatile boolean awaitingWorkerThread = true; + private boolean closed; + + private Registration(Set threads) { + threadsRef = new WeakReference<>(threads); + } + + private synchronized void awaitNextWorkerThread() { + if (!closed) { + awaitingWorkerThread = true; } - } catch (Throwable ignored) { - // Use the fallback name when the thread factory cannot be safely probed. } - return UNKNOWN; - } - private static void noopRunnable() {} + private void onWorkerThreadStarted(Executor executor, Thread thread) { + if (!awaitingWorkerThread) { + return; + } - private interface RegistrationFactory { - @Nullable - MetricsRegistration create(Executor executor, String ownerName, String executorName); - } + @Nullable AtomicReference previous; + synchronized (this) { + if (closed || !awaitingWorkerThread) { + return; + } - private static final class MetricsRegistration { - private final AtomicReference callback; - @Nullable private final LongCounter rejectedTasks; - private final Attributes attributes; - private final String threadName; - private final String executorName; - private final String ownerName; - private final RegistrationFactory registrationFactory; - - private MetricsRegistration( - AtomicReference callback, - @Nullable LongCounter rejectedTasks, - Attributes attributes, - String threadName, - String executorName, - String ownerName, - RegistrationFactory registrationFactory) { - this.callback = callback; - this.rejectedTasks = rejectedTasks; - this.attributes = attributes; - this.threadName = threadName; - this.executorName = executorName; - this.ownerName = ownerName; - this.registrationFactory = registrationFactory; - } + Set threads = threadsRef.get(); + if (threads == null) { + return; + } - @Nullable - private MetricsRegistration reregister( - Executor executor, String ownerName, String newExecutorName) { - if (this.ownerName.equals(ownerName) && executorName.equals(newExecutorName)) { - return this; + awaitingWorkerThread = false; + previous = callback; + String newThreadName = thread.getName(); + registerMetrics( + executor, threads, newThreadName, executorName(newThreadName, threadNameNormalization)); } - MetricsRegistration replacement = - registrationFactory.create(executor, ownerName, newExecutorName); - return replacement; + if (previous != null) { + closeCallback(previous); + } } - private MetricsRegistration reregister( - ThreadPoolExecutor executor, ThreadFactory threadFactory) { - String newThreadName = ThreadPoolExecutorMetrics.threadName(threadFactory); - String newExecutorName = executorName(newThreadName, THREAD_NAME_NORMALIZATION); - if (threadName.equals(newThreadName) && executorName.equals(newExecutorName)) { - return this; + private void reregister( + Executor executor, String newOwnerName, String newThreadNameNormalization) { + @Nullable AtomicReference previous; + synchronized (this) { + if (closed) { + return; + } + + String newExecutorName = executorName(threadName, newThreadNameNormalization); + boolean metricsUnchanged = + ownerName.equals(newOwnerName) && executorName.equals(newExecutorName); + ownerName = newOwnerName; + threadNameNormalization = newThreadNameNormalization; + if (callback == null || metricsUnchanged) { + return; + } + + Set threads = threadsRef.get(); + if (threads == null) { + return; + } + + previous = callback; + registerMetrics(executor, threads, threadName, newExecutorName); } - MetricsRegistration replacement = - createRegistration(executor, emptySet(), ownerName, newThreadName, newExecutorName); - return replacement; + closeCallback(previous); + } + + private synchronized void recordRejectedTask() { + if (callback != null && rejectedTasks != null) { + rejectedTasks.add(1, attributes); + } } private void close() { - closeCallback(callback); + @Nullable AtomicReference previous; + synchronized (this) { + closed = true; + awaitingWorkerThread = false; + previous = callback; + callback = null; + rejectedTasks = null; + attributes = Attributes.empty(); + } + + if (previous != null) { + closeCallback(previous); + } + } + + private void registerMetrics( + Executor executor, Set threads, String threadName, String executorName) { + JvmExecutorMetrics metrics = + JvmExecutorMetrics.create( + GlobalOpenTelemetry.get(), + INSTRUMENTATION_NAME, + executorName, + ownerName, + executor.getClass().getName()); + + if (executor instanceof ThreadPoolExecutor) { + callback = createBatchCallback(metrics, (ThreadPoolExecutor) executor); + rejectedTasks = metrics.rejectedTasks(); + } else { + callback = createBatchCallback(metrics, threads); + rejectedTasks = null; + } + + attributes = metrics.attributes(); + this.threadName = threadName; + this.executorName = executorName; } } diff --git a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPerTaskExecutorMetricsInstrumentation.java b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPerTaskExecutorMetricsInstrumentation.java index e24e14ea5bf4..d22d877c91de 100644 --- a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPerTaskExecutorMetricsInstrumentation.java +++ b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPerTaskExecutorMetricsInstrumentation.java @@ -9,6 +9,7 @@ import static net.bytebuddy.matcher.ElementMatchers.isConstructor; import static net.bytebuddy.matcher.ElementMatchers.named; import static net.bytebuddy.matcher.ElementMatchers.namedOneOf; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; import static net.bytebuddy.matcher.ElementMatchers.takesArguments; import io.opentelemetry.javaagent.bootstrap.executors.ThreadPoolExecutorMetrics; @@ -16,7 +17,7 @@ import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer; import java.util.Set; import java.util.concurrent.Executor; -import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ExecutorService; import net.bytebuddy.asm.Advice; import net.bytebuddy.description.type.TypeDescription; import net.bytebuddy.matcher.ElementMatcher; @@ -35,6 +36,9 @@ public ElementMatcher typeMatcher() { public void transform(TypeTransformer transformer) { transformer.applyAdviceToMethod( isConstructor().and(takesArguments(1)), getClass().getName() + "$ConstructorAdvice"); + transformer.applyAdviceToMethod( + named("start").and(takesArgument(0, Thread.class)).and(takesArguments(1)), + getClass().getName() + "$StartAdvice"); transformer.applyAdviceToMethod( namedOneOf("shutdown", "shutdownNow", "close") .and(takesArguments(0)) @@ -47,10 +51,17 @@ public static class ConstructorAdvice { @Advice.OnMethodExit(suppress = Throwable.class) public static void onExit( - @Advice.This Executor executor, - @Advice.Argument(0) ThreadFactory threadFactory, - @Advice.FieldValue("threads") Set threads) { - ThreadPoolExecutorMetrics.register(executor, threadFactory, threads); + @Advice.This Executor executor, @Advice.FieldValue("threads") Set threads) { + ThreadPoolExecutorMetrics.preRegister(executor, threads); + } + } + + @SuppressWarnings("unused") + public static class StartAdvice { + + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onExit(@Advice.This Executor executor, @Advice.Argument(0) Thread thread) { + ThreadPoolExecutorMetrics.onWorkerThreadStarted(executor, thread); } } @@ -58,8 +69,10 @@ public static void onExit( public static class ShutdownAdvice { @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) - public static void onExit(@Advice.This Executor executor) { - ThreadPoolExecutorMetrics.unregister(executor); + public static void onExit(@Advice.This ExecutorService executor) { + if (executor.isShutdown()) { + ThreadPoolExecutorMetrics.unregister(executor); + } } } } diff --git a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPoolExecutorMetricsInstrumentation.java b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPoolExecutorMetricsInstrumentation.java index 1ddc4a4fea8d..dd0d79f1f2a7 100644 --- a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPoolExecutorMetricsInstrumentation.java +++ b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPoolExecutorMetricsInstrumentation.java @@ -16,7 +16,7 @@ import io.opentelemetry.javaagent.bootstrap.executors.ThreadPoolExecutorMetrics; import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer; -import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; @@ -44,6 +44,11 @@ public void transform(TypeTransformer transformer) { .and(takesArguments(1)) .and(methodIsDeclaredByType(named(ThreadPoolExecutor.class.getName()))), getClass().getName() + "$SetThreadFactoryAdvice"); + transformer.applyAdviceToMethod( + named("runWorker") + .and(takesArguments(1)) + .and(methodIsDeclaredByType(named(ThreadPoolExecutor.class.getName()))), + getClass().getName() + "$RunWorkerAdvice"); transformer.applyAdviceToMethod( named("reject") .and(takesArgument(0, Runnable.class)) @@ -59,10 +64,9 @@ public void transform(TypeTransformer transformer) { public static class ConstructorAdvice { @Advice.OnMethodExit(suppress = Throwable.class) - public static void onExit( - @Advice.This ThreadPoolExecutor executor, @Advice.Argument(5) ThreadFactory threadFactory) { + public static void onExit(@Advice.This ThreadPoolExecutor executor) { if (!(executor instanceof ScheduledThreadPoolExecutor)) { - ThreadPoolExecutorMetrics.register(executor, threadFactory); + ThreadPoolExecutorMetrics.preRegister(executor); } } } @@ -71,10 +75,20 @@ public static void onExit( public static class SetThreadFactoryAdvice { @Advice.OnMethodExit(suppress = Throwable.class) - public static void onExit( - @Advice.This ThreadPoolExecutor executor, @Advice.Argument(0) ThreadFactory threadFactory) { + public static void onExit(@Advice.This ThreadPoolExecutor executor) { if (!(executor instanceof ScheduledThreadPoolExecutor)) { - ThreadPoolExecutorMetrics.reregister(executor, threadFactory); + ThreadPoolExecutorMetrics.onThreadFactoryChanged(executor); + } + } + } + + @SuppressWarnings("unused") + public static class RunWorkerAdvice { + + @Advice.OnMethodEnter(suppress = Throwable.class) + public static void onEnter(@Advice.This ThreadPoolExecutor executor) { + if (!(executor instanceof ScheduledThreadPoolExecutor)) { + ThreadPoolExecutorMetrics.onWorkerThreadStarted(executor, Thread.currentThread()); } } } @@ -94,8 +108,8 @@ public static void onEnter(@Advice.This ThreadPoolExecutor executor) { public static class ShutdownAdvice { @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) - public static void onExit(@Advice.This Executor executor) { - if (!(executor instanceof ScheduledThreadPoolExecutor)) { + public static void onExit(@Advice.This ExecutorService executor) { + if (!(executor instanceof ScheduledThreadPoolExecutor) && executor.isShutdown()) { ThreadPoolExecutorMetrics.unregister(executor); } } diff --git a/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java b/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java index b53a4f2871dd..5422e1dee0dd 100644 --- a/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java +++ b/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java @@ -10,19 +10,19 @@ import static java.util.concurrent.TimeUnit.SECONDS; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.awaitility.Awaitility.await; -import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.instrumentation.test.utils.GcUtils; import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension; import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; import io.opentelemetry.javaagent.bootstrap.executors.ThreadPoolExecutorMetrics; -import io.opentelemetry.sdk.metrics.data.LongPointData; import java.lang.ref.WeakReference; import java.time.Duration; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.atomic.AtomicInteger; @@ -33,18 +33,23 @@ class ThreadPoolExecutorMetricsTest { private static final String INSTRUMENTATION_NAME = "io.opentelemetry.executors"; private static final String DEFAULT_OWNER_NAME = "unknown"; + private static final String THREAD_POOL_EXECUTOR_TYPE = ThreadPoolExecutor.class.getName(); private static final String EXPECTED_THREAD_NAME_NORMALIZATION = "test.name-normalization.expected"; - private static final AttributeKey EXECUTOR_NAME = stringKey("jvm.executor.name"); - private static final AttributeKey EXECUTOR_OWNER_NAME = - stringKey("jvm.executor.owner.name"); - private static final AttributeKey EXECUTOR_TYPE = stringKey("jvm.executor.type"); - private static final AttributeKey EXECUTOR_STATE = stringKey("jvm.executor.state"); - @RegisterExtension static final InstrumentationExtension testing = AgentInstrumentationExtension.create(); + @Test + void requiresExpectedMetricValue() { + assertThatThrownBy( + () -> + JvmExecutorMetricsAssertions.create( + testing, INSTRUMENTATION_NAME, "executor", "owner", "type") + .assertExecutorEmitsMetrics()) + .isInstanceOf(IllegalStateException.class); + } + @Test void recordsThreadPoolMetricsAndUnregistersOnShutdown() throws Exception { ThreadPoolExecutor executor = @@ -76,58 +81,28 @@ void recordsThreadPoolMetricsAndUnregistersOnShutdown() throws Exception { assertThatThrownBy(() -> executor.execute(() -> {})) .isInstanceOf(RejectedExecutionException.class); - assertThreadCountMetric("metrics-pool-*", DEFAULT_OWNER_NAME, "active", 1); - assertThreadCountMetric("metrics-pool-*", DEFAULT_OWNER_NAME, "idle", 0); - assertExecutorMetric( - "jvm.executor.thread.core", - "{thread}", - "The core number of threads configured for the executor.", - "metrics-pool-*", - DEFAULT_OWNER_NAME, - 1); - assertExecutorMetric( - "jvm.executor.thread.max", - "{thread}", - "The maximum number of threads allowed for the executor.", - "metrics-pool-*", - DEFAULT_OWNER_NAME, - 1); - assertExecutorMetric( - "jvm.executor.queue.size", - "{task}", - "The number of tasks currently queued for execution.", - "metrics-pool-*", - DEFAULT_OWNER_NAME, - 1); - assertExecutorMetric( - "jvm.executor.queue.remaining", - "{task}", - "The remaining task capacity of the executor queue.", - "metrics-pool-*", - DEFAULT_OWNER_NAME, - 0); - assertExecutorMetric( - "jvm.executor.task.completed", - "{task}", - "The number of tasks completed by the executor.", - "metrics-pool-*", - DEFAULT_OWNER_NAME, - 0); - assertExecutorMetric( - "jvm.executor.task.rejected", - "{task}", - "The number of tasks rejected by the executor.", - "metrics-pool-*", - DEFAULT_OWNER_NAME, - 1); + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + "metrics-pool-*", + DEFAULT_OWNER_NAME, + THREAD_POOL_EXECUTOR_TYPE) + .withActiveThreads(1) + .withIdleThreads(0) + .withCoreThreads(1) + .withMaxThreads(1) + .withQueueSize(1) + .withQueueRemaining(0) + .withCompletedTasks(0) + .withRejectedTasks(1) + .assertExecutorEmitsMetrics(); } finally { release.countDown(); executor.shutdown(); assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); } - testing.clearData(); - assertNoExecutorMetric("jvm.executor.thread.count", "metrics-pool-*"); + assertNoExecutorMetrics("metrics-pool-*", DEFAULT_OWNER_NAME); } @Test @@ -136,37 +111,41 @@ void skipsScheduledThreadPoolExecutor() { new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("scheduled-pool")); try { - assertNoExecutorMetric("jvm.executor.thread.count", "scheduled-pool-*"); + assertNoExecutorMetrics("scheduled-pool-*", DEFAULT_OWNER_NAME); } finally { executor.shutdownNow(); } } @Test - void unregistersUnlistedThreadPoolExecutorOnOverriddenShutdown() { + void retainsMetricsWhenOverriddenShutdownDoesNotShutdownExecutor() { UnlistedThreadPoolExecutor executor = new UnlistedThreadPoolExecutor(new NamedThreadFactory("unlisted-pool")); try { - testing.waitAndAssertMetrics( - INSTRUMENTATION_NAME, - "jvm.executor.thread.core", - metrics -> - metrics.anySatisfy( - metric -> - assertThat(metric.getLongSumData().getPoints()) - .anySatisfy( - point -> - assertThat(point.getAttributes().get(EXECUTOR_TYPE)) - .isEqualTo(UnlistedThreadPoolExecutor.class.getName())))); + assertThat(executor.prestartCoreThread()).isTrue(); + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + "unlisted-pool-*", + DEFAULT_OWNER_NAME, + UnlistedThreadPoolExecutor.class.getName()) + .withCoreThreads(0) + .assertExecutorEmitsMetrics(); executor.shutdown(); + + assertThat(executor.isShutdown()).isFalse(); testing.clearData(); - assertNoExecutorMetric( - "jvm.executor.thread.core", - "unlisted-pool-*", - UnlistedThreadPoolExecutor.class.getName()); + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + "unlisted-pool-*", + DEFAULT_OWNER_NAME, + UnlistedThreadPoolExecutor.class.getName()) + .withCoreThreads(0) + .assertExecutorEmitsMetrics(); } finally { executor.shutdownNow(); } @@ -183,28 +162,30 @@ void doesNotRegisterMetricsWhenConstructorFails() { .isInstanceOf(IllegalArgumentException.class); assertThat(threadFactory.createdThreadCount()).isZero(); - assertNoExecutorMetric("jvm.executor.thread.count", "failed-pool-*"); + assertNoExecutorMetrics("failed-pool-*", DEFAULT_OWNER_NAME); } @Test - void getsThreadNameFromStartedThread() throws Exception { + void exportsMetricsOnlyAfterWorkerStarts() throws Exception { + NamedThreadFactory threadFactory = new NamedThreadFactory("started-pool"); ThreadPoolExecutor executor = - new ThreadPoolExecutor( - 0, - 1, - 1, - SECONDS, - new ArrayBlockingQueue<>(1), - new StartedThreadFactory("started-pool")); + new ThreadPoolExecutor(1, 1, 0, MILLISECONDS, new ArrayBlockingQueue<>(1), threadFactory); try { - assertExecutorMetric( - "jvm.executor.thread.core", - "{thread}", - "The core number of threads configured for the executor.", - "started-pool-*", - DEFAULT_OWNER_NAME, - 0); + assertThat(threadFactory.createdThreadCount()).isZero(); + assertNoExecutorMetrics("started-pool-*", DEFAULT_OWNER_NAME); + + assertThat(executor.prestartCoreThread()).isTrue(); + assertThat(threadFactory.createdThreadCount()).isEqualTo(1); + + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + "started-pool-*", + DEFAULT_OWNER_NAME, + THREAD_POOL_EXECUTOR_TYPE) + .withCoreThreads(0) + .assertExecutorEmitsMetrics(); } finally { executor.shutdown(); assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); @@ -215,29 +196,30 @@ void getsThreadNameFromStartedThread() throws Exception { void doesNotReregisterAfterUnregister() throws Exception { ThreadPoolExecutor executor = new ThreadPoolExecutor( - 0, + 1, 1, 1, SECONDS, new ArrayBlockingQueue<>(1), - new NamedThreadFactory("concurrent-reregister-pool")); - BlockingThreadFactory threadFactory = new BlockingThreadFactory(); - Thread reregisterThread = - new Thread(() -> ThreadPoolExecutorMetrics.reregister(executor, threadFactory)); + new NamedThreadFactory("reregister-pool")); try { - reregisterThread.start(); - assertThat(threadFactory.started.await(10, SECONDS)).isTrue(); + executor.prestartCoreThread(); + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + "reregister-pool-*", + DEFAULT_OWNER_NAME, + THREAD_POOL_EXECUTOR_TYPE) + .withCoreThreads(1) + .assertExecutorEmitsMetrics(); ThreadPoolExecutorMetrics.unregister(executor); - threadFactory.release.countDown(); - reregisterThread.join(10_000); - assertThat(reregisterThread.isAlive()).isFalse(); + ThreadPoolExecutorMetrics.reregister(executor, "tomcat", "trailing"); - testing.clearData(); - assertNoExecutorMetric("jvm.executor.thread.core", "reregister-pool"); + assertNoExecutorMetrics("reregister-pool-*", DEFAULT_OWNER_NAME); + assertNoExecutorMetrics("reregister-pool-*", "tomcat"); } finally { - threadFactory.release.countDown(); executor.shutdown(); assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); } @@ -255,10 +237,10 @@ void doesNotRecordMetricsWhenUnclosedExecutorIsCollected() throws Exception { new ArrayBlockingQueue<>(1), new NamedThreadFactory("collected-pool"))); + assertNoExecutorMetrics("collected-pool-*", DEFAULT_OWNER_NAME); GcUtils.awaitGc(executorRef, Duration.ofSeconds(10)); - testing.clearData(); - assertNoExecutorMetric("jvm.executor.thread.core", "collected-pool-*"); + assertNoExecutorMetrics("collected-pool-*", DEFAULT_OWNER_NAME); } @Test @@ -278,13 +260,16 @@ void normalizesExecutorThreadName() throws Exception { ? "name-normalization-1-test-*" : "name-normalization-*-test-*"; - assertExecutorMetric( - "jvm.executor.thread.core", - "{thread}", - "The core number of threads configured for the executor.", - expectedExecutorName, - DEFAULT_OWNER_NAME, - 1); + assertThat(executor.prestartCoreThread()).isTrue(); + + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + expectedExecutorName, + DEFAULT_OWNER_NAME, + THREAD_POOL_EXECUTOR_TYPE) + .withCoreThreads(1) + .assertExecutorEmitsMetrics(); } finally { executor.shutdown(); assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); @@ -296,46 +281,80 @@ void reregistersOwnerAndThreadFactory() throws Exception { NamedThreadFactory originalThreadFactory = new NamedThreadFactory("original-pool-42-worker"); ThreadPoolExecutor executor = new ThreadPoolExecutor( - 1, 1, 0, MILLISECONDS, new ArrayBlockingQueue<>(1), originalThreadFactory); + 0, 2, 0, MILLISECONDS, new SynchronousQueue<>(), originalThreadFactory); + CountDownLatch originalWorkerStarted = new CountDownLatch(1); + CountDownLatch releaseOriginalWorker = new CountDownLatch(1); try { String originalExecutorName = "trailing".equals(System.getProperty(EXPECTED_THREAD_NAME_NORMALIZATION, "all")) ? "original-pool-42-worker-*" : "original-pool-*-worker-*"; - assertExecutorMetric( - "jvm.executor.thread.core", - "{thread}", - "The core number of threads configured for the executor.", - originalExecutorName, - DEFAULT_OWNER_NAME, - 1); + + executor.execute( + () -> { + originalWorkerStarted.countDown(); + try { + releaseOriginalWorker.await(10, SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + }); + assertThat(originalWorkerStarted.await(10, SECONDS)).isTrue(); + + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + originalExecutorName, + DEFAULT_OWNER_NAME, + THREAD_POOL_EXECUTOR_TYPE) + .withCoreThreads(0) + .assertExecutorEmitsMetrics(); assertThat(originalThreadFactory.createdThreadCount()).isEqualTo(1); + ThreadPoolExecutorMetrics.reregister(executor, "tomcat", "trailing"); + NamedThreadFactory replacementThreadFactory = new NamedThreadFactory("original-pool-43-worker"); executor.setThreadFactory(replacementThreadFactory); + assertThat(replacementThreadFactory.createdThreadCount()).isZero(); + + testing.clearData(); + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + "original-pool-42-worker-*", + "tomcat", + THREAD_POOL_EXECUTOR_TYPE) + .withCoreThreads(0) + .assertExecutorEmitsMetrics(); + + CountDownLatch replacementWorkerStarted = new CountDownLatch(1); + executor.execute(replacementWorkerStarted::countDown); + assertThat(replacementWorkerStarted.await(10, SECONDS)).isTrue(); assertThat(replacementThreadFactory.createdThreadCount()).isEqualTo(1); - ThreadPoolExecutorMetrics.reregister(executor, "tomcat", "trailing"); testing.clearData(); - assertExecutorMetricAttributes( - "jvm.executor.thread.core", - "{thread}", - "The core number of threads configured for the executor.", - "original-pool-43-worker-*", - "tomcat"); - assertNoExecutorMetric("jvm.executor.thread.core", originalExecutorName); + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + "original-pool-43-worker-*", + "tomcat", + THREAD_POOL_EXECUTOR_TYPE) + .withCoreThreads(0) + .assertExecutorEmitsMetrics(); assertThat(replacementThreadFactory.createdThreadCount()).isEqualTo(1); } finally { + releaseOriginalWorker.countDown(); executor.shutdown(); assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); } } @Test - void doesNotAddSuffixWhenExecutorNamesCollide() throws Exception { + void recordsMetricsWhenExecutorNamesCollide() throws Exception { ThreadPoolExecutor first = new ThreadPoolExecutor( 1, @@ -354,25 +373,20 @@ void doesNotAddSuffixWhenExecutorNamesCollide() throws Exception { new NamedThreadFactory("shared-pool")); try { - testing.waitAndAssertMetrics( - INSTRUMENTATION_NAME, - "jvm.executor.thread.core", - metrics -> - metrics.anySatisfy( - metric -> { - long sharedExecutorCount = - metric.getLongSumData().getPoints().stream() - .filter( - point -> - "shared-pool-*" - .equals(point.getAttributes().get(EXECUTOR_NAME))) - .count(); - assertThat(sharedExecutorCount).isGreaterThan(0); - assertThat(metric.getLongSumData().getPoints()) - .noneMatch( - point -> - "shared-pool-*-2".equals(point.getAttributes().get(EXECUTOR_NAME))); - })); + assertThat(first.prestartCoreThread()).isTrue(); + assertThat(second.prestartCoreThread()).isTrue(); + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + "shared-pool-*", + DEFAULT_OWNER_NAME, + THREAD_POOL_EXECUTOR_TYPE) + .withActiveThreads(0) + .withIdleThreads(0) + .withMaxThreads(2) + .withCoreThreads(2) + .withQueueSize(0) + .assertExecutorEmitsMetrics(); } finally { first.shutdown(); second.shutdown(); @@ -381,93 +395,31 @@ void doesNotAddSuffixWhenExecutorNamesCollide() throws Exception { } } - private static void assertThreadCountMetric( - String executorName, String ownerName, String state, long expectedValue) { - testing.waitAndAssertMetrics( - INSTRUMENTATION_NAME, - "jvm.executor.thread.count", - metrics -> - metrics.anySatisfy( - metric -> - assertThat(metric.getLongSumData().getPoints()) - .anySatisfy( - point -> { - assertThat(metric.getUnit()).isEqualTo("{thread}"); - assertThat(metric.getDescription()) - .isEqualTo( - "The number of executor threads that are currently in the described state."); - assertExecutorAttributes(point, executorName, ownerName); - assertThat(point.getAttributes().get(EXECUTOR_STATE)) - .isEqualTo(state); - assertThat(point.getValue()).isEqualTo(expectedValue); - }))); - } - - private static void assertExecutorMetric( - String metricName, - String unit, - String description, - String executorName, - String ownerName, - long expectedValue) { - testing.waitAndAssertMetrics( - INSTRUMENTATION_NAME, - metricName, - metrics -> - metrics.anySatisfy( - metric -> - assertThat(metric.getLongSumData().getPoints()) - .anySatisfy( - point -> { - assertThat(metric.getUnit()).isEqualTo(unit); - assertThat(metric.getDescription()).isEqualTo(description); - assertExecutorAttributes(point, executorName, ownerName); - assertThat(point.getValue()).isEqualTo(expectedValue); - }))); - } - - private static void assertNoExecutorMetric(String metricName, String executorName) { - assertNoExecutorMetric(metricName, executorName, ThreadPoolExecutor.class.getName()); - } - - private static void assertNoExecutorMetric( - String metricName, String executorName, String executorType) { - testing.waitAndAssertMetrics( - INSTRUMENTATION_NAME, - metricName, - metrics -> - metrics.allSatisfy( - metric -> - assertThat(metric.getLongSumData().getPoints()) - .noneMatch( - point -> - executorName.equals(point.getAttributes().get(EXECUTOR_NAME)) - && executorType.equals( - point.getAttributes().get(EXECUTOR_TYPE))))); - } - - private static void assertExecutorMetricAttributes( - String metricName, String unit, String description, String executorName, String ownerName) { - testing.waitAndAssertMetrics( - INSTRUMENTATION_NAME, - metricName, - metrics -> - metrics.anySatisfy( - metric -> { - assertThat(metric.getUnit()).isEqualTo(unit); - assertThat(metric.getDescription()).isEqualTo(description); - assertThat(metric.getLongSumData().getPoints()) - .anySatisfy( - point -> assertExecutorAttributes(point, executorName, ownerName)); - })); - } - - private static void assertExecutorAttributes( - LongPointData point, String executorName, String ownerName) { - assertThat(point.getAttributes().get(EXECUTOR_NAME)).isEqualTo(executorName); - assertThat(point.getAttributes().get(EXECUTOR_OWNER_NAME)).isEqualTo(ownerName); - assertThat(point.getAttributes().get(EXECUTOR_TYPE)) - .isEqualTo(ThreadPoolExecutor.class.getName()); + private static void assertNoExecutorMetrics(String executorName, String ownerName) { + testing.clearData(); + await() + .untilAsserted( + () -> + assertThat(testing.metrics()) + .filteredOn( + metric -> + metric + .getInstrumentationScopeInfo() + .getName() + .equals(INSTRUMENTATION_NAME)) + .allSatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .noneMatch( + point -> + executorName.equals( + point + .getAttributes() + .get(stringKey("jvm.executor.name"))) + && ownerName.equals( + point + .getAttributes() + .get(stringKey("jvm.executor.owner.name")))))); } private static final class NamedThreadFactory implements ThreadFactory { @@ -488,39 +440,6 @@ private int createdThreadCount() { } } - private static final class StartedThreadFactory implements ThreadFactory { - private final String namePrefix; - private final AtomicInteger sequence = new AtomicInteger(); - - private StartedThreadFactory(String namePrefix) { - this.namePrefix = namePrefix; - } - - @Override - public Thread newThread(Runnable runnable) { - Thread thread = new Thread(runnable, namePrefix + "-" + sequence.incrementAndGet()); - thread.start(); - return thread; - } - } - - private static final class BlockingThreadFactory implements ThreadFactory { - private final CountDownLatch started = new CountDownLatch(1); - private final CountDownLatch release = new CountDownLatch(1); - - @Override - public Thread newThread(Runnable runnable) { - started.countDown(); - try { - release.await(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new AssertionError(e); - } - return new Thread(runnable, "reregister-pool"); - } - } - private static class UnlistedThreadPoolExecutor extends ThreadPoolExecutor { private UnlistedThreadPoolExecutor(ThreadFactory threadFactory) { diff --git a/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java b/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java index 0c857d435d91..5ae058dd39a2 100644 --- a/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java +++ b/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java @@ -8,8 +8,8 @@ import static io.opentelemetry.api.common.AttributeKey.stringKey; import static java.util.concurrent.TimeUnit.SECONDS; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; -import io.opentelemetry.api.common.AttributeKey; import io.opentelemetry.instrumentation.test.utils.GcUtils; import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension; import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; @@ -28,26 +28,24 @@ class ThreadPerTaskExecutorMetricsTest { private static final String INSTRUMENTATION_NAME = "io.opentelemetry.executors"; + private static final String DEFAULT_OWNER_NAME = "unknown"; private static final String EXECUTOR_NAME = "thread-per-task-*"; private static final String EXECUTOR_TYPE = "java.util.concurrent.ThreadPerTaskExecutor"; - private static final AttributeKey EXECUTOR_NAME_KEY = stringKey("jvm.executor.name"); - private static final AttributeKey EXECUTOR_OWNER_NAME_KEY = - stringKey("jvm.executor.owner.name"); - private static final AttributeKey EXECUTOR_TYPE_KEY = stringKey("jvm.executor.type"); - private static final AttributeKey EXECUTOR_STATE_KEY = stringKey("jvm.executor.state"); - @RegisterExtension static final InstrumentationExtension testing = AgentInstrumentationExtension.create(); @Test void recordsActiveThreadCountAndUnregistersOnShutdown() throws Exception { - ExecutorService executor = - Executors.newThreadPerTaskExecutor(new NamedThreadFactory("thread-per-task")); + NamedThreadFactory threadFactory = new NamedThreadFactory("thread-per-task"); + ExecutorService executor = Executors.newThreadPerTaskExecutor(threadFactory); CountDownLatch started = new CountDownLatch(1); CountDownLatch release = new CountDownLatch(1); try { + assertThat(threadFactory.createdThreadCount()).isZero(); + assertNoThreadPerTaskMetrics(EXECUTOR_NAME, DEFAULT_OWNER_NAME); + Future future = executor.submit( () -> { @@ -61,45 +59,43 @@ void recordsActiveThreadCountAndUnregistersOnShutdown() throws Exception { }); assertThat(started.await(10, SECONDS)).isTrue(); - assertActiveThreadCount(1); - assertNoUnsupportedMetrics(); + assertThat(threadFactory.createdThreadCount()).isEqualTo(1); + JvmExecutorMetricsAssertions.create( + testing, INSTRUMENTATION_NAME, EXECUTOR_NAME, DEFAULT_OWNER_NAME, EXECUTOR_TYPE) + .withActiveThreads(1) + .assertExecutorEmitsMetrics(); release.countDown(); future.get(10, SECONDS); - assertActiveThreadCount(0); + JvmExecutorMetricsAssertions.create( + testing, INSTRUMENTATION_NAME, EXECUTOR_NAME, DEFAULT_OWNER_NAME, EXECUTOR_TYPE) + .withActiveThreads(0) + .assertExecutorEmitsMetrics(); } finally { release.countDown(); executor.shutdown(); assertThat(executor.awaitTermination(10, SECONDS)).isTrue(); } - testing.clearData(); - assertNoThreadPerTaskMetric("jvm.executor.thread.count"); + assertNoThreadPerTaskMetrics(EXECUTOR_NAME, DEFAULT_OWNER_NAME); } @Test - void unregistersOnClose() { + void unregistersOnClose() throws Exception { ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor(); try { - testing.waitAndAssertMetrics( - INSTRUMENTATION_NAME, - "jvm.executor.thread.count", - metrics -> - metrics.anySatisfy( - metric -> - assertThat(metric.getLongSumData().getPoints()) - .anySatisfy( - point -> - assertThat(point.getAttributes().get(EXECUTOR_TYPE_KEY)) - .isEqualTo(EXECUTOR_TYPE)))); + executor.submit(() -> {}).get(10, SECONDS); + JvmExecutorMetricsAssertions.create( + testing, INSTRUMENTATION_NAME, DEFAULT_OWNER_NAME, DEFAULT_OWNER_NAME, EXECUTOR_TYPE) + .withActiveThreads(0) + .assertExecutorEmitsMetrics(); executor.close(); } finally { executor.shutdown(); } - testing.clearData(); - assertNoThreadPerTaskMetric("jvm.executor.thread.count"); + assertNoThreadPerTaskMetrics(DEFAULT_OWNER_NAME, DEFAULT_OWNER_NAME); } @Test @@ -111,78 +107,57 @@ void doesNotRecordMetricsWhenUnclosedExecutorIsCollected() throws Exception { GcUtils.awaitGc(executorRef, Duration.ofSeconds(10)); - testing.clearData(); - assertNoThreadPerTaskMetric("jvm.executor.thread.count"); + assertNoThreadPerTaskMetrics("collected-thread-per-task-*", DEFAULT_OWNER_NAME); } @Test - void reregistersOwnerAndThreadNameNormalization() { + void reregistersOwnerAndThreadNameNormalization() throws Exception { NamedThreadFactory threadFactory = new NamedThreadFactory("thread-per-task-42-worker"); ExecutorService executor = Executors.newThreadPerTaskExecutor(threadFactory); try { - assertActiveThreadCount(0, "thread-per-task-*-worker-*", "unknown"); - assertThat(threadFactory.createdThreadCount()).isEqualTo(1); + assertThat(threadFactory.createdThreadCount()).isZero(); + assertNoThreadPerTaskMetrics("thread-per-task-*-worker-*", DEFAULT_OWNER_NAME); ThreadPoolExecutorMetrics.reregister(executor, "tomcat", "trailing"); + executor.submit(() -> {}).get(10, SECONDS); testing.clearData(); - assertActiveThreadCount(0, "thread-per-task-42-worker-*", "tomcat"); + JvmExecutorMetricsAssertions.create( + testing, INSTRUMENTATION_NAME, "thread-per-task-42-worker-*", "tomcat", EXECUTOR_TYPE) + .withActiveThreads(0) + .assertExecutorEmitsMetrics(); assertThat(threadFactory.createdThreadCount()).isEqualTo(1); } finally { executor.shutdown(); } } - private static void assertActiveThreadCount(long expectedValue) { - assertActiveThreadCount(expectedValue, EXECUTOR_NAME, "unknown"); - } - - private static void assertActiveThreadCount( - long expectedValue, String executorName, String ownerName) { - testing.waitAndAssertMetrics( - INSTRUMENTATION_NAME, - "jvm.executor.thread.count", - metrics -> - metrics.anySatisfy( - metric -> - assertThat(metric.getLongSumData().getPoints()) - .anySatisfy( - point -> { - assertThat(metric.getUnit()).isEqualTo("{thread}"); - assertThat(point.getAttributes().get(EXECUTOR_NAME_KEY)) - .isEqualTo(executorName); - assertThat(point.getAttributes().get(EXECUTOR_OWNER_NAME_KEY)) - .isEqualTo(ownerName); - assertThat(point.getAttributes().get(EXECUTOR_TYPE_KEY)) - .isEqualTo(EXECUTOR_TYPE); - assertThat(point.getAttributes().get(EXECUTOR_STATE_KEY)) - .isEqualTo("active"); - assertThat(point.getValue()).isEqualTo(expectedValue); - }))); - } - - private static void assertNoUnsupportedMetrics() { - assertNoThreadPerTaskMetric("jvm.executor.thread.core"); - assertNoThreadPerTaskMetric("jvm.executor.thread.max"); - assertNoThreadPerTaskMetric("jvm.executor.queue.size"); - assertNoThreadPerTaskMetric("jvm.executor.queue.remaining"); - assertNoThreadPerTaskMetric("jvm.executor.task.completed"); - assertNoThreadPerTaskMetric("jvm.executor.task.rejected"); - } - - private static void assertNoThreadPerTaskMetric(String metricName) { - testing.waitAndAssertMetrics( - INSTRUMENTATION_NAME, - metricName, - metrics -> - metrics.allSatisfy( - metric -> - assertThat(metric.getLongSumData().getPoints()) - .noneMatch( - point -> - EXECUTOR_TYPE.equals( - point.getAttributes().get(EXECUTOR_TYPE_KEY))))); + private static void assertNoThreadPerTaskMetrics(String executorName, String ownerName) { + testing.clearData(); + await() + .untilAsserted( + () -> + assertThat(testing.metrics()) + .filteredOn( + metric -> + metric + .getInstrumentationScopeInfo() + .getName() + .equals(INSTRUMENTATION_NAME)) + .allSatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .noneMatch( + point -> + executorName.equals( + point + .getAttributes() + .get(stringKey("jvm.executor.name"))) + && ownerName.equals( + point + .getAttributes() + .get(stringKey("jvm.executor.owner.name")))))); } private static final class NamedThreadFactory implements ThreadFactory { diff --git a/instrumentation/executors/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/JvmExecutorMetricsAssertions.java b/instrumentation/executors/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/JvmExecutorMetricsAssertions.java new file mode 100644 index 000000000000..58d02809c077 --- /dev/null +++ b/instrumentation/executors/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/JvmExecutorMetricsAssertions.java @@ -0,0 +1,310 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.executors; + +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat; + +import com.google.errorprone.annotations.CanIgnoreReturnValue; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.testing.assertj.LongPointAssert; +import io.opentelemetry.sdk.testing.assertj.LongSumAssert; +import java.util.ArrayList; +import java.util.List; +import java.util.function.Consumer; +import javax.annotation.Nullable; + +public class JvmExecutorMetricsAssertions { + + private static final AttributeKey EXECUTOR_NAME_KEY = stringKey("jvm.executor.name"); + private static final AttributeKey EXECUTOR_OWNER_NAME_KEY = + stringKey("jvm.executor.owner.name"); + private static final AttributeKey EXECUTOR_TYPE_KEY = stringKey("jvm.executor.type"); + private static final AttributeKey EXECUTOR_STATE_KEY = stringKey("jvm.executor.state"); + + public static JvmExecutorMetricsAssertions create( + InstrumentationExtension testing, + String instrumentationName, + String executorName, + String ownerName, + String executorType) { + return new JvmExecutorMetricsAssertions( + testing, instrumentationName, executorName, ownerName, executorType); + } + + private final InstrumentationExtension testing; + private final String instrumentationName; + private final String executorName; + private final String ownerName; + private final String executorType; + + @Nullable private Long expectedActiveThreads; + @Nullable private Long expectedIdleThreads; + @Nullable private Long expectedCoreThreads; + @Nullable private Long expectedMaxThreads; + @Nullable private Long expectedQueueSize; + @Nullable private Long expectedQueueRemaining; + @Nullable private Long expectedCompletedTasks; + @Nullable private Long expectedRejectedTasks; + + JvmExecutorMetricsAssertions( + InstrumentationExtension testing, + String instrumentationName, + String executorName, + String ownerName, + String executorType) { + this.testing = testing; + this.instrumentationName = instrumentationName; + this.executorName = executorName; + this.ownerName = ownerName; + this.executorType = executorType; + } + + @CanIgnoreReturnValue + public JvmExecutorMetricsAssertions withActiveThreads(long value) { + expectedActiveThreads = value; + return this; + } + + @CanIgnoreReturnValue + public JvmExecutorMetricsAssertions withIdleThreads(long value) { + expectedIdleThreads = value; + return this; + } + + @CanIgnoreReturnValue + public JvmExecutorMetricsAssertions withCoreThreads(long value) { + expectedCoreThreads = value; + return this; + } + + @CanIgnoreReturnValue + public JvmExecutorMetricsAssertions withMaxThreads(long value) { + expectedMaxThreads = value; + return this; + } + + @CanIgnoreReturnValue + public JvmExecutorMetricsAssertions withQueueSize(long value) { + expectedQueueSize = value; + return this; + } + + @CanIgnoreReturnValue + public JvmExecutorMetricsAssertions withQueueRemaining(long value) { + expectedQueueRemaining = value; + return this; + } + + @CanIgnoreReturnValue + public JvmExecutorMetricsAssertions withCompletedTasks(long value) { + expectedCompletedTasks = value; + return this; + } + + @CanIgnoreReturnValue + public JvmExecutorMetricsAssertions withRejectedTasks(long value) { + expectedRejectedTasks = value; + return this; + } + + public void assertExecutorEmitsMetrics() { + if (expectedActiveThreads == null + && expectedIdleThreads == null + && expectedCoreThreads == null + && expectedMaxThreads == null + && expectedQueueSize == null + && expectedQueueRemaining == null + && expectedCompletedTasks == null + && expectedRejectedTasks == null) { + throw new IllegalStateException("At least one expected executor metric value must be set."); + } + + if (expectedActiveThreads != null || expectedIdleThreads != null) { + verifyThreadCount(); + } + if (expectedCoreThreads != null) { + verifyCoreThreads(expectedCoreThreads); + } + if (expectedMaxThreads != null) { + verifyMaxThreads(expectedMaxThreads); + } + if (expectedQueueSize != null) { + verifyQueueSize(expectedQueueSize); + } + if (expectedQueueRemaining != null) { + verifyQueueRemaining(expectedQueueRemaining); + } + if (expectedCompletedTasks != null) { + verifyCompletedTasks(expectedCompletedTasks); + } + if (expectedRejectedTasks != null) { + verifyRejectedTasks(expectedRejectedTasks); + } + } + + private void verifyThreadCount() { + List> pointAssertions = new ArrayList<>(2); + Long activeThreads = expectedActiveThreads; + if (activeThreads != null) { + pointAssertions.add(point -> verifyThreadCountPoint(point, "active", activeThreads)); + } + Long idleThreads = expectedIdleThreads; + if (idleThreads != null) { + pointAssertions.add(point -> verifyThreadCountPoint(point, "idle", idleThreads)); + } + + testing.waitAndAssertMetrics( + instrumentationName, + "jvm.executor.thread.count", + metrics -> metrics.anySatisfy(metric -> verifyThreadCountMetric(metric, pointAssertions))); + } + + private static void verifyThreadCountMetric( + MetricData metric, List> pointAssertions) { + assertThat(metric) + .hasUnit("{thread}") + .hasDescription("The number of executor threads that are currently in the described state.") + .hasLongSumSatisfying( + sum -> sum.isNotMonotonic().containsPointsSatisfying(pointAssertions)); + } + + private void verifyThreadCountPoint(LongPointAssert point, String state, long expectedValue) { + point + .hasAttributes( + Attributes.of( + EXECUTOR_NAME_KEY, + executorName, + EXECUTOR_OWNER_NAME_KEY, + ownerName, + EXECUTOR_TYPE_KEY, + executorType, + EXECUTOR_STATE_KEY, + state)) + .hasValue(expectedValue); + } + + private void verifyCoreThreads(long expectedValue) { + testing.waitAndAssertMetrics( + instrumentationName, + "jvm.executor.thread.core", + metrics -> + metrics.anySatisfy( + metric -> + verifyExecutorMetric( + metric, + "{thread}", + "The core number of threads configured for the executor.", + false, + expectedValue))); + } + + private void verifyMaxThreads(long expectedValue) { + testing.waitAndAssertMetrics( + instrumentationName, + "jvm.executor.thread.max", + metrics -> + metrics.anySatisfy( + metric -> + verifyExecutorMetric( + metric, + "{thread}", + "The maximum number of threads allowed for the executor.", + false, + expectedValue))); + } + + private void verifyQueueSize(long expectedValue) { + testing.waitAndAssertMetrics( + instrumentationName, + "jvm.executor.queue.size", + metrics -> + metrics.anySatisfy( + metric -> + verifyExecutorMetric( + metric, + "{task}", + "The number of tasks currently queued for execution.", + false, + expectedValue))); + } + + private void verifyQueueRemaining(long expectedValue) { + testing.waitAndAssertMetrics( + instrumentationName, + "jvm.executor.queue.remaining", + metrics -> + metrics.anySatisfy( + metric -> + verifyExecutorMetric( + metric, + "{task}", + "The remaining task capacity of the executor queue.", + false, + expectedValue))); + } + + private void verifyCompletedTasks(long expectedValue) { + testing.waitAndAssertMetrics( + instrumentationName, + "jvm.executor.task.completed", + metrics -> + metrics.anySatisfy( + metric -> + verifyExecutorMetric( + metric, + "{task}", + "The number of tasks completed by the executor.", + true, + expectedValue))); + } + + private void verifyRejectedTasks(long expectedValue) { + testing.waitAndAssertMetrics( + instrumentationName, + "jvm.executor.task.rejected", + metrics -> + metrics.anySatisfy( + metric -> + verifyExecutorMetric( + metric, + "{task}", + "The number of tasks rejected by the executor.", + true, + expectedValue))); + } + + private void verifyExecutorMetric( + MetricData metric, String unit, String description, boolean monotonic, long expectedValue) { + assertThat(metric) + .hasUnit(unit) + .hasDescription(description) + .hasLongSumSatisfying(sum -> verifyExecutorAttributes(sum, monotonic, expectedValue)); + } + + private void verifyExecutorAttributes(LongSumAssert sum, boolean monotonic, long expectedValue) { + if (monotonic) { + sum.isMonotonic(); + } else { + sum.isNotMonotonic(); + } + sum.containsPointsSatisfying( + point -> + point + .hasAttributes( + Attributes.of( + EXECUTOR_NAME_KEY, + executorName, + EXECUTOR_OWNER_NAME_KEY, + ownerName, + EXECUTOR_TYPE_KEY, + executorType)) + .hasValue(expectedValue)); + } +} From 6b031d931ac1b4f00099b6afcf77c83915d228f4 Mon Sep 17 00:00:00 2001 From: yaoyinglong <906271196@qq.com> Date: Mon, 20 Jul 2026 20:49:34 +0800 Subject: [PATCH 3/5] fix: failed test cases --- instrumentation/executors/README.md | 10 ++-- .../ThreadPoolExecutorMetricsTest.java | 47 +++++++++---------- .../ThreadPerTaskExecutorMetricsTest.java | 47 +++++++++---------- instrumentation/executors/metadata.yaml | 2 +- 4 files changed, 52 insertions(+), 54 deletions(-) diff --git a/instrumentation/executors/README.md b/instrumentation/executors/README.md index ff6fc8f10e28..a025fde98946 100644 --- a/instrumentation/executors/README.md +++ b/instrumentation/executors/README.md @@ -1,7 +1,7 @@ # Settings for the executors instrumentation -| System property | Type | Default | Description | -| -------------------------------------------- | ------- | ------- | -------------------------------------------------------------------------- | -| `otel.instrumentation.executors.include` | List | Empty | List of `Executor` subclasses to be instrumented. | -| `otel.instrumentation.executors.include-all` | Boolean | `false` | Whether to instrument all classes that implement the `Executor` interface. | -| `otel.instrumentation.executors.name-normalization` | String | `all` | Replaces all consecutive digits in executor thread names with `*` for `all`; `trailing` replaces only trailing digits; other values use `all`. | +| System property | Type | Default | Description | +| --------------------------------------------------- | ------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------- | +| `otel.instrumentation.executors.include` | List | Empty | List of `Executor` subclasses to be instrumented. | +| `otel.instrumentation.executors.include-all` | Boolean | `false` | Whether to instrument all classes that implement the `Executor` interface. | +| `otel.instrumentation.executors.name-normalization` | String | `all` | Replaces all consecutive digits in executor thread names with `*` for `all`; `trailing` replaces only trailing digits; other values use `all`. | diff --git a/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java b/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java index 5422e1dee0dd..6c4de2918232 100644 --- a/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java +++ b/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java @@ -10,7 +10,6 @@ import static java.util.concurrent.TimeUnit.SECONDS; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.awaitility.Awaitility.await; import io.opentelemetry.instrumentation.test.utils.GcUtils; import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension; @@ -397,29 +396,29 @@ void recordsMetricsWhenExecutorNamesCollide() throws Exception { private static void assertNoExecutorMetrics(String executorName, String ownerName) { testing.clearData(); - await() - .untilAsserted( - () -> - assertThat(testing.metrics()) - .filteredOn( - metric -> - metric - .getInstrumentationScopeInfo() - .getName() - .equals(INSTRUMENTATION_NAME)) - .allSatisfy( - metric -> - assertThat(metric.getLongSumData().getPoints()) - .noneMatch( - point -> - executorName.equals( - point - .getAttributes() - .get(stringKey("jvm.executor.name"))) - && ownerName.equals( - point - .getAttributes() - .get(stringKey("jvm.executor.owner.name")))))); + testing + .getOpenTelemetry() + .getMeter("test") + .counterBuilder("test.executor.metrics.collection") + .build() + .add(1); + testing.waitAndAssertMetrics( + "test", "test.executor.metrics.collection", metrics -> metrics.isNotEmpty()); + + assertThat(testing.metrics()) + .filteredOn( + metric -> metric.getInstrumentationScopeInfo().getName().equals(INSTRUMENTATION_NAME)) + .allSatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .noneMatch( + point -> + executorName.equals( + point.getAttributes().get(stringKey("jvm.executor.name"))) + && ownerName.equals( + point + .getAttributes() + .get(stringKey("jvm.executor.owner.name"))))); } private static final class NamedThreadFactory implements ThreadFactory { diff --git a/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java b/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java index 5ae058dd39a2..fa345f333a11 100644 --- a/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java +++ b/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java @@ -8,7 +8,6 @@ import static io.opentelemetry.api.common.AttributeKey.stringKey; import static java.util.concurrent.TimeUnit.SECONDS; import static org.assertj.core.api.Assertions.assertThat; -import static org.awaitility.Awaitility.await; import io.opentelemetry.instrumentation.test.utils.GcUtils; import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension; @@ -135,29 +134,29 @@ void reregistersOwnerAndThreadNameNormalization() throws Exception { private static void assertNoThreadPerTaskMetrics(String executorName, String ownerName) { testing.clearData(); - await() - .untilAsserted( - () -> - assertThat(testing.metrics()) - .filteredOn( - metric -> - metric - .getInstrumentationScopeInfo() - .getName() - .equals(INSTRUMENTATION_NAME)) - .allSatisfy( - metric -> - assertThat(metric.getLongSumData().getPoints()) - .noneMatch( - point -> - executorName.equals( - point - .getAttributes() - .get(stringKey("jvm.executor.name"))) - && ownerName.equals( - point - .getAttributes() - .get(stringKey("jvm.executor.owner.name")))))); + testing + .getOpenTelemetry() + .getMeter("test") + .counterBuilder("test.executor.metrics.collection") + .build() + .add(1); + testing.waitAndAssertMetrics( + "test", "test.executor.metrics.collection", metrics -> metrics.isNotEmpty()); + + assertThat(testing.metrics()) + .filteredOn( + metric -> metric.getInstrumentationScopeInfo().getName().equals(INSTRUMENTATION_NAME)) + .allSatisfy( + metric -> + assertThat(metric.getLongSumData().getPoints()) + .noneMatch( + point -> + executorName.equals( + point.getAttributes().get(stringKey("jvm.executor.name"))) + && ownerName.equals( + point + .getAttributes() + .get(stringKey("jvm.executor.owner.name"))))); } private static final class NamedThreadFactory implements ThreadFactory { diff --git a/instrumentation/executors/metadata.yaml b/instrumentation/executors/metadata.yaml index 4b0d33c33e60..411654d7f979 100644 --- a/instrumentation/executors/metadata.yaml +++ b/instrumentation/executors/metadata.yaml @@ -23,7 +23,7 @@ configurations: type: boolean default: false - name: otel.instrumentation.executors.name-normalization - declarative_name: java.executors.thread_name_normalization + declarative_name: java.executors.name_normalization description: > Determines how executor thread names are normalized. all replaces every group of digits with *; trailing replaces only the final group; other values act as all. From 994e465e44b72c4cc4f2c2b0c9304098dc859a6a Mon Sep 17 00:00:00 2001 From: yaoyinglong <906271196@qq.com> Date: Tue, 21 Jul 2026 13:14:02 +0800 Subject: [PATCH 4/5] fix(executors): disable executor metrics by default --- instrumentation/executors/README.md | 1 + .../executors/javaagent/build.gradle.kts | 1 + .../ExecutorsInstrumentationModule.java | 4 --- ...ExecutorsMetricsInstrumentationModule.java | 35 +++++++++++++++++++ .../executors/jdk21-testing/build.gradle.kts | 1 + instrumentation/executors/metadata.yaml | 5 +++ 6 files changed, 43 insertions(+), 4 deletions(-) create mode 100644 instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ExecutorsMetricsInstrumentationModule.java diff --git a/instrumentation/executors/README.md b/instrumentation/executors/README.md index a025fde98946..088fbfe3cbb4 100644 --- a/instrumentation/executors/README.md +++ b/instrumentation/executors/README.md @@ -5,3 +5,4 @@ | `otel.instrumentation.executors.include` | List | Empty | List of `Executor` subclasses to be instrumented. | | `otel.instrumentation.executors.include-all` | Boolean | `false` | Whether to instrument all classes that implement the `Executor` interface. | | `otel.instrumentation.executors.name-normalization` | String | `all` | Replaces all consecutive digits in executor thread names with `*` for `all`; `trailing` replaces only trailing digits; other values use `all`. | +| `otel.instrumentation.executors-metrics.enabled` | Boolean | `false` | Enables executor metrics instrumentation. | diff --git a/instrumentation/executors/javaagent/build.gradle.kts b/instrumentation/executors/javaagent/build.gradle.kts index 2344d0ccafd5..a8a57ca9428e 100644 --- a/instrumentation/executors/javaagent/build.gradle.kts +++ b/instrumentation/executors/javaagent/build.gradle.kts @@ -77,6 +77,7 @@ tasks { jvmArgs( "-Dotel.instrumentation.executors.include=io.opentelemetry.javaagent.instrumentation.executors.ExecutorInstrumentationTest\$CustomThreadPoolExecutor,io.opentelemetry.javaagent.instrumentation.executors.ThreadPoolExecutorTest\$RunnableCheckingThreadPoolExecutor" ) + jvmArgs("-Dotel.instrumentation.executors-metrics.enabled=true") jvmArgs("-Djava.awt.headless=true") } diff --git a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/ExecutorsInstrumentationModule.java b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/ExecutorsInstrumentationModule.java index 86ba74e51c27..b7e57eeee455 100644 --- a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/ExecutorsInstrumentationModule.java +++ b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/ExecutorsInstrumentationModule.java @@ -11,8 +11,6 @@ import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule; import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; import io.opentelemetry.javaagent.extension.instrumentation.internal.EarlyInstrumentationModule; -import io.opentelemetry.javaagent.instrumentation.executors.metrics.ThreadPerTaskExecutorMetricsInstrumentation; -import io.opentelemetry.javaagent.instrumentation.executors.metrics.ThreadPoolExecutorMetricsInstrumentation; import java.util.List; @AutoService({InstrumentationModule.class, EarlyInstrumentationModule.class}) @@ -32,8 +30,6 @@ public List typeInstrumentations() { new JavaForkJoinTaskInstrumentation(), new RunnableInstrumentation(), new ThreadPoolExtendingExecutorInstrumentation(), - new ThreadPoolExecutorMetricsInstrumentation(), - new ThreadPerTaskExecutorMetricsInstrumentation(), new VirtualThreadInstrumentation(), new StructuredTaskScopeInstrumentation()); } diff --git a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ExecutorsMetricsInstrumentationModule.java b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ExecutorsMetricsInstrumentationModule.java new file mode 100644 index 000000000000..07005b92befe --- /dev/null +++ b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ExecutorsMetricsInstrumentationModule.java @@ -0,0 +1,35 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.executors.metrics; + +import static java.util.Arrays.asList; + +import com.google.auto.service.AutoService; +import io.opentelemetry.javaagent.extension.instrumentation.InstrumentationModule; +import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; +import io.opentelemetry.javaagent.extension.instrumentation.internal.EarlyInstrumentationModule; +import java.util.List; + +@AutoService({InstrumentationModule.class, EarlyInstrumentationModule.class}) +public class ExecutorsMetricsInstrumentationModule extends InstrumentationModule + implements EarlyInstrumentationModule { + + public ExecutorsMetricsInstrumentationModule() { + super("executors-metrics", "executors"); + } + + @Override + public boolean defaultEnabled() { + return false; + } + + @Override + public List typeInstrumentations() { + return asList( + new ThreadPoolExecutorMetricsInstrumentation(), + new ThreadPerTaskExecutorMetricsInstrumentation()); + } +} diff --git a/instrumentation/executors/jdk21-testing/build.gradle.kts b/instrumentation/executors/jdk21-testing/build.gradle.kts index 4cc9f92e53bc..0db7e8bd4080 100644 --- a/instrumentation/executors/jdk21-testing/build.gradle.kts +++ b/instrumentation/executors/jdk21-testing/build.gradle.kts @@ -36,6 +36,7 @@ tasks.test { // needed for VirtualThreadTest jvmArgs("--add-opens=java.base/java.lang=ALL-UNNAMED") jvmArgs("-XX:+IgnoreUnrecognizedVMOptions") + jvmArgs("-Dotel.instrumentation.executors-metrics.enabled=true") // needed for structured concurrency test jvmArgs("--enable-preview") } diff --git a/instrumentation/executors/metadata.yaml b/instrumentation/executors/metadata.yaml index 411654d7f979..7b49db92cdca 100644 --- a/instrumentation/executors/metadata.yaml +++ b/instrumentation/executors/metadata.yaml @@ -32,3 +32,8 @@ configurations: examples: - all - trailing + - name: otel.instrumentation.executors-metrics.enabled + declarative_name: java.executors_metrics.enabled + description: Enables executor metrics instrumentation. + type: boolean + default: false From 1421ab9eda792396f16a73d822046c60597035fb Mon Sep 17 00:00:00 2001 From: yaoyinglong <906271196@qq.com> Date: Tue, 21 Jul 2026 17:00:32 +0800 Subject: [PATCH 5/5] fix: check-javaagent-suppression-keys --- .../metrics/ExecutorsMetricsInstrumentationModule.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ExecutorsMetricsInstrumentationModule.java b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ExecutorsMetricsInstrumentationModule.java index 07005b92befe..8aac854c2594 100644 --- a/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ExecutorsMetricsInstrumentationModule.java +++ b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ExecutorsMetricsInstrumentationModule.java @@ -18,7 +18,7 @@ public class ExecutorsMetricsInstrumentationModule extends InstrumentationModule implements EarlyInstrumentationModule { public ExecutorsMetricsInstrumentationModule() { - super("executors-metrics", "executors"); + super("executors", "executors-metrics"); } @Override