working writer

Author: gary-huang

dd-trace-api/src/main/java/datadog/trace/api/llmobs/LLMObsConstants.java

@@ -68,6 +68,7 @@ dependencies {
   implementation project(':components:json')
   implementation project(':utils:container-utils')
   implementation project(':utils:socket-utils')
+
   // for span exception debugging
   compileOnly project(':dd-java-agent:agent-debugger:debugger-bootstrap')
 

dd-trace-core/build.gradle

@@ -151,6 +151,7 @@ public DDAgentWriter build() {
       }
 
       final DDAgentMapperDiscovery mapperDiscovery = new DDAgentMapperDiscovery(featureDiscovery);
+
       final PayloadDispatcher dispatcher =
           new PayloadDispatcherImpl(mapperDiscovery, agentApi, healthMetrics, monitoring);
       final TraceProcessingWorker traceProcessingWorker =

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

@@ -82,7 +82,7 @@ public static Writer createWriter(
 
     // The AgentWriter doesn't support the CI Visibility protocol. If CI Visibility is
     // enabled, check if we can use the IntakeWriter instead.
-    if (DD_AGENT_WRITER_TYPE.equals(configuredType) && config.isCiVisibilityEnabled()) {
+    if (DD_AGENT_WRITER_TYPE.equals(configuredType) && (config.isCiVisibilityEnabled())) {
       if (featuresDiscovery.supportsEvpProxy() || config.isCiVisibilityAgentlessEnabled()) {
         configuredType = DD_INTAKE_WRITER_TYPE;
       } else {
@@ -116,6 +116,10 @@ public static Writer createWriter(
         builder.addTrack(TrackType.CITESTCOV, coverageApi);
       }
 
+      final RemoteApi llmobsApi =
+          createDDIntakeRemoteApi(config, commObjects, featuresDiscovery, TrackType.LLMOBS);
+      builder.addTrack(TrackType.LLMOBS, llmobsApi);
+
       remoteWriter = builder.build();
 
     } else { // configuredType == DDAgentWriter
@@ -171,20 +175,34 @@ private static RemoteApi createDDIntakeRemoteApi(
       SharedCommunicationObjects commObjects,
       DDAgentFeaturesDiscovery featuresDiscovery,
       TrackType trackType) {
-    if (featuresDiscovery.supportsEvpProxy() && !config.isCiVisibilityAgentlessEnabled()) {
+    boolean evpProxySupported = featuresDiscovery.supportsEvpProxy();
+    boolean useProxyApi =
+        (evpProxySupported && TrackType.LLMOBS == trackType && !config.isLlmObsAgentlessEnabled())
+            || (evpProxySupported
+                && (TrackType.CITESTCOV == trackType || TrackType.CITESTCYCLE == trackType)
+                && !config.isCiVisibilityAgentlessEnabled());
+
+    if (useProxyApi) {
       return DDEvpProxyApi.builder()
           .httpClient(commObjects.okHttpClient)
           .agentUrl(commObjects.agentUrl)
           .evpProxyEndpoint(featuresDiscovery.getEvpProxyEndpoint())
           .trackType(trackType)
           .compressionEnabled(featuresDiscovery.supportsContentEncodingHeadersWithEvpProxy())
           .build();
-
     } else {
       HttpUrl hostUrl = null;
+      String llmObsAgentlessUrl = config.getLlMObsAgentlessUrl();
+
       if (config.getCiVisibilityAgentlessUrl() != null) {
         hostUrl = HttpUrl.get(config.getCiVisibilityAgentlessUrl());
         log.info("Using host URL '{}' to report CI Visibility traces in Agentless mode.", hostUrl);
+      } else if (config.isLlmObsEnabled()
+          && config.isLlmObsAgentlessEnabled()
+          && llmObsAgentlessUrl != null
+          && !llmObsAgentlessUrl.isEmpty()) {
+        hostUrl = HttpUrl.get(llmObsAgentlessUrl);
+        log.info("Using host URL '{}' to report LLM Obs traces in Agentless mode.", hostUrl);
       }
       return DDIntakeApi.builder()
           .hostUrl(hostUrl)

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

@@ -90,6 +90,7 @@ public void addResponseListener(final RemoteResponseListener listener) {
   public Response sendSerializedTraces(final Payload payload) {
     final int sizeInBytes = payload.sizeInBytes();
     String tracesEndpoint = featuresDiscovery.getTraceEndpoint();
+
     if (null == tracesEndpoint) {
       featuresDiscovery.discoverIfOutdated();
       tracesEndpoint = featuresDiscovery.getTraceEndpoint();

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

@@ -131,6 +131,7 @@ public Response sendSerializedTraces(Payload payload) {
             .post(payload.toRequest())
             .tag(OkHttpUtils.CustomListener.class, telemetryListener)
             .build();
+
     totalTraces += payload.traceCount();
     receivedTraces += payload.traceCount();
 

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

@@ -6,6 +6,7 @@
 import datadog.trace.civisibility.writer.ddintake.CiTestCycleMapperV1;
 import datadog.trace.common.writer.RemoteMapper;
 import datadog.trace.common.writer.RemoteMapperDiscovery;
+import datadog.trace.llmobs.writer.ddintake.LLMObsSpanMapper;
 
 /**
  * Mapper discovery logic when a DDIntake is used. The mapper is discovered based on a backend
@@ -40,6 +41,8 @@ public void discover() {
       mapper = new CiTestCycleMapperV1(wellKnownTags, compressionEnabled);
     } else if (TrackType.CITESTCOV.equals(trackType)) {
       mapper = new CiTestCovMapperV2(compressionEnabled);
+    } else if (TrackType.LLMOBS.equals(trackType)) {
+      mapper = new LLMObsSpanMapper();
     } else {
       mapper = RemoteMapper.NO_OP;
     }