working writer

Author: gary-huang

communication/build.gradle

@@ -4,6 +4,7 @@ dependencies {
   implementation libs.slf4j
 
   api project(':remote-config:remote-config-api')
+  implementation project(':components:json')
   implementation project(':remote-config:remote-config-core')
   implementation project(':internal-api')
   implementation project(':utils:container-utils')

dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java

@@ -37,6 +37,7 @@
 import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
 import datadog.trace.bootstrap.instrumentation.api.AgentTracer.TracerAPI;
 import datadog.trace.bootstrap.instrumentation.api.ProfilingContextIntegration;
+import datadog.trace.bootstrap.instrumentation.api.WriterConstants;
 import datadog.trace.bootstrap.instrumentation.jfr.InstrumentationBasedProfiling;
 import datadog.trace.util.AgentTaskScheduler;
 import datadog.trace.util.AgentThreadFactory.AgentThread;
@@ -111,7 +112,9 @@ private enum AgentFeature {
     DATA_JOBS(propertyNameToSystemPropertyName(GeneralConfig.DATA_JOBS_ENABLED), false),
     AGENTLESS_LOG_SUBMISSION(
         propertyNameToSystemPropertyName(GeneralConfig.AGENTLESS_LOG_SUBMISSION_ENABLED), false),
-    LLMOBS(propertyNameToSystemPropertyName(LlmObsConfig.LLMOBS_ENABLED), false);
+    LLMOBS(propertyNameToSystemPropertyName(LlmObsConfig.LLMOBS_ENABLED), false),
+    LLMOBS_AGENTLESS(
+        propertyNameToSystemPropertyName(LlmObsConfig.LLMOBS_AGENTLESS_ENABLED), false);
 
     private final String systemProp;
     private final boolean enabledByDefault;
@@ -153,6 +156,7 @@ public boolean isEnabledByDefault() {
   private static boolean cwsEnabled = false;
   private static boolean ciVisibilityEnabled = false;
   private static boolean llmObsEnabled = false;
+  private static boolean llmObsAgentlessEnabled = false;
   private static boolean usmEnabled = false;
   private static boolean telemetryEnabled = true;
   private static boolean debuggerEnabled = false;
@@ -273,6 +277,23 @@ public static void start(
     agentlessLogSubmissionEnabled = isFeatureEnabled(AgentFeature.AGENTLESS_LOG_SUBMISSION);
     llmObsEnabled = isFeatureEnabled(AgentFeature.LLMOBS);
 
+    if (llmObsEnabled) {
+      // for llm obs spans, use agent proxy by default, apm spans will use agent writer
+      setSystemPropertyDefault(
+          propertyNameToSystemPropertyName(TracerConfig.WRITER_TYPE),
+          WriterConstants.MULTI_WRITER_TYPE
+              + ":"
+              + WriterConstants.DD_INTAKE_WRITER_TYPE
+              + ","
+              + WriterConstants.DD_AGENT_WRITER_TYPE);
+      if (llmObsAgentlessEnabled) {
+        // use API writer only
+        setSystemPropertyDefault(
+            propertyNameToSystemPropertyName(TracerConfig.WRITER_TYPE),
+            WriterConstants.DD_INTAKE_WRITER_TYPE);
+      }
+    }
+
     if (profilingEnabled) {
       if (!isOracleJDK8()) {
         // Profiling agent startup code is written in a way to allow `startProfilingAgent` be called

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

@@ -68,6 +68,11 @@ dependencies {
   implementation project(':components:json')
   implementation project(':utils:container-utils')
   implementation project(':utils:socket-utils')
+
+  implementation group: 'org.msgpack', name: 'msgpack-core', version: '0.8.10'
+  implementation group: 'org.msgpack', name: 'jackson-dataformat-msgpack', version: '0.8.10'
+  implementation group: 'com.fasterxml.jackson.core', name: 'jackson-core', version: '2.10.0'
+
   // 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;
     }