LLM Obs SDK Mapper (#8372)

* add APIs for llm obs

* add llm message class to support llm spans

* add llm message class to support llm spans

* impl llmobs agent and llmobs apis

* support llm messages with tool calls

* handle default model name and provider

* rm unneeded file

* impl llmobs agent and llmobs apis

* impl llmobs agent

* working writer

* add support for llm message and tool calls

* cleaned up whitespace

* resolve merge conflicts

* remaining merge conflicts

* fix bad method call

* fixed llmobs intake creation if llmobs not enabled

* removed print statements

* added tests for llmobsspanmapper

* fixed coverage for tags

---------

Co-authored-by: Nayeem Kamal <[email protected]>

Author: gary-huang

dd-java-agent/agent-jmxfetch/integrations-core

@@ -51,6 +51,7 @@ excludedClassesCoverage += [
   'datadog.trace.context.NoopTraceScope',
   'datadog.trace.payloadtags.PayloadTagsData',
   'datadog.trace.payloadtags.PayloadTagsData.PathAndValue',
+  'datadog.trace.api.llmobs.LLMObsTags',
 ]
 
 description = 'dd-trace-api'

dd-trace-api/build.gradle

@@ -36,6 +36,7 @@ public class DDSpanTypes {
   public static final String PROTOBUF = "protobuf";
 
   public static final String MULE = "mule";
+
   public static final String VALKEY = "valkey";
   public static final String WEBSOCKET = "websocket";
 

dd-trace-api/src/main/java/datadog/trace/api/DDSpanTypes.java

@@ -37,6 +37,8 @@ excludedClassesCoverage += [
   // Interface with an empty defender method
   'datadog.trace.core.propagation.HttpCodec.Extractor',
   'datadog.trace.core.flare.*',
+  'datadog.trace.llmobs.writer.ddintake.LLMObsSpanMapper',
+  'datadog.trace.llmobs.writer.ddintake.LLMObsSpanMapper.PayloadV1',
   // FIXME(DSM): test coverage needed
   'datadog.trace.core.datastreams.DataStreamContextInjector',
   'datadog.trace.common.sampling.TraceSamplingRules.RuleAdapter',

dd-trace-core/build.gradle

@@ -68,7 +68,8 @@ public static Writer createWriter(
     if (!DD_AGENT_WRITER_TYPE.equals(configuredType)
         && !DD_INTAKE_WRITER_TYPE.equals(configuredType)) {
       log.warn(
-          "Writer type not configured correctly: Type {} not recognized. Ignoring", configuredType);
+          "Writer type not configured correctly: Type {} not recognized. Ignoring ",
+          configuredType);
       configuredType = datadog.trace.api.ConfigDefaults.DEFAULT_AGENT_WRITER_TYPE;
     }
 
@@ -84,14 +85,22 @@ 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 {
         log.info(
             "CI Visibility functionality is limited. Please upgrade to Agent v6.40+ or v7.40+ or enable Agentless mode.");
       }
     }
+    if (DD_AGENT_WRITER_TYPE.equals(configuredType) && (config.isLlmObsEnabled())) {
+      if (featuresDiscovery.supportsEvpProxy() || config.isLlmObsAgentlessEnabled()) {
+        configuredType = DD_INTAKE_WRITER_TYPE;
+      } else {
+        log.info("LLM Observability functionality is limited.");
+        // TODO: add supported agent version to this log line for llm obs
+      }
+    }
 
     RemoteWriter remoteWriter;
     if (DD_INTAKE_WRITER_TYPE.equals(configuredType)) {
@@ -117,7 +126,11 @@ public static Writer createWriter(
             createDDIntakeRemoteApi(config, commObjects, featuresDiscovery, TrackType.CITESTCOV);
         builder.addTrack(TrackType.CITESTCOV, coverageApi);
       }
-
+      if (config.isLlmObsEnabled()) {
+        final RemoteApi llmobsApi =
+            createDDIntakeRemoteApi(config, commObjects, featuresDiscovery, TrackType.LLMOBS);
+        builder.addTrack(TrackType.LLMOBS, llmobsApi);
+      }
       remoteWriter = builder.build();
 
     } else { // configuredType == DDAgentWriter
@@ -173,20 +186,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

@@ -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;
     }