From 4196c2186d9aea4404da68a5aa676747354832d7 Mon Sep 17 00:00:00 2001 From: Rey Abolofia Date: Thu, 20 Aug 2026 13:44:54 -0700 Subject: [PATCH 1/3] Emit LLM Obs spans over the intake track when using DDAgentWriter The DDAgentWriter branch of WriterFactory never wired up an LLM Obs DDIntakeWriter, so LLM Obs spans were silently dropped whenever the tracer picked DDAgentWriter (e.g. in Lambda/serverless mode with CI Visibility disabled). Broadcast to a dedicated LLM Obs DDIntakeWriter via MultiWriter, flushed synchronously in serverless environments just like the primary writer. --- .../trace/common/writer/WriterFactory.java | 21 ++++++++- .../common/writer/WriterFactoryTest.groovy | 46 +++++++++++++++++++ 2 files changed, 66 insertions(+), 1 deletion(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java index 73875a0e408..b85d0918199 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java @@ -134,7 +134,7 @@ public static Writer createWriter( } } - RemoteWriter remoteWriter; + Writer remoteWriter; if (DD_INTAKE_WRITER_TYPE.equals(configuredType)) { final TrackType trackType = DDIntakeTrackTypeResolver.resolve(config); final RemoteApi remoteApi = @@ -215,6 +215,25 @@ public static Writer createWriter( } remoteWriter = builder.build(); + + // DDAgentWriter only speaks the regular trace protocol, so LLM Observability spans + // (tagged with DDSpanTypes.LLMOBS) need their own track sent via the EVP proxy -- without + // this, they're silently dropped by the agent/extension instead of reaching LLM Obs + // intake. Flush this track synchronously too when the primary writer is (i.e. in a + // serverless environment), so spans aren't lost when the execution environment freezes. + if (config.isLlmObsEnabled()) { + final RemoteApi llmObsApi = + createDDIntakeRemoteApi(config, commObjects, featuresDiscovery, TrackType.LLMOBS); + final DDIntakeWriter llmObsWriter = + DDIntakeWriter.builder() + .addTrack(TrackType.LLMOBS, llmObsApi) + .healthMetrics(healthMetrics) + .monitoring(commObjects.monitoring) + .alwaysFlush(alwaysFlush) + .flushIntervalMilliseconds(flushIntervalMilliseconds) + .build(); + remoteWriter = new MultiWriter(new Writer[] {remoteWriter, llmObsWriter}); + } } return remoteWriter; diff --git a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy index d20bd475cac..40e34448390 100644 --- a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy +++ b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy @@ -168,6 +168,52 @@ class WriterFactoryTest extends DDSpecification { "DDIntakeWriter" | false | false | true | DDIntakeWriter | [DDIntakeApi] } + def "test DDAgentWriter also wires up an LLM Observability track when llm obs is enabled"() { + setup: + def config = Mock(Config) + config.apiKey >> "my-api-key" + config.agentUrl >> "http://my-agent.url" + config.getEnumValue(PRIORITIZATION_TYPE, _, _) >> Prioritization.FAST_LANE + config.tracerMetricsEnabled >> true + config.isLlmObsEnabled() >> true + config.llmObsAgentlessEnabled >> false + + def response = buildHttpResponse(true, true, HttpUrl.parse(config.agentUrl + "/info")) + def mockCall = Mock(Call) + def mockHttpClient = Mock(OkHttpClient) + mockCall.execute() >> { + sleep(400) + return response + } + mockHttpClient.newCall(_ as Request) >> mockCall + + def sharedComm = new SharedCommunicationObjects() + sharedComm.agentHttpClient = mockHttpClient + sharedComm.agentUrl = HttpUrl.parse(config.agentUrl) + sharedComm.createRemaining(config) + + def sampler = Mock(Sampler) + + when: + def writer = WriterFactory.createWriter(config, sharedComm, sampler, null, HealthMetrics.NO_OP, "DDAgentWriter") + + then: + writer.class == MultiWriter + def subWriters = readField(writer, "writers") + subWriters.length == 2 + subWriters[0].class == DDAgentWriter + subWriters[1].class == DDIntakeWriter + def llmObsApis = ((RemoteWriter) subWriters[1]).apis + llmObsApis.size() == 1 + llmObsApis[0].class == DDEvpProxyApi + } + + static readField(Object instance, String fieldName) { + def field = instance.class.getDeclaredField(fieldName) + field.setAccessible(true) + return field.get(instance) + } + def "test writer creation for OtlpWriter wires #protocol+#compression"() { setup: def config = Mock(Config) From 487006b7f4c29d462d151006a120e0a8306317a9 Mon Sep 17 00:00:00 2001 From: Rey Abolofia Date: Mon, 24 Aug 2026 13:29:29 -0700 Subject: [PATCH 2/3] Flush the DDIntakeWriter track instead of adding a second LLM Obs writer The DDAgentWriter branch's dedicated LLM Obs DDIntakeWriter duplicated the LLM Obs track that the DD_INTAKE_WRITER_TYPE branch already builds whenever llmObsEnabled is true (Agent.java defaults to MultiWriter:DDIntakeWriter,DDAgentWriter for that case), so every span was sent twice. Verified live on Lambda: the pre-fix build sent LLMOBS/v2 payloads twice per invocation. The original bug this branch was fixing -- spans silently dropped in Lambda -- was actually caused by the DDIntakeWriter branch never setting alwaysFlush, so the periodic flush timer rarely beat the execution environment freezing after the handler returns. Verified live: without alwaysFlush, 0/8 invocations delivered a span; with it, 8/8 did, one send each. --- .../trace/common/writer/WriterFactory.java | 29 ++++-------- .../common/writer/WriterFactoryTest.groovy | 46 ------------------- 2 files changed, 9 insertions(+), 66 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java index b85d0918199..901e5d8c980 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java @@ -134,12 +134,19 @@ public static Writer createWriter( } } - Writer remoteWriter; + RemoteWriter remoteWriter; if (DD_INTAKE_WRITER_TYPE.equals(configuredType)) { final TrackType trackType = DDIntakeTrackTypeResolver.resolve(config); final RemoteApi remoteApi = createDDIntakeRemoteApi(config, commObjects, featuresDiscovery, trackType); + // In a serverless environment the execution environment can freeze as soon as the handler + // returns, before this writer's periodic flush timer next fires -- flush synchronously so + // buffered events (e.g. LLM Observability spans) aren't lost when that happens. + boolean alwaysFlush = + config.isAgentConfiguredUsingDefault() + && ServerlessInfo.get().isRunningInServerlessEnvironment(); + DDIntakeWriter.DDIntakeWriterBuilder builder = DDIntakeWriter.builder() .addTrack(trackType, remoteApi) @@ -147,6 +154,7 @@ public static Writer createWriter( .healthMetrics(healthMetrics) .monitoring(commObjects.monitoring) .singleSpanSampler(singleSpanSampler) + .alwaysFlush(alwaysFlush) .flushIntervalMilliseconds(flushIntervalMilliseconds); if (config.isCiVisibilityEnabled()) { @@ -215,25 +223,6 @@ public static Writer createWriter( } remoteWriter = builder.build(); - - // DDAgentWriter only speaks the regular trace protocol, so LLM Observability spans - // (tagged with DDSpanTypes.LLMOBS) need their own track sent via the EVP proxy -- without - // this, they're silently dropped by the agent/extension instead of reaching LLM Obs - // intake. Flush this track synchronously too when the primary writer is (i.e. in a - // serverless environment), so spans aren't lost when the execution environment freezes. - if (config.isLlmObsEnabled()) { - final RemoteApi llmObsApi = - createDDIntakeRemoteApi(config, commObjects, featuresDiscovery, TrackType.LLMOBS); - final DDIntakeWriter llmObsWriter = - DDIntakeWriter.builder() - .addTrack(TrackType.LLMOBS, llmObsApi) - .healthMetrics(healthMetrics) - .monitoring(commObjects.monitoring) - .alwaysFlush(alwaysFlush) - .flushIntervalMilliseconds(flushIntervalMilliseconds) - .build(); - remoteWriter = new MultiWriter(new Writer[] {remoteWriter, llmObsWriter}); - } } return remoteWriter; diff --git a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy index 40e34448390..d20bd475cac 100644 --- a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy +++ b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy @@ -168,52 +168,6 @@ class WriterFactoryTest extends DDSpecification { "DDIntakeWriter" | false | false | true | DDIntakeWriter | [DDIntakeApi] } - def "test DDAgentWriter also wires up an LLM Observability track when llm obs is enabled"() { - setup: - def config = Mock(Config) - config.apiKey >> "my-api-key" - config.agentUrl >> "http://my-agent.url" - config.getEnumValue(PRIORITIZATION_TYPE, _, _) >> Prioritization.FAST_LANE - config.tracerMetricsEnabled >> true - config.isLlmObsEnabled() >> true - config.llmObsAgentlessEnabled >> false - - def response = buildHttpResponse(true, true, HttpUrl.parse(config.agentUrl + "/info")) - def mockCall = Mock(Call) - def mockHttpClient = Mock(OkHttpClient) - mockCall.execute() >> { - sleep(400) - return response - } - mockHttpClient.newCall(_ as Request) >> mockCall - - def sharedComm = new SharedCommunicationObjects() - sharedComm.agentHttpClient = mockHttpClient - sharedComm.agentUrl = HttpUrl.parse(config.agentUrl) - sharedComm.createRemaining(config) - - def sampler = Mock(Sampler) - - when: - def writer = WriterFactory.createWriter(config, sharedComm, sampler, null, HealthMetrics.NO_OP, "DDAgentWriter") - - then: - writer.class == MultiWriter - def subWriters = readField(writer, "writers") - subWriters.length == 2 - subWriters[0].class == DDAgentWriter - subWriters[1].class == DDIntakeWriter - def llmObsApis = ((RemoteWriter) subWriters[1]).apis - llmObsApis.size() == 1 - llmObsApis[0].class == DDEvpProxyApi - } - - static readField(Object instance, String fieldName) { - def field = instance.class.getDeclaredField(fieldName) - field.setAccessible(true) - return field.get(instance) - } - def "test writer creation for OtlpWriter wires #protocol+#compression"() { setup: def config = Mock(Config) From b83d309c8f4b414c88b9db5065b0f0f21e506198 Mon Sep 17 00:00:00 2001 From: Rey Abolofia Date: Tue, 25 Aug 2026 14:36:00 -0700 Subject: [PATCH 3/3] Clarify alwaysFlush comment refers to Lambda, not serverless generally --- .../java/datadog/trace/common/writer/WriterFactory.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java index 901e5d8c980..da4df51fd82 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java @@ -140,9 +140,9 @@ public static Writer createWriter( final RemoteApi remoteApi = createDDIntakeRemoteApi(config, commObjects, featuresDiscovery, trackType); - // In a serverless environment the execution environment can freeze as soon as the handler - // returns, before this writer's periodic flush timer next fires -- flush synchronously so - // buffered events (e.g. LLM Observability spans) aren't lost when that happens. + // In Lambda the execution environment can freeze as soon as the handler returns, before + // this writer's periodic flush timer next fires -- flush synchronously so buffered events + // (e.g. LLM Observability spans) aren't lost when that happens. boolean alwaysFlush = config.isAgentConfiguredUsingDefault() && ServerlessInfo.get().isRunningInServerlessEnvironment();