diff --git a/instrumentation/executors/README.md b/instrumentation/executors/README.md index deb0e9ed46cc..088fbfe3cbb4 100644 --- a/instrumentation/executors/README.md +++ b/instrumentation/executors/README.md @@ -1,6 +1,8 @@ # 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. | +| 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`. | +| `otel.instrumentation.executors-metrics.enabled` | Boolean | `false` | Enables executor metrics instrumentation. | 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..152c77709860 --- /dev/null +++ b/instrumentation/executors/bootstrap/src/main/java/io/opentelemetry/javaagent/bootstrap/executors/ThreadPoolExecutorMetrics.java @@ -0,0 +1,300 @@ +/* + * 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.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 preRegister(ThreadPoolExecutor executor) { + preRegister(executor, emptySet()); + } + + 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; + Registration registration = registrations.get(executor); + if (registration != null) { + registration.reregister(executor, executorOwnerName, threadNameNormalization); + } + } + + public static void onThreadFactoryChanged(Executor executor) { + Registration registration = registrations.get(executor); + if (registration != null) { + 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) { + Registration registration = registrations.get(executor); + if (registration != null) { + registration.recordRejectedTask(); + } + } + + public static void unregister(Executor executor) { + Registration registration; + synchronized (registrations) { + registration = registrations.get(executor); + if (registration != null) { + registrations.remove(executor); + } + } + if (registration != null) { + registration.close(); + } + } + + 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 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; + } + } + + private void onWorkerThreadStarted(Executor executor, Thread thread) { + if (!awaitingWorkerThread) { + return; + } + + @Nullable AtomicReference previous; + synchronized (this) { + if (closed || !awaitingWorkerThread) { + return; + } + + Set threads = threadsRef.get(); + if (threads == null) { + return; + } + + awaitingWorkerThread = false; + previous = callback; + String newThreadName = thread.getName(); + registerMetrics( + executor, threads, newThreadName, executorName(newThreadName, threadNameNormalization)); + } + + if (previous != null) { + closeCallback(previous); + } + } + + 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); + } + + closeCallback(previous); + } + + private synchronized void recordRejectedTask() { + if (callback != null && rejectedTasks != null) { + rejectedTasks.add(1, attributes); + } + } + + private void close() { + @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; + } + } + + private ThreadPoolExecutorMetrics() {} +} diff --git a/instrumentation/executors/javaagent/build.gradle.kts b/instrumentation/executors/javaagent/build.gradle.kts index 1b3b676c5dec..a8a57ca9428e 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") + } + } + } + } } } @@ -54,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/metrics/ExecutorsMetricsInstrumentationModule.java b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ExecutorsMetricsInstrumentationModule.java new file mode 100644 index 000000000000..8aac854c2594 --- /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", "executors-metrics"); + } + + @Override + public boolean defaultEnabled() { + return false; + } + + @Override + public List typeInstrumentations() { + return asList( + new ThreadPoolExecutorMetricsInstrumentation(), + new ThreadPerTaskExecutorMetricsInstrumentation()); + } +} 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..d22d877c91de --- /dev/null +++ b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPerTaskExecutorMetricsInstrumentation.java @@ -0,0 +1,78 @@ +/* + * 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.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.Set; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +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( + named("start").and(takesArgument(0, Thread.class)).and(takesArguments(1)), + getClass().getName() + "$StartAdvice"); + 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.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); + } + } + + @SuppressWarnings("unused") + public static class ShutdownAdvice { + + @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) + 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 new file mode 100644 index 000000000000..dd0d79f1f2a7 --- /dev/null +++ b/instrumentation/executors/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/executors/metrics/ThreadPoolExecutorMetricsInstrumentation.java @@ -0,0 +1,117 @@ +/* + * 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.ExecutorService; +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("runWorker") + .and(takesArguments(1)) + .and(methodIsDeclaredByType(named(ThreadPoolExecutor.class.getName()))), + getClass().getName() + "$RunWorkerAdvice"); + 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) { + if (!(executor instanceof ScheduledThreadPoolExecutor)) { + ThreadPoolExecutorMetrics.preRegister(executor); + } + } + } + + @SuppressWarnings("unused") + public static class SetThreadFactoryAdvice { + + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onExit(@Advice.This ThreadPoolExecutor executor) { + if (!(executor instanceof ScheduledThreadPoolExecutor)) { + 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()); + } + } + } + + @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 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 new file mode 100644 index 000000000000..6c4de2918232 --- /dev/null +++ b/instrumentation/executors/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPoolExecutorMetricsTest.java @@ -0,0 +1,451 @@ +/* + * 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.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.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; +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 THREAD_POOL_EXECUTOR_TYPE = ThreadPoolExecutor.class.getName(); + private static final String EXPECTED_THREAD_NAME_NORMALIZATION = + "test.name-normalization.expected"; + + @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 = + 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); + + 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(); + } + + assertNoExecutorMetrics("metrics-pool-*", DEFAULT_OWNER_NAME); + } + + @Test + void skipsScheduledThreadPoolExecutor() { + ScheduledThreadPoolExecutor executor = + new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("scheduled-pool")); + + try { + assertNoExecutorMetrics("scheduled-pool-*", DEFAULT_OWNER_NAME); + } finally { + executor.shutdownNow(); + } + } + + @Test + void retainsMetricsWhenOverriddenShutdownDoesNotShutdownExecutor() { + UnlistedThreadPoolExecutor executor = + new UnlistedThreadPoolExecutor(new NamedThreadFactory("unlisted-pool")); + + try { + 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(); + + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + "unlisted-pool-*", + DEFAULT_OWNER_NAME, + UnlistedThreadPoolExecutor.class.getName()) + .withCoreThreads(0) + .assertExecutorEmitsMetrics(); + } 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(); + assertNoExecutorMetrics("failed-pool-*", DEFAULT_OWNER_NAME); + } + + @Test + void exportsMetricsOnlyAfterWorkerStarts() throws Exception { + NamedThreadFactory threadFactory = new NamedThreadFactory("started-pool"); + ThreadPoolExecutor executor = + new ThreadPoolExecutor(1, 1, 0, MILLISECONDS, new ArrayBlockingQueue<>(1), threadFactory); + + try { + 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(); + } + } + + @Test + void doesNotReregisterAfterUnregister() throws Exception { + ThreadPoolExecutor executor = + new ThreadPoolExecutor( + 1, + 1, + 1, + SECONDS, + new ArrayBlockingQueue<>(1), + new NamedThreadFactory("reregister-pool")); + + try { + executor.prestartCoreThread(); + JvmExecutorMetricsAssertions.create( + testing, + INSTRUMENTATION_NAME, + "reregister-pool-*", + DEFAULT_OWNER_NAME, + THREAD_POOL_EXECUTOR_TYPE) + .withCoreThreads(1) + .assertExecutorEmitsMetrics(); + + ThreadPoolExecutorMetrics.unregister(executor); + ThreadPoolExecutorMetrics.reregister(executor, "tomcat", "trailing"); + + assertNoExecutorMetrics("reregister-pool-*", DEFAULT_OWNER_NAME); + assertNoExecutorMetrics("reregister-pool-*", "tomcat"); + } finally { + 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"))); + + assertNoExecutorMetrics("collected-pool-*", DEFAULT_OWNER_NAME); + GcUtils.awaitGc(executorRef, Duration.ofSeconds(10)); + + assertNoExecutorMetrics("collected-pool-*", DEFAULT_OWNER_NAME); + } + + @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-*"; + + 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(); + } + } + + @Test + void reregistersOwnerAndThreadFactory() throws Exception { + NamedThreadFactory originalThreadFactory = new NamedThreadFactory("original-pool-42-worker"); + ThreadPoolExecutor executor = + new ThreadPoolExecutor( + 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-*"; + + 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); + + testing.clearData(); + + 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 recordsMetricsWhenExecutorNamesCollide() 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 { + 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(); + assertThat(first.awaitTermination(10, SECONDS)).isTrue(); + assertThat(second.awaitTermination(10, SECONDS)).isTrue(); + } + } + + private static void assertNoExecutorMetrics(String executorName, String ownerName) { + testing.clearData(); + 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 { + 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 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/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/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..fa345f333a11 --- /dev/null +++ b/instrumentation/executors/jdk21-testing/src/test/java/io/opentelemetry/javaagent/instrumentation/executors/ThreadPerTaskExecutorMetricsTest.java @@ -0,0 +1,179 @@ +/* + * 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.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 DEFAULT_OWNER_NAME = "unknown"; + private static final String EXECUTOR_NAME = "thread-per-task-*"; + private static final String EXECUTOR_TYPE = "java.util.concurrent.ThreadPerTaskExecutor"; + + @RegisterExtension + static final InstrumentationExtension testing = AgentInstrumentationExtension.create(); + + @Test + void recordsActiveThreadCountAndUnregistersOnShutdown() throws Exception { + 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( + () -> { + started.countDown(); + try { + release.await(10, SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + }); + + assertThat(started.await(10, SECONDS)).isTrue(); + 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); + 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(); + } + + assertNoThreadPerTaskMetrics(EXECUTOR_NAME, DEFAULT_OWNER_NAME); + } + + @Test + void unregistersOnClose() throws Exception { + ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor(); + + try { + 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(); + } + + assertNoThreadPerTaskMetrics(DEFAULT_OWNER_NAME, DEFAULT_OWNER_NAME); + } + + @Test + void doesNotRecordMetricsWhenUnclosedExecutorIsCollected() throws Exception { + WeakReference executorRef = + new WeakReference<>( + Executors.newThreadPerTaskExecutor( + new NamedThreadFactory("collected-thread-per-task"))); + + GcUtils.awaitGc(executorRef, Duration.ofSeconds(10)); + + assertNoThreadPerTaskMetrics("collected-thread-per-task-*", DEFAULT_OWNER_NAME); + } + + @Test + void reregistersOwnerAndThreadNameNormalization() throws Exception { + NamedThreadFactory threadFactory = new NamedThreadFactory("thread-per-task-42-worker"); + ExecutorService executor = Executors.newThreadPerTaskExecutor(threadFactory); + + try { + assertThat(threadFactory.createdThreadCount()).isZero(); + assertNoThreadPerTaskMetrics("thread-per-task-*-worker-*", DEFAULT_OWNER_NAME); + + ThreadPoolExecutorMetrics.reregister(executor, "tomcat", "trailing"); + executor.submit(() -> {}).get(10, SECONDS); + testing.clearData(); + + 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 assertNoThreadPerTaskMetrics(String executorName, String ownerName) { + testing.clearData(); + 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 { + 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..7b49db92cdca 100644 --- a/instrumentation/executors/metadata.yaml +++ b/instrumentation/executors/metadata.yaml @@ -22,3 +22,18 @@ 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.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 + - name: otel.instrumentation.executors-metrics.enabled + declarative_name: java.executors_metrics.enabled + description: Enables executor metrics instrumentation. + type: boolean + default: false 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)); + } +}