feat(profiler): emit synthetic SpanNode events for ThreadPoolExecutor worker threads

Author: kaahos

dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/instrumentation/java/concurrent/TPEHelper.java

@@ -7,6 +7,9 @@
 import datadog.trace.api.InstrumenterConfig;
 import datadog.trace.bootstrap.ContextStore;
 import datadog.trace.bootstrap.instrumentation.api.AgentScope;
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
+import datadog.trace.bootstrap.instrumentation.api.ProfilerContext;
 import java.util.Set;
 import java.util.concurrent.ThreadPoolExecutor;
 
@@ -25,6 +28,8 @@ public final class TPEHelper {
   private static final Set<String> excludedClasses;
   // A ThreadLocal to store the Scope between beforeExecute and afterExecute if wrapping is not used
   private static final ThreadLocal<AgentScope> threadLocalScope;
+  // Stores System.nanoTime() at task activation so onTaskDeactivation can compute duration
+  private static final ThreadLocal<Long> threadLocalActivationNano;
 
   private static final ClassValue<Boolean> WRAP =
       GenericClassValue.of(
@@ -42,8 +47,10 @@ public final class TPEHelper {
     excludedClasses = config.getTraceThreadPoolExecutorsExclude();
     if (useWrapping) {
       threadLocalScope = null;
+      threadLocalActivationNano = null;
     } else {
       threadLocalScope = new ThreadLocal<>();
+      threadLocalActivationNano = new ThreadLocal<>();
     }
   }
 
@@ -82,7 +89,18 @@ public static AgentScope startScope(ContextStore<Runnable, State> contextStore,
     if (task == null || exclude(RUNNABLE, task)) {
       return null;
     }
-    return AdviceUtils.startTaskScope(contextStore, task);
+    AgentScope scope = AdviceUtils.startTaskScope(contextStore, task);
+    if (scope != null && threadLocalActivationNano != null) {
+      long startNano = System.nanoTime();
+      threadLocalActivationNano.set(startNano);
+      AgentSpan span = scope.span();
+      if (span != null && span.context() instanceof ProfilerContext) {
+        AgentTracer.get()
+            .getProfilingContext()
+            .onTaskActivation((ProfilerContext) span.context(), startNano);
+      }
+    }
+    return scope;
   }
 
   public static void setThreadLocalScope(AgentScope scope, Runnable task) {
@@ -112,6 +130,19 @@ public static void endScope(AgentScope scope, Runnable task) {
     if (task == null || exclude(RUNNABLE, task)) {
       return;
     }
+    if (scope != null && threadLocalActivationNano != null) {
+      Long startNano = threadLocalActivationNano.get();
+      // noinspection ThreadLocalSetWithNull
+      threadLocalActivationNano.set(null);
+      if (startNano != null) {
+        AgentSpan span = scope.span();
+        if (span != null && span.context() instanceof ProfilerContext) {
+          AgentTracer.get()
+              .getProfilingContext()
+              .onTaskDeactivation((ProfilerContext) span.context(), startNano);
+        }
+      }
+    }
     AdviceUtils.endTaskScope(scope);
   }
 

dd-java-agent/agent-profiling/profiling-ddprof/src/main/java/com/datadog/profiling/ddprof/DatadogProfilingIntegration.java

@@ -119,6 +119,40 @@ public void onSpanFinished(AgentSpan span) {
         ctx.getEncodedResourceName());
   }
 
+  // Calibration offset so System.nanoTime() values can be expressed as epoch nanoseconds.
+  // Computed once at class load to keep activation/deactivation overhead minimal.
+  private static final long EPOCH_NANOS_OFFSET =
+      System.currentTimeMillis() * 1_000_000L - System.nanoTime();
+
+  @Override
+  public void onTaskActivation(ProfilerContext profilerContext, long startTicks) {
+    // startTicks captured by TPEHelper is the authoritative start; nothing to do here.
+  }
+
+  @Override
+  public void onTaskDeactivation(ProfilerContext profilerContext, long startTicks) {
+    if (profilerContext == null) {
+      return;
+    }
+    long endNano = System.nanoTime();
+    long startNano = startTicks; // startTicks carries nanoTime at activation (see TPEHelper)
+    long durationNanos = endNano - startNano;
+    if (durationNanos <= 0) {
+      return;
+    }
+    long startNanos = startNano + EPOCH_NANOS_OFFSET;
+    long syntheticSpanId =
+        profilerContext.getSpanId() ^ ((long) Thread.currentThread().getId() << 32) ^ startNano;
+    DDPROF.recordSpanNodeEvent(
+        syntheticSpanId,
+        profilerContext.getSpanId(),
+        profilerContext.getRootSpanId(),
+        startNanos,
+        durationNanos,
+        profilerContext.getEncodedOperationName(),
+        profilerContext.getEncodedResourceName());
+  }
+
   public void clearContext() {
     DDPROF.clearSpanContext();
     DDPROF.clearContextValue(SPAN_NAME_INDEX);

internal-api/src/main/java/datadog/trace/bootstrap/instrumentation/api/ProfilingContextIntegration.java

@@ -58,6 +58,24 @@ default void recordTaskBlock(
    */
   default void onSpanFinished(AgentSpan span) {}
 
+  /**
+   * Called when a span context continuation activates on a worker thread (task execution start).
+   * Implementations record the start tick so a synthetic SpanNode can be emitted at deactivation.
+   *
+   * @param profilerContext the activated span context
+   * @param startTicks TSC tick at activation (from {@link #getCurrentTicks()})
+   */
+  default void onTaskActivation(ProfilerContext profilerContext, long startTicks) {}
+
+  /**
+   * Called when a span context continuation deactivates on a worker thread (task execution end).
+   * Implementations emit a lightweight synthetic SpanNode covering the worker thread interval.
+   *
+   * @param profilerContext the deactivated span context
+   * @param startTicks TSC tick recorded at {@link #onTaskActivation}
+   */
+  default void onTaskDeactivation(ProfilerContext profilerContext, long startTicks) {}
+
   String name();
 
   final class NoOp implements ProfilingContextIntegration {