Tag consumer spans with the pathway hash on every data streams checkpoint (#11808)

Tag consumer spans with the pathway hash on every data streams checkpoint

The pathway.hash span tag was only set on the produce/inject path
(DataStreamsPropagator.inject). Consumers that checkpoint without injecting
(e.g. RabbitMQ) had no pathway.hash, unlike the JS and Python tracers which tag
the span on every checkpoint. Set it centrally in
DefaultDataStreamsMonitoring.setCheckpoint so all consume-side integrations get
it.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Remove explanatory comment on pathway hash tagging

Per review on #11808; the rationale lives in the PR description.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Address review on #11808

- cache pathwayContext.getHash() in a local instead of calling it twice
- use a negative hash in the test so it actually exercises Long.toUnsignedString

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Expect pathway.hash on the inferred-proxy server span when DSM is enabled

Now that DefaultDataStreamsMonitoring.setCheckpoint tags the span on every
checkpoint, the inferred-proxy (API Gateway) HTTP server span gets a
pathway.hash like any other DSM-enabled server span. The shared HttpServerTest
already asserts this conditionally; SpringBootBasedTest's hand-rolled
inferred-proxy assertions were missing it. Mirror the same guard.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Merge branch 'master' into eric.firth/dsm-consumer-pathway-hash

Co-authored-by: eric.firth <eric.firth@datadoghq.com>

Author: ericfirth

dd-java-agent/instrumentation/spring/spring-webmvc/spring-webmvc-3.1/src/test/groovy/test/boot/SpringBootBasedTest.groovy

@@ -570,6 +570,9 @@ class SpringBootBasedTest extends HttpServerTest<ConfigurableApplicationContext>
             "$Tags.HTTP_ROUTE" "/success"
             "servlet.context" "/boot-context"
             "servlet.path" "/success"
+            if ({ isDataStreamsEnabled() }) {
+              "$DDTags.PATHWAY_HASH" { String }
+            }
             defaultTags()
           }
         }

dd-trace-core/src/main/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoring.java

@@ -1,6 +1,7 @@
 package datadog.trace.core.datastreams;
 
 import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V01_DATASTREAMS_ENDPOINT;
+import static datadog.trace.api.DDTags.PATHWAY_HASH;
 import static datadog.trace.api.datastreams.DataStreamsContext.fromTags;
 import static datadog.trace.api.datastreams.DataStreamsTags.Direction.INBOUND;
 import static datadog.trace.api.datastreams.DataStreamsTags.Direction.OUTBOUND;
@@ -309,6 +310,10 @@ public void setCheckpoint(AgentSpan span, DataStreamsContext context) {
     PathwayContext pathwayContext = span.spanContext().getPathwayContext();
     if (pathwayContext != null) {
       pathwayContext.setCheckpoint(context, this::add);
+      long pathwayHash = pathwayContext.getHash();
+      if (pathwayHash != 0) {
+        span.setTag(PATHWAY_HASH, Long.toUnsignedString(pathwayHash));
+      }
     }
   }
 

dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTest.java

@@ -10,22 +10,29 @@
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.RETURNS_SMART_NULLS;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
 import datadog.metrics.api.Histograms;
 import datadog.metrics.impl.DDSketchHistograms;
 import datadog.trace.api.Config;
+import datadog.trace.api.DDTags;
 import datadog.trace.api.TraceConfig;
+import datadog.trace.api.datastreams.DataStreamsContext;
 import datadog.trace.api.datastreams.DataStreamsTags;
 import datadog.trace.api.datastreams.KafkaConfigReport;
+import datadog.trace.api.datastreams.PathwayContext;
 import datadog.trace.api.datastreams.SchemaRegistryUsage;
 import datadog.trace.api.datastreams.StatsPoint;
 import datadog.trace.api.experimental.DataStreamsContextCarrier;
 import datadog.trace.api.time.ControllableTimeSource;
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import datadog.trace.bootstrap.instrumentation.api.AgentSpanContext;
 import datadog.trace.common.metrics.EventListener;
 import datadog.trace.common.metrics.Sink;
 import datadog.trace.core.DDCoreJavaSpecification;
@@ -1559,6 +1566,67 @@ void close() {
     }
   }
 
+  private DefaultDataStreamsMonitoring newDataStreamsMonitoring() {
+    return new DefaultDataStreamsMonitoring(
+        mock(Sink.class),
+        stubFeatures(true),
+        new ControllableTimeSource(),
+        () -> stubTraceConfig(true),
+        mock(DatastreamsPayloadWriter.class),
+        DEFAULT_BUCKET_DURATION_NANOS);
+  }
+
+  @Test
+  void setCheckpointTagsTheSpanWithThePathwayHash() {
+    DefaultDataStreamsMonitoring dataStreams = newDataStreamsMonitoring();
+
+    // Negative so the signed and unsigned string representations differ, proving the tag uses
+    // Long.toUnsignedString (pathway hashes are unsigned 64-bit values).
+    long hash = -1234567890123456789L;
+    PathwayContext pathwayContext = mock(PathwayContext.class);
+    when(pathwayContext.getHash()).thenReturn(hash);
+    AgentSpanContext spanContext = mock(AgentSpanContext.class);
+    when(spanContext.getPathwayContext()).thenReturn(pathwayContext);
+    AgentSpan span = mock(AgentSpan.class);
+    when(span.spanContext()).thenReturn(spanContext);
+    DataStreamsContext context = mock(DataStreamsContext.class);
+
+    dataStreams.setCheckpoint(span, context);
+
+    verify(pathwayContext).setCheckpoint(eq(context), any());
+    verify(span).setTag(DDTags.PATHWAY_HASH, Long.toUnsignedString(hash));
+  }
+
+  @Test
+  void setCheckpointDoesNotTagTheSpanWhenThePathwayHashIsZero() {
+    DefaultDataStreamsMonitoring dataStreams = newDataStreamsMonitoring();
+
+    PathwayContext pathwayContext = mock(PathwayContext.class);
+    when(pathwayContext.getHash()).thenReturn(0L);
+    AgentSpanContext spanContext = mock(AgentSpanContext.class);
+    when(spanContext.getPathwayContext()).thenReturn(pathwayContext);
+    AgentSpan span = mock(AgentSpan.class);
+    when(span.spanContext()).thenReturn(spanContext);
+
+    dataStreams.setCheckpoint(span, mock(DataStreamsContext.class));
+
+    verify(span, never()).setTag(eq(DDTags.PATHWAY_HASH), any(String.class));
+  }
+
+  @Test
+  void setCheckpointDoesNotTagTheSpanWhenThereIsNoPathwayContext() {
+    DefaultDataStreamsMonitoring dataStreams = newDataStreamsMonitoring();
+
+    AgentSpanContext spanContext = mock(AgentSpanContext.class);
+    when(spanContext.getPathwayContext()).thenReturn(null);
+    AgentSpan span = mock(AgentSpan.class);
+    when(span.spanContext()).thenReturn(spanContext);
+
+    dataStreams.setCheckpoint(span, mock(DataStreamsContext.class));
+
+    verify(span, never()).setTag(eq(DDTags.PATHWAY_HASH), any(String.class));
+  }
+
   static class CustomContextCarrier implements DataStreamsContextCarrier {
     private final Map<String, Object> data = new HashMap<>();