🍒 12028 - Stamp _dd.apm.enabled on every exported chunk when APM tracing is disabled

dd-smoke-tests/apm-tracing-disabled/application/src/main/java/datadog/smoketest/apmtracingdisabled/Controller.java

@@ -1,10 +1,13 @@
 package datadog.smoketest.apmtracingdisabled;
 
+import io.opentracing.Scope;
 import io.opentracing.Span;
 import io.opentracing.util.GlobalTracer;
 import java.io.IOException;
 import java.util.Map;
 import javax.servlet.http.HttpServletResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
@@ -14,6 +17,8 @@
 @RequestMapping("/rest-api")
 public class Controller {
 
+  private static final Logger log = LoggerFactory.getLogger(Controller.class);
+
   @GetMapping("/greetings")
   public String greetings(
       @RequestParam(name = "url", required = false) String url,
@@ -71,6 +76,33 @@ public void write(
     }
   }
 
+  @GetMapping("/late-outbound")
+  public String lateOutbound(@RequestParam(name = "url") String url) {
+    final Span span = GlobalTracer.get().activeSpan();
+    // Thread synchronization relies on waitForTraceCount rather than Thread completion, no race
+    // issue.
+    Thread thread =
+        new Thread(
+            () -> {
+              try {
+                // Sleep past PendingTraceBuffer's 500ms flush delay so the root chunk exports
+                // before this late child.
+                Thread.sleep(3000);
+              } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return;
+              }
+              try (Scope scope = GlobalTracer.get().activateSpan(span)) {
+                new RestTemplate().getForObject(url, String.class);
+              } catch (Exception e) {
+                log.debug("late outbound call to {} failed", url, e);
+              }
+            });
+    thread.setDaemon(true);
+    thread.start();
+    return "late-outbound";
+  }
+
   private String forceKeepSpan() {
     final Span span = GlobalTracer.get().activeSpan();
     if (span != null) {

dd-smoke-tests/apm-tracing-disabled/src/test/groovy/datadog/smoketest/apmtracingdisabled/AbstractApmTracingDisabledSmokeTest.groovy

@@ -78,7 +78,20 @@ abstract class AbstractApmTracingDisabledSmokeTest extends AbstractServerSmokeTe
   }
 
   protected hasApmDisabledTag(DecodedTrace trace) {
-    return trace.spans[0].metrics['_dd.apm.enabled'] == 0
+    return trace.spans.any { it.metrics['_dd.apm.enabled'] == 0 }
+  }
+
+  protected hasApmDisabledTagOnEverySpan(DecodedTrace trace) {
+    return trace.spans.every { it.metrics['_dd.apm.enabled'] == 0 }
+  }
+
+  /** The chunk holding the delayed outbound client span (service == serviceName, http.url == url). */
+  protected DecodedTrace getOutboundChunk(String serviceName, String url) {
+    return traces.find { trace ->
+      trace.spans.any { span ->
+        span.service == serviceName && span.meta['http.url'] == url
+      }
+    }
   }
 
   protected hasASMEvents(DecodedTrace trace){

dd-smoke-tests/apm-tracing-disabled/src/test/groovy/datadog/smoketest/apmtracingdisabled/ApmTracingDisabledSmokeTest.groovy

@@ -50,6 +50,28 @@ abstract class ApmTracingDisabledSmokeTest extends AbstractApmTracingDisabledSmo
     !hasApmDisabledTag(getServiceTrace(APM_ENABLED_SERVICE_NAME))
   }
 
+  void 'When APM is disabled, a delayed outbound child flushed in its own chunk must still carry _dd.apm.enabled:0'() {
+    setup:
+    final targetUrl = "http://localhost:${httpPorts[1]}/rest-api/greetings"
+    final url = "http://localhost:${httpPorts[0]}/rest-api/late-outbound?url=${URLEncoder.encode(targetUrl, 'UTF-8')}"
+    final request = new Request.Builder().url(url).get().build()
+
+    when:
+    final response = client.newCall(request).execute()
+
+    then: 'the request root, the delayed outbound child, and the downstream service each report a chunk'
+    response.successful
+    waitForTraceCount(3)
+    // The delayed outbound client span belongs to the APM-disabled service but is exported alone,
+    // in a separate chunk from the /late-outbound service-entry span that carries the marker.
+    def outboundChunk = getOutboundChunk(APM_TRACING_DISABLED_SERVICE_NAME, targetUrl)
+    assert outboundChunk != null
+    assert outboundChunk.spans.every { it.resource != 'GET /rest-api/late-outbound' }
+
+    then: 'every span of the delayed child chunk must carry the billing marker'
+    hasApmDisabledTagOnEverySpan(outboundChunk)
+  }
+
   void 'When APM is disabled, libraries must completely disable the generation of APM trace metrics'(){
     setup:
     final url1 = "http://localhost:${httpPorts[0]}/rest-api/greetings"

dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java

@@ -1,5 +1,6 @@
 package datadog.trace.core;
 
+import static datadog.trace.api.DDTags.APM_ENABLED;
 import static datadog.trace.api.DDTags.DJM_ENABLED;
 import static datadog.trace.api.DDTags.DSM_ENABLED;
 import static datadog.trace.api.DDTags.PROFILING_CONTEXT_ENGINE;
@@ -191,6 +192,12 @@ public static CoreTracerBuilder builder() {
 
   private final boolean localRootSpanTagsNeedIntercept;
 
+  /**
+   * When {@code false}, every exported span is stamped with the {@code _dd.apm.enabled:0} billing
+   * marker — see {@link #write(SpanList)}.
+   */
+  private final boolean apmTracingEnabled;
+
   /** A set of tags that are added to every span */
   private final TagMap defaultSpanTags;
 
@@ -665,6 +672,7 @@ private CoreTracer(
     this.serviceName = serviceName;
 
     this.initialConfig = config;
+    this.apmTracingEnabled = config.isApmTracingEnabled();
     this.initialSampler = sampler;
 
     // Get initial Trace Sampling Rules from config
@@ -1297,6 +1305,13 @@ void write(final SpanList trace) {
     spanToSample.forceKeep(forceKeep);
     boolean published = forceKeep || traceCollector.sample(spanToSample);
     if (published) {
+      if (!apmTracingEnabled) {
+        // Stamp the billing marker on every span of each exported chunk so the intake does not bill
+        // APM host usage.
+        for (int i = 0; i < writtenTrace.size(); i++) {
+          writtenTrace.get(i).setTag(APM_ENABLED, 0);
+        }
+      }
       writer.write(writtenTrace);
     } else {
       // with span streaming this won't work - it needs to be changed

dd-trace-core/src/test/java/datadog/trace/core/ApmTracingDisabledChunkMarkerTest.java

@@ -0,0 +1,175 @@
+package datadog.trace.core;
+
+import static datadog.trace.api.DDTags.APM_ENABLED;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import datadog.trace.common.writer.ListWriter;
+import datadog.trace.junit.utils.config.WithConfig;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * When APM tracing is disabled ({@code DD_APM_TRACING_ENABLED=false}), the intake bills APM host
+ * usage for every exported trace <em>chunk</em> that does not carry the {@code _dd.apm.enabled:0}
+ * marker. To make sure no chunk gets flushed without the correct billing tag, it gets added to
+ * every single span. These tests lock that in.
+ */
+@WithConfig(key = "apm.tracing.enabled", value = "false")
+class ApmTracingDisabledChunkMarkerTest extends DDCoreJavaSpecification {
+
+  private ListWriter writer;
+  private CoreTracer tracer;
+
+  @Override
+  protected boolean useStrictTraceWrites() {
+    // Production standalone-ASM uses the delaying PendingTraceBuffer, which is what lets a buffered
+    // root be written on its own before a late child arrives. Strict writes would instead select
+    // the discarding buffer and coalesce both spans into a single chunk, hiding the bug.
+    return false;
+  }
+
+  @BeforeEach
+  void setup() {
+    writer = new ListWriter();
+    tracer = tracerBuilder().writer(writer).build();
+  }
+
+  @AfterEach
+  void cleanup() {
+    if (tracer != null) {
+      tracer.close();
+    }
+  }
+
+  @Test
+  void everyExportedChunkCarriesApmDisabledMarker() throws InterruptedException, TimeoutException {
+    DDSpan root = (DDSpan) tracer.buildSpan("test", "root").start();
+    DDSpan child = (DDSpan) tracer.buildSpan("test", "child").asChildOf(root.spanContext()).start();
+
+    PendingTrace trace = (PendingTrace) root.spanContext().getTraceCollector();
+
+    // Root finishes while the child is still running -> root is buffered, not yet written.
+    root.finish();
+    assertTrue(writer.isEmpty());
+
+    // Write the buffered root out on its own chunk. This sets rootSpanWritten=true and is exactly
+    // what the PendingTraceBuffer worker does when a buffered root ages out (SEND_DELAY_NS).
+    trace.write();
+    writer.waitForTraces(1);
+
+    // The long-lived child finishes later. Because the root was already written, it flushes alone.
+    child.finish();
+    trace.write();
+    writer.waitForTraces(2);
+
+    List<DDSpan> rootChunk = writer.get(0);
+    List<DDSpan> childChunk = writer.get(1);
+    assertEquals(1, rootChunk.size(), "expected the root to flush alone in the first chunk");
+    assertEquals(1, childChunk.size(), "expected the child to flush alone in the second chunk");
+    assertEquals(root, rootChunk.get(0), "first chunk should be the local root");
+    assertEquals(child, childChunk.get(0), "second chunk should be the delayed child");
+
+    // Sanity: the child really is a child of the local root (a local, non-remote parent) — this is
+    // exactly the case that stamping the marker only on the local root span would miss.
+    assertEquals(root.getSpanId(), child.getParentId());
+
+    // Both chunks must carry the billing marker. The delayed child chunk is load-bearing: without
+    // the marker the intake would bill APM host usage for this otherwise-unmarked chunk.
+    assertAllSpansMarked(rootChunk, "root chunk");
+    assertAllSpansMarked(childChunk, "delayed child chunk");
+  }
+
+  @Test
+  void singleChunkTraceCarriesApmDisabledMarker() throws InterruptedException, TimeoutException {
+    // Positive control: when the whole trace is exported as a single chunk (child finishes before
+    // the root, so nothing is written until the root closes), the marker is present on every span.
+    DDSpan root = (DDSpan) tracer.buildSpan("test", "root").start();
+    DDSpan child = (DDSpan) tracer.buildSpan("test", "child").asChildOf(root.spanContext()).start();
+
+    child.finish();
+    assertTrue(writer.isEmpty(), "trace must not be written while the root is still open");
+    root.finish();
+    writer.waitForTraces(1);
+
+    assertEquals(1, writer.size(), "expected the whole trace in a single chunk");
+    List<DDSpan> chunk = writer.get(0);
+    assertEquals(2, chunk.size(), "expected both spans in the same chunk");
+    assertAllSpansMarked(chunk, "single-chunk trace");
+  }
+
+  @Test
+  void rootlessMultiSpanChunkMarksEverySpan() throws InterruptedException, TimeoutException {
+    // A multi-span chunk exported WITHOUT its local root (a partial flush / orphaned subtree): the
+    // root is written first, then its descendants flush together in a later, root-less chunk. Every
+    // span in that chunk must still carry the marker.
+    DDSpan root = (DDSpan) tracer.buildSpan("test", "root").start();
+    DDSpan childA =
+        (DDSpan) tracer.buildSpan("test", "childA").asChildOf(root.spanContext()).start();
+    DDSpan grandchild =
+        (DDSpan) tracer.buildSpan("test", "grandchild").asChildOf(childA.spanContext()).start();
+
+    PendingTrace trace = (PendingTrace) root.spanContext().getTraceCollector();
+
+    // Flush the root on its own first (rootSpanWritten=true), so the descendants can only export in
+    // a later, root-less chunk.
+    root.finish();
+    trace.write();
+    writer.waitForTraces(1);
+    assertEquals(1, writer.get(0).size(), "expected the root to flush alone in the first chunk");
+
+    // Both descendants finish, then flush together in a single root-less chunk.
+    childA.finish();
+    grandchild.finish();
+    trace.write();
+    writer.waitForTraces(2);
+
+    List<DDSpan> chunk = writer.get(1);
+    assertEquals(2, chunk.size(), "expected childA and grandchild in one root-less chunk");
+    assertTrue(chunk.contains(childA) && chunk.contains(grandchild), "chunk must hold both spans");
+    assertTrue(!chunk.contains(root), "the local root must not be part of this chunk");
+
+    // Every span in the root-less chunk carries the marker, including the inner grandchild whose
+    // parent is present in the chunk.
+    assertAllSpansMarked(chunk, "root-less descendant chunk");
+  }
+
+  @Test
+  @WithConfig(key = "apm.tracing.enabled", value = "true")
+  void apmTracingEnabledLeavesChunkUnmarked() throws InterruptedException, TimeoutException {
+    // Negative control: the method-level override flips APM tracing back on (overriding the
+    // class-level false before setup() builds the tracer), so the billing marker must NOT be
+    // stamped on any span. Guards against a regression that marks unconditionally, ignoring the
+    // apm.tracing.enabled flag.
+    DDSpan root = (DDSpan) tracer.buildSpan("test", "root").start();
+    DDSpan child = (DDSpan) tracer.buildSpan("test", "child").asChildOf(root.spanContext()).start();
+
+    child.finish();
+    root.finish();
+    writer.waitForTraces(1);
+
+    List<DDSpan> chunk = writer.get(0);
+    assertEquals(2, chunk.size(), "expected both spans in the same chunk");
+    for (DDSpan span : chunk) {
+      assertNull(
+          span.getTag(APM_ENABLED),
+          "span '"
+              + span.getOperationName()
+              + "' must not carry _dd.apm.enabled when APM tracing is enabled");
+    }
+  }
+
+  /** Every span of an exported chunk must carry the numeric {@code _dd.apm.enabled:0} marker. */
+  private static void assertAllSpansMarked(List<DDSpan> chunk, String description) {
+    for (DDSpan span : chunk) {
+      assertEquals(
+          Integer.valueOf(0),
+          span.getTag(APM_ENABLED),
+          description + " span '" + span.getOperationName() + "' is missing _dd.apm.enabled:0");
+    }
+  }
+}

internal-api/src/main/java/datadog/trace/api/Config.java

@@ -194,7 +194,6 @@
 import static datadog.trace.api.ConfigDefaults.DEFAULT_WEBSOCKET_MESSAGES_SEPARATE_TRACES;
 import static datadog.trace.api.ConfigDefaults.DEFAULT_WEBSOCKET_TAG_SESSION_ID;
 import static datadog.trace.api.ConfigSetting.NON_DEFAULT_SEQ_ID;
-import static datadog.trace.api.DDTags.APM_ENABLED;
 import static datadog.trace.api.DDTags.HOST_TAG;
 import static datadog.trace.api.DDTags.INTERNAL_HOST_NAME;
 import static datadog.trace.api.DDTags.LANGUAGE_TAG_KEY;
@@ -5169,9 +5168,10 @@ public TagMap getLocalRootSpanTags() {
     result.put(LANGUAGE_TAG_KEY, LANGUAGE_TAG_VALUE);
     result.put(SCHEMA_VERSION_TAG_KEY, SpanNaming.instance().version());
     result.put(DDTags.PROFILING_ENABLED, isProfilingEnabled() ? 1 : 0);
-    if (!isApmTracingEnabled()) {
-      result.put(APM_ENABLED, 0);
-    }
+    // The _dd.apm.enabled:0 billing marker is intentionally NOT set here. When APM tracing is
+    // disabled it is stamped on every span of each exported chunk (see CoreTracer.write), so that
+    // chunks flushed without their local root span (e.g. a late child span) still opt out of APM
+    // host billing.
 
     if (reportHostName) {
       final String hostName = getHostName();