Gary/add llmobs svc functionalities

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')

communication/src/main/java/datadog/communication/BackendApiFactory.java

@@ -72,6 +72,7 @@ private HttpUrl getAgentlessUrl(Intake intake) {
 
   public enum Intake {
     API("api", "v2", Config::isCiVisibilityAgentlessEnabled, Config::getCiVisibilityAgentlessUrl),
+    LLMOBS_API("api", "v2", Config::isLlmObsAgentlessEnabled, Config::getLlMObsAgentlessUrl),
     LOGS(
         "http-intake.logs",
         "v2",

communication/src/main/java/datadog/communication/serialization/json/JSONWritableFormatter.java

@@ -0,0 +1,131 @@
+package datadog.communication.serialization.json;
+
+import datadog.communication.serialization.EncodingCache;
+import datadog.communication.serialization.Mapper;
+import datadog.communication.serialization.WritableFormatter;
+import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+public class JSONWritableFormatter implements WritableFormatter {
+
+  @Override
+  public <T> boolean format(T message, Mapper<T> mapper) {
+    return false;
+  }
+
+  @Override
+  public void flush() {
+
+  }
+
+  @Override
+  public void writeNull() {
+
+  }
+
+  @Override
+  public void writeBoolean(boolean value) {
+
+  }
+
+  @Override
+  public void writeObject(Object value, EncodingCache encodingCache) {
+
+  }
+
+  @Override
+  public void writeObjectString(Object value, EncodingCache encodingCache) {
+
+  }
+
+  @Override
+  public void writeMap(Map<? extends CharSequence, ?> map, EncodingCache encodingCache) {
+
+  }
+
+  @Override
+  public void writeString(CharSequence s, EncodingCache encodingCache) {
+
+  }
+
+  @Override
+  public void writeUTF8(byte[] string, int offset, int length) {
+
+  }
+
+  @Override
+  public void writeUTF8(byte[] string) {
+
+  }
+
+  @Override
+  public void writeUTF8(UTF8BytesString string) {
+
+  }
+
+  @Override
+  public void writeBinary(byte[] binary) {
+
+  }
+
+  @Override
+  public void writeBinary(byte[] binary, int offset, int length) {
+
+  }
+
+  @Override
+  public void startMap(int elementCount) {
+
+  }
+
+  @Override
+  public void startStruct(int elementCount) {
+
+  }
+
+  @Override
+  public void startArray(int elementCount) {
+
+  }
+
+  @Override
+  public void writeBinary(ByteBuffer buffer) {
+
+  }
+
+  @Override
+  public void writeInt(int value) {
+
+  }
+
+  @Override
+  public void writeSignedInt(int value) {
+
+  }
+
+  @Override
+  public void writeLong(long value) {
+
+  }
+
+  @Override
+  public void writeUnsignedLong(long value) {
+
+  }
+
+  @Override
+  public void writeSignedLong(long value) {
+
+  }
+
+  @Override
+  public void writeFloat(float value) {
+
+  }
+
+  @Override
+  public void writeDouble(double value) {
+
+  }
+}

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

@@ -23,6 +23,7 @@
 import datadog.trace.api.config.GeneralConfig;
 import datadog.trace.api.config.IastConfig;
 import datadog.trace.api.config.JmxFetchConfig;
+import datadog.trace.api.config.LlmObsConfig;
 import datadog.trace.api.config.ProfilingConfig;
 import datadog.trace.api.config.RemoteConfigConfig;
 import datadog.trace.api.config.TraceInstrumentationConfig;
@@ -36,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;
@@ -109,7 +111,9 @@ private enum AgentFeature {
         false),
     DATA_JOBS(propertyNameToSystemPropertyName(GeneralConfig.DATA_JOBS_ENABLED), false),
     AGENTLESS_LOG_SUBMISSION(
-        propertyNameToSystemPropertyName(GeneralConfig.AGENTLESS_LOG_SUBMISSION_ENABLED), false);
+        propertyNameToSystemPropertyName(GeneralConfig.AGENTLESS_LOG_SUBMISSION_ENABLED), false),
+    LLMOBS(propertyNameToSystemPropertyName(LlmObsConfig.LLMOBS_ENABLED), false),
+    LLMOBS_AGENTLESS(propertyNameToSystemPropertyName(LlmObsConfig.LLMOBS_AGENTLESS_ENABLED), false);
 
     private final String systemProp;
     private final boolean enabledByDefault;
@@ -150,6 +154,8 @@ public boolean isEnabledByDefault() {
   private static boolean iastFullyDisabled;
   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;
@@ -268,6 +274,16 @@ public static void start(
     exceptionDebuggingEnabled = isFeatureEnabled(AgentFeature.EXCEPTION_DEBUGGING);
     spanOriginEnabled = isFeatureEnabled(AgentFeature.SPAN_ORIGIN);
     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()) {
@@ -556,6 +572,7 @@ public void execute() {
 
       maybeStartAppSec(scoClass, sco);
       maybeStartCiVisibility(instrumentation, scoClass, sco);
+      maybeStartLLMObs(instrumentation, scoClass, sco);
       // start debugger before remote config to subscribe to it before starting to poll
       maybeStartDebugger(instrumentation, scoClass, sco);
       maybeStartRemoteConfig(scoClass, sco);
@@ -903,6 +920,24 @@ private static void maybeStartCiVisibility(Instrumentation inst, Class<?> scoCla
     }
   }
 
+  private static void maybeStartLLMObs(Instrumentation inst, Class<?> scoClass, Object sco) {
+    if (llmObsEnabled) {
+      StaticEventLogger.begin("LLM Observability");
+
+      try {
+        final Class<?> llmObsSysClass =
+            AGENT_CLASSLOADER.loadClass("datadog.trace.llmobs.LLMObsSystem");
+        final Method llmObsInstallerMethod =
+            llmObsSysClass.getMethod("start", Instrumentation.class, scoClass);
+        llmObsInstallerMethod.invoke(null, inst, sco);
+      } catch (final Throwable e) {
+        log.warn("Not starting LLM Observability subsystem", e);
+      }
+
+      StaticEventLogger.end("LLM Observability");
+    }
+  }
+
   private static void maybeInstallLogsIntake(Class<?> scoClass, Object sco) {
     if (agentlessLogSubmissionEnabled) {
       StaticEventLogger.begin("Logs Intake");

dd-java-agent/agent-llmobs/build.gradle

@@ -0,0 +1,60 @@
+buildscript {
+  repositories {
+    mavenCentral()
+  }
+
+  dependencies {
+    classpath group: 'org.jetbrains.kotlin', name: 'kotlin-gradle-plugin', version: libs.versions.kotlin.get()
+  }
+}
+
+plugins {
+  id 'com.github.johnrengelman.shadow'
+  id 'java-test-fixtures'
+}
+
+apply from: "$rootDir/gradle/java.gradle"
+apply from: "$rootDir/gradle/version.gradle"
+apply from: "$rootDir/gradle/test-with-kotlin.gradle"
+apply from: "$rootDir/gradle/test-with-scala.gradle"
+
+minimumBranchCoverage = 0.0
+minimumInstructionCoverage = 0.0
+
+dependencies {
+  api libs.slf4j
+
+  implementation libs.bundles.asm
+  implementation group: 'org.jacoco', name: 'org.jacoco.core', version: '0.8.12'
+  implementation group: 'org.jacoco', name: 'org.jacoco.report', version: '0.8.12'
+
+  implementation project(':communication')
+  implementation project(':components:json')
+  implementation project(':internal-api')
+  implementation project(':internal-api:internal-api-9')
+
+  testImplementation project(":utils:test-utils")
+  testImplementation("com.google.jimfs:jimfs:1.1") // an in-memory file system for testing code that works with files
+
+  testImplementation libs.scala
+  testImplementation libs.kotlin
+
+  testFixturesApi project(':dd-java-agent:testing')
+  testFixturesApi project(':utils:test-utils')
+
+  testFixturesApi group: 'org.skyscreamer', name: 'jsonassert', version: '1.5.1'
+  testFixturesApi group: 'org.freemarker', name: 'freemarker', version: '2.3.30'
+  testFixturesApi group: 'com.jayway.jsonpath', name: 'json-path', version: '2.8.0'
+  testFixturesApi group: 'com.fasterxml.jackson.core', name: 'jackson-databind', version: '2.16.0'
+  testFixturesApi group: 'org.msgpack', name: 'jackson-dataformat-msgpack', version: '0.9.6'
+
+  testRuntimeOnly("org.junit.platform:junit-platform-launcher:1.9.2") // Required to update dependency lock files
+}
+
+shadowJar {
+  dependencies deps.excludeShared
+}
+
+jar {
+  archiveClassifier = 'unbundled'
+}

dd-java-agent/agent-llmobs/src/main/java/datadog/trace/llmobs/LLMObsServices.java

@@ -0,0 +1,59 @@
+package datadog.trace.llmobs;
+
+import datadog.communication.BackendApi;
+import datadog.communication.BackendApiFactory;
+import datadog.communication.ddagent.SharedCommunicationObjects;
+import datadog.trace.api.Config;
+import datadog.trace.llmobs.domain.SpanContextInfo;
+import java.util.Stack;
+import javax.annotation.Nonnull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LLMObsServices {
+
+  private static final Logger logger = LoggerFactory.getLogger(LLMObsServices.class);
+
+  final Config config;
+  final BackendApi backendApi;
+
+  ThreadLocal<Stack<SpanContextInfo>> activeSpanContext = new ThreadLocal<>();
+
+  LLMObsServices(Config config, SharedCommunicationObjects sco) {
+    this.config = config;
+    this.backendApi =
+        new BackendApiFactory(config, sco).createBackendApi(BackendApiFactory.Intake.LLMOBS_API);
+  }
+
+  public SpanContextInfo getParentContext() {
+    if (activeSpanContext.get() == null || activeSpanContext.get().isEmpty()) {
+      return null;
+    }
+    return activeSpanContext.get().peek();
+  }
+
+  @Nonnull
+  public SpanContextInfo getActiveSpanContext() {
+    if (activeSpanContext.get() == null || activeSpanContext.get().isEmpty()) {
+      return new SpanContextInfo();
+    }
+    return activeSpanContext.get().peek();
+  }
+
+  public void setActiveSpanContext(SpanContextInfo spanContext) {
+    Stack<SpanContextInfo> contexts = activeSpanContext.get();
+    if (contexts == null) {
+      contexts = new Stack<>();
+    }
+    contexts.push(spanContext);
+    this.activeSpanContext.set(contexts);
+  }
+
+  public void removeActiveSpanContext() {
+    Stack<SpanContextInfo> contexts = activeSpanContext.get();
+    if (contexts == null) {
+      return;
+    }
+    contexts.pop();
+  }
+}