feat(profiler): wire submittingSpanId into QueueTime events

Author: kaahos

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

@@ -453,8 +453,8 @@ public void recordSetting(String name, String value, String unit) {
     profiler.recordSetting(name, value, unit);
   }
 
-  public QueueTimeTracker newQueueTimeTracker() {
-    return new QueueTimeTracker(this, profiler.getCurrentTicks());
+  public QueueTimeTracker newQueueTimeTracker(long submittingSpanId) {
+    return new QueueTimeTracker(this, profiler.getCurrentTicks(), submittingSpanId);
   }
 
   boolean shouldRecordQueueTimeEvent(long startMillis) {
@@ -495,15 +495,23 @@ void recordQueueTimeEvent(
       Class<?> scheduler,
       Class<?> queueType,
       int queueLength,
-      Thread origin) {
+      Thread origin,
+      long submittingSpanId) {
     if (profiler != null) {
       // note: because this type traversal can update secondary_super_cache (see JDK-8180450)
       // we avoid doing this unless we are absolutely certain we will record the event
       Class<?> taskType = TaskWrapper.getUnwrappedType(task);
       if (taskType != null) {
         long endTicks = profiler.getCurrentTicks();
         profiler.recordQueueTime(
-            startTicks, endTicks, taskType, scheduler, queueType, queueLength, origin);
+            startTicks,
+            endTicks,
+            taskType,
+            scheduler,
+            queueType,
+            queueLength,
+            origin,
+            submittingSpanId);
       }
     }
   }

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

@@ -6,6 +6,7 @@
 import datadog.trace.api.profiling.ProfilingScope;
 import datadog.trace.api.profiling.Timing;
 import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
 import datadog.trace.bootstrap.instrumentation.api.ProfilerContext;
 import datadog.trace.bootstrap.instrumentation.api.ProfilingContextIntegration;
 
@@ -164,7 +165,10 @@ public EndpointTracker onRootSpanStarted(AgentSpan rootSpan) {
   @Override
   public Timing start(TimerType type) {
     if (IS_PROFILING_QUEUEING_TIME_ENABLED && type == TimerType.QUEUEING) {
-      return DDPROF.newQueueTimeTracker();
+      AgentSpan span = AgentTracer.activeSpan();
+      long submittingSpanId =
+          (span instanceof ProfilerContext) ? ((ProfilerContext) span).getSpanId() : 0L;
+      return DDPROF.newQueueTimeTracker(submittingSpanId);
     }
     return Timing.NoOp.INSTANCE;
   }

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

@@ -9,18 +9,20 @@ public class QueueTimeTracker implements QueueTiming {
   private final Thread origin;
   private final long startTicks;
   private final long startMillis;
+  private final long submittingSpanId;
   private WeakReference<Object> weakTask;
   // FIXME this can be eliminated by altering the instrumentation
   //  since it is known when the item is polled from the queue
   private Class<?> scheduler;
   private Class<?> queue;
   private int queueLength;
 
-  public QueueTimeTracker(DatadogProfiler profiler, long startTicks) {
+  public QueueTimeTracker(DatadogProfiler profiler, long startTicks, long submittingSpanId) {
     this.profiler = profiler;
     this.origin = Thread.currentThread();
     this.startTicks = startTicks;
     this.startMillis = System.currentTimeMillis();
+    this.submittingSpanId = submittingSpanId;
   }
 
   @Override
@@ -49,7 +51,8 @@ public void report() {
     Object task = this.weakTask.get();
     if (task != null) {
       // indirection reduces shallow size of the tracker instance
-      profiler.recordQueueTimeEvent(startTicks, task, scheduler, queue, queueLength, origin);
+      profiler.recordQueueTimeEvent(
+          startTicks, task, scheduler, queue, queueLength, origin, submittingSpanId);
     }
   }