Decouple Kafka DSM instrumentation from APM tracing enablement - #12350
Decouple Kafka DSM instrumentation from APM tracing enablement#12350ericfirth wants to merge 24 commits into
Conversation
Implements the pre-instrumentation config accessor for the data streams feature flag, mirroring the existing usmEnabled pattern. Reads the same data.streams.enabled key that Config.isDataStreamsEnabled() uses, enabling DSM to work independently of the general DD_INTEGRATIONS_ENABLED APM gate. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…ledOverridesFalse Save and restore the original dataStreamsEnabled value instead of hardcoding false in the finally block. This prevents test pollution where an incorrect state could leak to the next test.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Add DATA_STREAMS sampling mechanism (byte value 14) to support decoupling Kafka DSM from APM tracing. This mechanism enforces USER_DROP priority for DSM-only spans and allows overriding locked priority when DSM is enabled. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Use qualified PrioritySampling.USER_DROP form in validateWithSamplingPriority to match the DATA_JOBS case style (line 70) which uses qualified form. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…reams base class Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
When Kafka tracing is disabled but Data Streams Monitoring is enabled, produce-path spans created without a propagated trace context should be marked with PrioritySampling.USER_DROP to avoid billing as APM spans. This implements the 'local-root spans only' scoping from the spec: spans with an extracted parent context (genuine distributed traces) are left completely untouched. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…eams base class Change extends InstrumenterModule.Tracing to extends InstrumenterModule.DataStreams for all 8 instrumentation classes in the kafka-clients-3.8 module. This decouples Data Streams Monitoring from APM tracing and prepares for billing-suppression logic. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…me paths) Add TRACING_ENABLED constant and billing-suppression logic for produce and consume paths to suppress unnecessary billing when APM tracing is disabled but DSM is enabled. The suppression only applies to local-root spans (no extracted parent context) per the DSM-Kafka decoupling plan. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…tion to DataStreams This class populates StreamingContextUpdater's source/disabled-topic registry via BuildAdvice.exit, which KafkaStreamTaskInstrumentation's DSM checkpoint logic reads. Switching from InstrumenterModule.Tracing to DataStreams ensures the registry is populated when DSM is the only reason Kafka is instrumented, allowing DSM's STREAMING_CONTEXT.isSourceTopic()/isDisabledForTopic() checks to work correctly instead of silently falling back to defaults. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Change ConnectWorkerInstrumentation to extend InstrumenterModule.DataStreams instead of InstrumenterModule.Tracing. This class only wraps TaskStatus.Listener for forwarding task status callbacks and does not create spans, making it appropriate for the DataStreams module instead of the Tracing module. Task 11 of DSMS-161 (kafka-connect-0.11: base-class swap) Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
…d tests Converts the Groovy/Spock InstrumenterConfigTest to a JUnit 5 Java test (per repo workflow), preserving all 44 existing cases, and adds explicit coverage for InstrumenterConfig.isDataStreamsEnabled(): default false, and true via DD_DATA_STREAMS_ENABLED. Part of the DSM/Kafka decoupling test task (Task 12a). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Convert the Spock SamplingMechanismTest to a parameterized JUnit 5 Java test, and add DATA_STREAMS cases mirroring the existing DATA_JOBS coverage for validateWithSamplingPriority() and canAvoidSamplingPriorityLock(), per DSMS-161 task 12. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…TREAMS flag Test that AgentInstaller.getEnabledSystems() correctly includes or excludes TargetSystem.DATA_STREAMS based on the data.streams.enabled configuration flag. Also test USM and LLMOBS for consistency. Uses @WithConfig extension to manage configuration state across test cases. Tests verify: - DATA_STREAMS not included when data.streams.enabled is false (default) - DATA_STREAMS included when data.streams.enabled is true - USM not included when usm.enabled is false (default) - USM included when usm.enabled is true - LLMOBS not included when llmobs.enabled is false (default) - LLMOBS included when llmobs.enabled is true Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…lients-0.11 Mirrors the fix already applied to kafka-clients-3.8's TracingIterator: when Kafka APM tracing is disabled but Data Streams Monitoring is enabled, and no genuine propagated trace context was extracted from the record headers, the local-root consume span is marked USER_DROP so it is not billed as a normal APM span on kafka-clients < 3.8.0.
Cover the local-root sampling-priority suppression (USER_DROP) that ProducerAdvice and TracingIterator apply when kafka tracing is disabled (via integrations.enabled or the per-integration trace.kafka.enabled override) but DSM is enabled, plus a regression guard confirming spans that join a real propagated trace are not suppressed. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
… trace `AgentSpan.setSamplingPriority` is trace-level: it resolves to the local root span. The DSM billing-suppression guard was gated on "no context was extracted from the record headers", which is not the same predicate as "this span is the local root". Because the kafka-clients producer and the kafka-streams StreamTask advices create their spans with scope-honouring `startSpan` overloads (USE_SCOPE), the created span can inherit the locally active span -- and since the new DATA_STREAMS mechanism bypasses the sampling priority lock, the guard could force-drop a genuine customer trace (e.g. producing a Kafka message inside an instrumented HTTP request). Producer sites (kafka-clients-0.11 and kafka-clients-3.8) now additionally require `span.getLocalRootSpan() == span`. Entering the else branch already rules out a header-extracted parent, so the local root check is exactly the remaining condition. The kafka-streams sites need a stronger predicate: - `TIME_IN_QUEUE_ENABLED` (on by default) parents the consume span onto the queue span, so `getLocalRootSpan() == span` is never true there; the root candidate is the queue span when one exists. - An `ExtractedContext` parent allocates a fresh TraceCollector, so a span continuing a header-propagated trace is still its own local root. The sibling ContextPropagationAdvice attaches that extracted context to the scope before this advice runs, so the active span captured on entry is what detects it. Both sites therefore require no active span on entry and that the local root is the first span the advice created. Also switches the producer sites from the static `Config.get()` to the remote-config-aware `traceConfig()`, matching the consumer/streams sites so a dynamic DSM toggle cannot make them disagree. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
`MessageListenerInstrumentation` applies Code Origin (span origin) advice to Spring Kafka message listeners -- a purely-APM concern with no Data Streams behaviour. Swapping it to `InstrumenterModule.DataStreams` ORed `isDataStreamsEnabled()` into `isEnabled()`, so with `code.origin.enabled=true` and `data.streams.enabled=true` it installed in DSM-only deployments where Kafka tracing is off. Revert this one module to `InstrumenterModule.Tracing` and record why in a class comment. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…and streams Regression guards for the two billing-suppression bugs, plus the coverage gaps the final review flagged. All four new specs were confirmed red against a temporarily reverted guard. kafka-streams-0.11 (new spec, module previously had no working DSM coverage -- its only end-to-end test is @ignore'd): - a local-root streams consume span is still forced to USER_DROP - a streams consume span continuing a header-propagated trace is not. The local-root case needs kafka.client.propagation.disabled.topics, since an in-JVM producer would otherwise inject its own context into the headers. kafka-clients-0.11: - producing inside an active local trace does not force that trace to USER_DROP -- the one-line test that would have caught the producer bug - strengthen the extracted-parent scenario from "!= USER_DROP" to the concrete expected SAMPLER_KEEP, and assert the consume span's trace id too kafka-clients-3.8 (new spec, module had no suppression coverage at all): the local-root USER_DROP scenario and the active-local-trace regression guard. These extend InstrumentationSpecification rather than KafkaClientTestBase, whose inherited test asserts Code Origin tags that correctly no longer apply once Kafka tracing is off. SamplingMechanismTest: the only DATA_STREAMS row in the parameterized lock-bypass test left DSM at its disabled default, so it passed for a config reason rather than a mechanism one. Add dedicated tests with data.streams.enabled set both ways, including that enabling DSM does not unlock any other mechanism. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…eview findings Fixes a billing gap where the standalone kafka.poll span/trace created for DSM was never forced to USER_DROP, so it would be billed as APM even with Kafka tracing disabled. Also delegates Config.isDataStreamsEnabled() to InstrumenterConfig, dedupes each KafkaDecorator's hardcoded integration-name list into shared constants used by TRACING_ENABLED and every module constructor, extracts the duplicated DSM-drop guard in KafkaStreamTaskInstrumentation into a helper, and rewrites InstrumenterModuleTest to use @WithConfig instead of raw reflection. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
ericfirth
added
type: feature
ericfirth
added
comp: data streams
|
❌ New Groovy Files Detected Please avoid introducing new
Instead, rewrite the new file(s) in Java / JUnit. See the How to Test With JUnit Guide for more details. If this PR needs an exception, add the |
|
🎯 Code Coverage (details) 🔗 Commit SHA: 8f5f263 | Docs | View more details | Give us feedback! |
🟢 Java Benchmark SLOs — All performance SLOs passed
PR vs. master results
Commit: Load and DaCapo benchmarks can be triggered manually in the GitLab pipeline. Results will appear in the Benchmarking Platform UI after completion. |
Kafka / consumer-benchmarkParameters
See matching parameters
SummaryFound 0 performance improvements and 0 performance regressions! Performance is the same for 3 metrics, 0 unstable metrics. See unchanged results
|
| // spans are written out together by TraceStructureWriter when running in strict mode | ||
| } | ||
|
|
||
| if (spanContext == null |
There was a problem hiding this comment.
This guard is skipped whenever spanContext is non-null, but that doesn't account for DD_TRACE_PROPAGATION_BEHAVIOR_EXTRACT=restart/ignore, under which CoreTracer.CoreSpanBuilder.startSpan nulls the extracted context internally and still creates a fresh local-root span. With Kafka tracing disabled + DSM enabled + that propagation mode, a record carrying real x-datadog-* headers extracts a non-null spanContext, so this guard never fires and the resulting local-root span never gets forced to USER_DROP — it gets billed as full APM despite Kafka tracing being off. Same pattern is mirrored in kafka-clients-3.8's TracingIterator.java. Might be worth gating on span.getLocalRootSpan() == span (as KafkaConsumerInfoInstrumentation does for the poll span) instead of the spanContext == null check, since that's what actually determines whether this span is APM-billable.
| return (!Config.get().isApmTracingEnabled() && mechanism == SamplingMechanism.APPSEC) | ||
| || (Config.get().isDataJobsEnabled() && mechanism == DATA_JOBS); | ||
| || (Config.get().isDataJobsEnabled() && mechanism == DATA_JOBS) | ||
| || (Config.get().isDataStreamsEnabled() && mechanism == DATA_STREAMS); |
There was a problem hiding this comment.
This reads the static, startup-only Config.get().isDataStreamsEnabled(), but every Kafka call site that actually forces USER_DROP for DSM gates on the dynamic, remote-config-aware traceConfig().isDataStreamsEnabled(). If DSM is off at JVM startup and later turned on via Remote Config, this method keeps returning false for DATA_STREAMS even though the call site's own dynamic check says DSM is enabled — so the force-drop falls into the CAS-guarded path in DDSpanContext, and if the priority was already set by another mechanism earlier in the span's life, the CAS fails and the DSM-only drop silently never applies (the trace gets billed as APM). Should this use traceConfig().isDataStreamsEnabled() to match the call sites?
| iastFullyDisabled = true; | ||
| telemetryEnabled = false; | ||
| usmEnabled = false; | ||
| dataStreamsEnabled = false; |
There was a problem hiding this comment.
This forces dataStreamsEnabled to false unconditionally under Platform.isNativeImageBuilder(). Before this PR, Config.isDataStreamsEnabled() read data.streams.enabled directly with no native-image carve-out; now Config.isDataStreamsEnabled() delegates entirely to this class (see the Config.java diff), so a GraalVM native-image deployment (e.g. Lambda) that sets DD_DATA_STREAMS_ENABLED=true will silently get DSM disabled — a behavior regression introduced by this PR, not a pre-existing constraint. (isUsmEnabled/isLlmObsEnabled already had this native-image gating before this PR; isDataStreamsEnabled didn't.) Is the native-image restriction intentional for DSM, or should this follow the same pattern as usmEnabled/llmObsEnabled only if DSM genuinely can't work under native-image?
| public static final boolean KAFKA_LEGACY_TRACING = Config.get().isKafkaLegacyTracingEnabled(); | ||
| public static final boolean TIME_IN_QUEUE_ENABLED = | ||
| Config.get().isTimeInQueueEnabled(!KAFKA_LEGACY_TRACING, KAFKA); | ||
| public static final boolean TRACING_ENABLED = |
There was a problem hiding this comment.
Within this module, the DSM-only USER_DROP guard is hand-copied 4 times: twice in TracingIterator.java (using spanContext == null, no local-root check — see my comment there), once in KafkaConsumerInfoInstrumentation.java's poll span, and once in KafkaProducerInstrumentation.java's produce span (the latter two correctly use span.getLocalRootSpan() == span plus a comment explaining why). That the fixed/correct version only made it to 2 of 4 in-module copies is exactly the risk of this duplication. Cross-module sharing with kafka-streams-0.11 etc. is understandably not worth the trouble, but a small static helper here in KafkaDecorator (e.g. maybeForceDsmOnlyDrop(AgentSpan span)), or a private helper in TracingIterator shared between its two call sites, seems low-cost and would have kept all four in-module copies in sync.
|
Heads up: the That test disables the Kafka integration and asserts no This will need a system-tests-side update before merge: either exempt/adjust |
2 participants
Summary
InstrumenterModule.DataStreams, so DSM checkpoints are created even when APM tracing/integrations are disabled.PrioritySampling.USER_DROPviaSamplingMechanism.DATA_STREAMSwhen they are the local root span, so they aren't billed as APM — while never dropping an ambient customer trace.KafkaDecorator's hardcoded integration-name list into shared constants reused byTRACING_ENABLEDand every module constructor, delegatesConfig.isDataStreamsEnabled()toInstrumenterConfig, extracts a duplicated DSM-drop guard inKafkaStreamTaskInstrumentationinto a helper, and rewritesInstrumenterModuleTestto use@WithConfiginstead of raw reflection.Motivation
Copart uses JRuby, which doesn't support DSM. They previously ran dd-trace-java and had DSM working, but moved to dd-trace-rb for better APM tracing — and dd-trace-rb doesn't support DSM either, so their DSM broke. This explores whether Java's DSM instrumentation can run standalone (independent of APM tracing) as a path back to DSM support for cases like this.
This is exploratory — tracked in DSM2-278. It may or may not be accepted upstream.
Test plan
./gradlew :dd-java-agent:instrumentation:kafka-clients-0.11:test./gradlew :dd-java-agent:instrumentation:kafka-clients-3.8:test./gradlew :dd-java-agent:instrumentation:kafka-streams-0.11:test./gradlew :internal-api:test./gradlew :dd-java-agent:agent-tooling:test🤖 Generated with Claude Code