Drop APM traces when APM tracing is disabled and keep LLM Observability working

dd-java-agent/agent-llmobs/src/main/java/datadog/trace/llmobs/LLMObsSystem.java

@@ -42,6 +42,14 @@ public static void start(Instrumentation inst, SharedCommunicationObjects sco) {
       return;
     }
 
+    // LLMObs spans need a real tracer (buildSpan() returns null on the no-op one), so keep the
+    // no-op SDK when none is installed. Mirrors TracerInstaller.installGlobalTracer, which still
+    // installs one for CI Visibility.
+    if (!config.isTraceEnabled() && !config.isCiVisibilityEnabled()) {
+      LOGGER.debug("LLM Observability is disabled: no tracer is installed (dd.trace.enabled)");
+      return;
+    }
+
     sco.createRemaining(config);
 
     String mlApp = config.getLlmObsMlApp();

dd-java-agent/agent-llmobs/src/test/java/datadog/trace/llmobs/LLMObsSystemTest.java

@@ -0,0 +1,48 @@
+package datadog.trace.llmobs;
+
+import static datadog.trace.api.config.CiVisibilityConfig.CIVISIBILITY_ENABLED;
+import static datadog.trace.api.config.LlmObsConfig.LLMOBS_ENABLED;
+import static datadog.trace.api.config.TraceInstrumentationConfig.TRACE_ENABLED;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import datadog.trace.api.llmobs.LLMObs;
+import datadog.trace.api.llmobs.noop.NoOpLLMObsSpan;
+import datadog.trace.test.junit.utils.config.WithConfig;
+import datadog.trace.test.util.DDJavaSpecification;
+import org.junit.jupiter.api.Test;
+
+class LLMObsSystemTest extends DDJavaSpecification {
+
+  /**
+   * LLM Observability spans are backed by tracer spans, so the subsystem must stay off when tracing
+   * is disabled. Starting it would install the real SDK implementation on top of the no-op tracer,
+   * whose {@code buildSpan()} returns null, and every span-starting call would throw.
+   */
+  @WithConfig(key = LLMOBS_ENABLED, value = "true")
+  @WithConfig(key = TRACE_ENABLED, value = "false")
+  @Test
+  void staysNoOpWhenTracingIsDisabled() {
+    // The null SharedCommunicationObjects doubles as an assertion that start() returns before
+    // touching it: any use would throw.
+    LLMObsSystem.start(null, null);
+
+    assertSame(
+        NoOpLLMObsSpan.INSTANCE, LLMObs.startLLMSpan("span", "model", "provider", null, null));
+  }
+
+  /**
+   * CI Visibility installs a {@code CoreTracer} even with tracing disabled (see {@code
+   * TracerInstaller#installGlobalTracer}), so the guard above must not fire in that combination —
+   * the tracer LLM Observability needs does exist.
+   */
+  @WithConfig(key = LLMOBS_ENABLED, value = "true")
+  @WithConfig(key = TRACE_ENABLED, value = "false")
+  @WithConfig(key = CIVISIBILITY_ENABLED, value = "true")
+  @Test
+  void startsWhenCiVisibilityInstallsTheTracer() {
+    // Reaching the null SharedCommunicationObjects is the assertion: start() only dereferences it
+    // after the guard, so the NPE proves the subsystem was not short-circuited.
+    assertThrows(NullPointerException.class, () -> LLMObsSystem.start(null, null));
+  }
+}

dd-trace-core/src/main/java/datadog/trace/common/sampling/ApmTracingDisabledSampler.java

@@ -0,0 +1,43 @@
+package datadog.trace.common.sampling;
+
+import datadog.trace.api.sampling.PrioritySampling;
+import datadog.trace.api.sampling.SamplingMechanism;
+import datadog.trace.bootstrap.ActiveSubsystems;
+import datadog.trace.core.CoreSpan;
+import java.time.Clock;
+
+/**
+ * Drops APM traces for {@code dd.apm.tracing.enabled=false}. If Remote Configuration activates
+ * AppSec at runtime, falls back to the standalone ASM trickle of 1 APM trace per minute, so it
+ * checks {@link ActiveSubsystems#APPSEC_ACTIVE} per trace.
+ */
+public class ApmTracingDisabledSampler implements Sampler, PrioritySampler {
+
+  private final AsmStandaloneSampler asmStandaloneSampler;
+  private final ForcePrioritySampler dropSampler;
+
+  public ApmTracingDisabledSampler(final Clock clock) {
+    this.asmStandaloneSampler = new AsmStandaloneSampler(clock);
+    this.dropSampler =
+        new ForcePrioritySampler(PrioritySampling.SAMPLER_DROP, SamplingMechanism.DEFAULT);
+  }
+
+  @Override
+  public <T extends CoreSpan<T>> boolean sample(final T span) {
+    // Both delegates keep sending the trace to the agent so it can collect stats on dropped traces.
+    return isAsmActive() ? asmStandaloneSampler.sample(span) : dropSampler.sample(span);
+  }
+
+  @Override
+  public <T extends CoreSpan<T>> void setSamplingPriority(final T span) {
+    if (isAsmActive()) {
+      asmStandaloneSampler.setSamplingPriority(span);
+    } else {
+      dropSampler.setSamplingPriority(span);
+    }
+  }
+
+  private static boolean isAsmActive() {
+    return ActiveSubsystems.APPSEC_ACTIVE;
+  }
+}

dd-trace-core/src/main/java/datadog/trace/common/sampling/AsmStandaloneSampler.java

@@ -51,8 +51,12 @@ public <T extends CoreSpan<T>> void setSamplingPriority(final T span) {
 
   private boolean shouldSample() {
     long now = clock.millis();
-    return lastSampleTime.updateAndGet(
-            lastTime -> now - lastTime >= RATE_IN_MILLISECONDS ? now : lastTime)
-        == now;
+    // Compare against the time before the update, not after: two traces landing in the same
+    // millisecond as the last sample would both see the updated value equal to now and both be
+    // kept, sending 2 APM traces a minute instead of 1.
+    long lastSample =
+        lastSampleTime.getAndUpdate(
+            lastTime -> now - lastTime >= RATE_IN_MILLISECONDS ? now : lastTime);
+    return now - lastSample >= RATE_IN_MILLISECONDS;
   }
 }

dd-trace-core/src/main/java/datadog/trace/common/sampling/Sampler.java

@@ -57,9 +57,17 @@ public static Sampler forConfig(
         @Nullable final RateByServiceTraceSampler agentSampler) {
       Sampler sampler;
       if (config != null) {
-        if (!config.isApmTracingEnabled() && isAsmEnabled(config)) {
-          log.debug("APM is disabled. Only 1 trace per minute will be sent.");
-          return new AsmStandaloneSampler(Clock.systemUTC());
+        if (!config.isApmTracingEnabled()) {
+          if (isAsmEnabled(config)) {
+            log.debug(
+                "APM tracing is disabled, but ASM is enabled. Only 1 APM trace per minute will be sent.");
+            return new AsmStandaloneSampler(Clock.systemUTC());
+          }
+          // Only ASM needs a trickle of APM traces, so drop them all unless AppSec is activated
+          // later by Remote Configuration.
+          log.debug(
+              "APM tracing is disabled. APM traces will be dropped, unless AppSec is activated at runtime.");
+          return new ApmTracingDisabledSampler(Clock.systemUTC());
         }
         final Map<String, String> serviceRules = config.getTraceSamplingServiceRules();
         final Map<String, String> operationRules = config.getTraceSamplingOperationRules();
@@ -133,6 +141,7 @@ public static Sampler forConfig(
       return sampler;
     }
 
+    /** Whether any Application Security product is on at startup. */
     private static boolean isAsmEnabled(Config config) {
       return config.getAppSecActivation() == ProductActivation.FULLY_ENABLED
           || config.getIastActivation() == ProductActivation.FULLY_ENABLED

dd-trace-core/src/main/java/datadog/trace/common/writer/TraceProcessingWorker.java

@@ -128,6 +128,11 @@ MessagePassingBlockingQueue<Object> getPrimaryQueue() {
     return primaryQueue;
   }
 
+  @VisibleForTesting
+  MessagePassingBlockingQueue<Object> getSecondaryQueue() {
+    return secondaryQueue;
+  }
+
   private static MessagePassingBlockingQueue<Object> createQueue(int capacity) {
     return Queues.mpscBlockingConsumerArrayQueue(capacity);
   }
@@ -195,6 +200,9 @@ public void onEvent(Object event) {
           // TODO populate `_sample_rate` metric in a way that accounts for lost/dropped traces
           payloadDispatcher.addTrace(trace);
         } else if (event instanceof FlushEvent) {
+          // The flush marker only goes to the primary queue, so drain the secondary queue too:
+          // sampled-out traces can carry LLM Observability spans that must still be sent.
+          consumeFromSecondaryQueue();
           payloadDispatcher.flush();
           ((FlushEvent) event).sync();
         }

dd-trace-core/src/main/java/datadog/trace/core/propagation/ContextInterpreter.java

@@ -19,6 +19,7 @@
 import datadog.trace.api.DDSpanId;
 import datadog.trace.api.DDTraceId;
 import datadog.trace.api.Functions;
+import datadog.trace.api.ProductTraceSource;
 import datadog.trace.api.TagMap;
 import datadog.trace.api.TraceConfig;
 import datadog.trace.api.TracePropagationStyle;
@@ -72,6 +73,7 @@ public abstract class ContextInterpreter implements AgentPropagation.KeyClassifi
   private final boolean requestHeaderTagsCommaAllowed;
   private final int baggageMaxItems;
   private final int baggageMaxBytes;
+  private final boolean apmTracingEnabled;
 
   protected static final boolean LOG_EXTRACT_HEADER_NAMES = Config.get().isLogExtractHeaderNames();
   private static final DDCache<String, String> CACHE = DDCaches.newFixedSizeCache(64);
@@ -89,6 +91,7 @@ protected ContextInterpreter(Config config) {
     this.requestHeaderTagsCommaAllowed = config.isRequestHeaderTagsCommaAllowed();
     this.baggageMaxItems = config.getTraceBaggageMaxItems();
     this.baggageMaxBytes = config.getTraceBaggageMaxBytes();
+    this.apmTracingEnabled = config.isApmTracingEnabled();
   }
 
   final TagMap.Ledger tagLedger() {
@@ -345,6 +348,17 @@ private TagContext.HttpHeaders getHeaders() {
   }
 
   private int samplingPriorityOrDefault(DDTraceId traceId, int samplingPriority) {
+    if (!apmTracingEnabled
+        && (propagationTags == null
+            || !ProductTraceSource.isProductMarked(
+                propagationTags.getTraceSource(),
+                ProductTraceSource.ASM,
+                ProductTraceSource.AI_GUARD))) {
+      // With APM tracing disabled, don't inherit an upstream APM decision: let the local sampler
+      // set the priority. Same products as TraceCollector.setSamplingPriorityIfNecessary. Needed
+      // when Data Jobs is enabled, where the drop can't override an already-set priority.
+      return PrioritySampling.UNSET;
+    }
     return samplingPriority == PrioritySampling.UNSET || DDTraceId.ZERO.equals(traceId)
         ? defaultSamplingPriority()
         : samplingPriority;