feat(llmobs): evaluation metrics can be submitted by agent proxy (#5466)

* rough implementation

* pass in agentlessEnabled for eval metrics writer

* all tests except module tests fixed

* default agentlessEnabled to undefined

* fix tests

* try check against agent

* enable tests for checking agent running

* fix llmobs plugins

* fix tests

* remove uncaught exception in favor of better solution later

* remove .only

* fix tests

* fix tests

* fix tests and error content

* fix test

* add additional agent-proxy endpoint parsing and error parsing

* simplify

* mostly fix tests

* refactor _url

* remove cb references for now

* reset test

* more readable

* more readable

* maybe fix tests

* update constants

* clean up merge

* only compute url on agentless change

* remove utility function

* fmt

* add agentlessEnabled back for plugins

* fix vertexai

* error logic

* pass processedEventSizeBytes to super.append

* remove api key enforcement in submitEvaluation

Author: sabrenner

docs/test.ts

@@ -574,7 +574,7 @@ const enabled = llmobs.enabled
 // manually enable
 llmobs.enable({
   mlApp: 'mlApp',
-  agentlessEnabled: true
+  agentlessEnabled: false
 })
 
 // manually disable
@@ -594,7 +594,7 @@ llmobs.wrap({ kind: 'llm' }, function myLLM () {})()
 llmobs.wrap({ kind: 'llm', name: 'myLLM', modelName: 'myModel', modelProvider: 'myProvider' }, function myFunction () {})()
 
 // export a span
-llmobs.enable({ mlApp: 'myApp' })
+llmobs.enable({ mlApp: 'myApp', agentlessEnabled: false })
 llmobs.trace({ kind: 'llm', name: 'myLLM' }, (span) => {
   const llmobsSpanCtx = llmobs.exportSpan(span)
   llmobsSpanCtx.traceId;

packages/dd-trace/src/config.js

@@ -530,7 +530,7 @@ class Config {
     this._setValue(defaults, 'isManualApiEnabled', false)
     this._setValue(defaults, 'langchain.spanCharLimit', 128)
     this._setValue(defaults, 'langchain.spanPromptCompletionSampleRate', 1.0)
-    this._setValue(defaults, 'llmobs.agentlessEnabled', false)
+    this._setValue(defaults, 'llmobs.agentlessEnabled', undefined)
     this._setValue(defaults, 'llmobs.enabled', false)
     this._setValue(defaults, 'llmobs.mlApp', undefined)
     this._setValue(defaults, 'ciVisibilityTestSessionName', '')

packages/dd-trace/src/llmobs/constants/writers.js

@@ -1,12 +1,16 @@
 'use strict'
 
 module.exports = {
-  EVP_PROXY_AGENT_BASE_PATH: 'evp_proxy/v2',
-  EVP_PROXY_AGENT_ENDPOINT: 'evp_proxy/v2/api/v2/llmobs',
+  EVP_PROXY_AGENT_BASE_PATH: '/evp_proxy/v2/',
   EVP_SUBDOMAIN_HEADER_NAME: 'X-Datadog-EVP-Subdomain',
-  EVP_SUBDOMAIN_HEADER_VALUE: 'llmobs-intake',
-  AGENTLESS_SPANS_ENDPOINT: '/api/v2/llmobs',
-  AGENTLESS_EVALULATIONS_ENDPOINT: '/api/intake/llm-obs/v1/eval-metric',
+
+  SPANS_EVENT_TYPE: 'span',
+  SPANS_INTAKE: 'llmobs-intake',
+  SPANS_ENDPOINT: '/api/v2/llmobs',
+
+  EVALUATIONS_INTAKE: 'api',
+  EVALUATIONS_EVENT_TYPE: 'evaluation_metric',
+  EVALUATIONS_ENDPOINT: '/api/intake/llm-obs/v1/eval-metric',
 
   EVP_PAYLOAD_SIZE_LIMIT: 5 << 20, // 5MB (actual limit is 5.1MB)
   EVP_EVENT_SIZE_LIMIT: (1 << 20) - 1024 // 999KB (actual limit is 1MB)

packages/dd-trace/src/llmobs/index.js

@@ -13,9 +13,9 @@ const evalMetricAppendCh = channel('llmobs:eval-metric:append')
 const flushCh = channel('llmobs:writers:flush')
 const injectCh = channel('dd-trace:span:inject')
 
-const LLMObsAgentlessSpanWriter = require('./writers/spans/agentless')
-const LLMObsAgentProxySpanWriter = require('./writers/spans/agentProxy')
 const LLMObsEvalMetricsWriter = require('./writers/evaluations')
+const LLMObsSpanWriter = require('./writers/spans')
+const { setAgentStrategy } = require('./writers/util')
 
 /**
  * Setting writers and processor globally when LLMObs is enabled
@@ -25,16 +25,22 @@ const LLMObsEvalMetricsWriter = require('./writers/evaluations')
  * if the tracer is `init`ed. But, in those cases, we don't want to start writers or subscribe
  * to channels.
  */
+
+/** @type {LLMObsSpanProcessor | null} */
 let spanProcessor
+
+/** @type {LLMObsSpanWriter | null} */
 let spanWriter
+
+/** @type {LLMObsEvalMetricsWriter | null} */
 let evalWriter
 
 function enable (config) {
   const startTime = performance.now()
   // create writers and eval writer append and flush channels
   // span writer append is handled by the span processor
   evalWriter = new LLMObsEvalMetricsWriter(config)
-  spanWriter = createSpanWriter(config)
+  spanWriter = new LLMObsSpanWriter(config)
 
   evalMetricAppendCh.subscribe(handleEvalMetricAppend)
   flushCh.subscribe(handleFlush)
@@ -46,7 +52,20 @@ function enable (config) {
 
   // distributed tracing for llmobs
   injectCh.subscribe(handleLLMObsParentIdInjection)
-  telemetry.recordLLMObsEnabled(startTime, config)
+
+  setAgentStrategy(config, useAgentless => {
+    if (useAgentless && !(config.apiKey && config.site)) {
+      throw new Error(
+        'Cannot send LLM Observability data without a running agent or without both a Datadog API key and site.\n' +
+        'Ensure these configurations are set before running your application.'
+      )
+    }
+
+    evalWriter?.setAgentless(useAgentless)
+    spanWriter?.setAgentless(useAgentless)
+
+    telemetry.recordLLMObsEnabled(startTime, config)
+  })
 }
 
 function disable () {
@@ -74,11 +93,6 @@ function handleLLMObsParentIdInjection ({ carrier }) {
   carrier['x-datadog-tags'] += `,${PROPAGATED_PARENT_ID_KEY}=${parentId}`
 }
 
-function createSpanWriter (config) {
-  const SpanWriter = config.llmobs.agentlessEnabled ? LLMObsAgentlessSpanWriter : LLMObsAgentProxySpanWriter
-  return new SpanWriter(config)
-}
-
 function handleFlush () {
   try {
     spanWriter.flush()

packages/dd-trace/src/llmobs/sdk.js

@@ -287,13 +287,6 @@ class LLMObs extends NoopLLMObs {
   submitEvaluation (llmobsSpanContext, options = {}) {
     if (!this.enabled) return
 
-    if (!this._config.apiKey) {
-      throw new Error(
-        'DD_API_KEY is required for sending evaluation metrics. Evaluation metric data will not be sent.\n' +
-        'Ensure this configuration is set before running your application.'
-      )
-    }
-
     const { traceId, spanId } = llmobsSpanContext
     if (!traceId || !spanId) {
       throw new Error(

packages/dd-trace/src/llmobs/writers/base.js

@@ -1,16 +1,22 @@
 'use strict'
 
 const request = require('../../exporters/common/request')
-const { URL, format } = require('url')
+const { URL, format } = require('node:url')
+const path = require('node:path')
 
 const logger = require('../../log')
 
 const { encodeUnicode } = require('../util')
 const telemetry = require('../telemetry')
 const log = require('../../log')
+const {
+  EVP_SUBDOMAIN_HEADER_NAME,
+  EVP_PROXY_AGENT_BASE_PATH
+} = require('../constants/writers')
+const { parseResponseAndLog } = require('./util')
 
 class BaseLLMObsWriter {
-  constructor ({ interval, timeout, endpoint, intake, eventType, protocol, port }) {
+  constructor ({ interval, timeout, eventType, config, endpoint, intake }) {
     this._interval = interval || 1000 // 1s
     this._timeout = timeout || 5000 // 5s
     this._eventType = eventType
@@ -19,28 +25,20 @@ class BaseLLMObsWriter {
     this._bufferLimit = 1000
     this._bufferSize = 0
 
-    this._url = new URL(format({
-      protocol: protocol || 'https:',
-      hostname: intake,
-      port: port || 443,
-      pathname: endpoint
-    }))
-
-    this._headers = {
-      'Content-Type': 'application/json'
-    }
+    this._config = config
+    this._endpoint = endpoint
+    this._intake = intake
 
     this._periodic = setInterval(() => {
       this.flush()
     }, this._interval).unref()
 
-    process.once('beforeExit', () => {
+    this._beforeExitHandler = () => {
       this.destroy()
-    })
+    }
+    process.once('beforeExit', this._beforeExitHandler)
 
     this._destroyed = false
-
-    logger.debug(`Started ${this.constructor.name} to ${this._url}`)
   }
 
   append (event, byteLength) {
@@ -55,7 +53,9 @@ class BaseLLMObsWriter {
   }
 
   flush () {
-    if (this._buffer.length === 0) {
+    const noAgentStrategy = this._agentless == null
+
+    if (this._buffer.length === 0 || noAgentStrategy) {
       return
     }
 
@@ -64,29 +64,12 @@ class BaseLLMObsWriter {
     this._bufferSize = 0
     const payload = this._encode(this.makePayload(events))
 
-    const options = {
-      headers: this._headers,
-      method: 'POST',
-      url: this._url,
-      timeout: this._timeout
-    }
-
     log.debug(`Encoded LLMObs payload: ${payload}`)
 
+    const options = this._getOptions()
+
     request(payload, options, (err, resp, code) => {
-      if (err) {
-        logger.error(
-          'Error sending %d LLMObs %s events to %s: %s', events.length, this._eventType, this._url, err.message, err
-        )
-        telemetry.recordDroppedPayload(events.length, this._eventType, 'request_error')
-      } else if (code >= 300) {
-        logger.error(
-          'Error sending %d LLMObs %s events to %s: %s', events.length, this._eventType, this._url, code
-        )
-        telemetry.recordDroppedPayload(events.length, this._eventType, 'http_error')
-      } else {
-        logger.debug(`Sent ${events.length} LLMObs ${this._eventType} events to ${this._url}`)
-      }
+      parseResponseAndLog(err, code, events.length, options.url.href, this._eventType)
     })
   }
 
@@ -96,12 +79,57 @@ class BaseLLMObsWriter {
     if (!this._destroyed) {
       logger.debug(`Stopping ${this.constructor.name}`)
       clearInterval(this._periodic)
-      process.removeListener('beforeExit', this.destroy)
+      process.removeListener('beforeExit', this._beforeExitHandler)
       this.flush()
       this._destroyed = true
     }
   }
 
+  setAgentless (agentless) {
+    this._agentless = agentless
+    this._url = this._getUrl()
+    logger.debug(`Configuring ${this.constructor.name} to ${this._url.href}`)
+  }
+
+  _getUrl () {
+    if (this._agentless) {
+      return new URL(format({
+        protocol: 'https:',
+        hostname: `${this._intake}.${this._config.site}`,
+        pathname: this._endpoint
+      }))
+    }
+
+    const { hostname, port } = this._config
+    const base = this._config.url || new URL(format({
+      protocol: 'http:',
+      hostname,
+      port
+    }))
+
+    const proxyPath = path.join(EVP_PROXY_AGENT_BASE_PATH, this._endpoint)
+    return new URL(proxyPath, base)
+  }
+
+  _getOptions () {
+    const options = {
+      headers: {
+        'Content-Type': 'application/json'
+      },
+      method: 'POST',
+      timeout: this._timeout,
+      url: this._url
+    }
+
+    if (this._agentless) {
+      options.headers['DD-API-KEY'] = this._config.apiKey || ''
+    } else {
+      options.headers[EVP_SUBDOMAIN_HEADER_NAME] = this._intake
+    }
+
+    return options
+  }
+
   _encode (payload) {
     return JSON.stringify(payload, (key, value) => {
       if (typeof value === 'string') {

packages/dd-trace/src/llmobs/writers/evaluations.js

@@ -1,17 +1,20 @@
 'use strict'
 
-const { AGENTLESS_EVALULATIONS_ENDPOINT } = require('../constants/writers')
+const {
+  EVALUATIONS_ENDPOINT,
+  EVALUATIONS_EVENT_TYPE,
+  EVALUATIONS_INTAKE
+} = require('../constants/writers')
 const BaseWriter = require('./base')
 
 class LLMObsEvalMetricsWriter extends BaseWriter {
   constructor (config) {
     super({
-      endpoint: AGENTLESS_EVALULATIONS_ENDPOINT,
-      intake: `api.${config.site}`,
-      eventType: 'evaluation_metric'
+      config,
+      intake: EVALUATIONS_INTAKE,
+      eventType: EVALUATIONS_EVENT_TYPE,
+      endpoint: EVALUATIONS_ENDPOINT
     })
-
-    this._headers['DD-API-KEY'] = config.apiKey
   }
 
   makePayload (events) {

packages/dd-trace/src/llmobs/writers/spans/base.js renamed to packages/dd-trace/src/llmobs/writers/spans.js

@@ -1,19 +1,27 @@
 'use strict'
 
-const { EVP_EVENT_SIZE_LIMIT, EVP_PAYLOAD_SIZE_LIMIT } = require('../../constants/writers')
-const { DROPPED_VALUE_TEXT } = require('../../constants/text')
-const { DROPPED_IO_COLLECTION_ERROR } = require('../../constants/tags')
-const BaseWriter = require('../base')
-const telemetry = require('../../telemetry')
-const logger = require('../../../log')
+const {
+  EVP_EVENT_SIZE_LIMIT,
+  EVP_PAYLOAD_SIZE_LIMIT,
+  SPANS_ENDPOINT,
+  SPANS_EVENT_TYPE,
+  SPANS_INTAKE
+} = require('../constants/writers')
+const { DROPPED_VALUE_TEXT } = require('../constants/text')
+const { DROPPED_IO_COLLECTION_ERROR } = require('../constants/tags')
+const BaseWriter = require('./base')
+const telemetry = require('../telemetry')
+const logger = require('../../log')
 
-const tracerVersion = require('../../../../../../package.json').version
+const tracerVersion = require('../../../../../package.json').version
 
 class LLMObsSpanWriter extends BaseWriter {
-  constructor (options) {
+  constructor (config) {
     super({
-      ...options,
-      eventType: 'span'
+      config,
+      eventType: SPANS_EVENT_TYPE,
+      intake: SPANS_INTAKE,
+      endpoint: SPANS_ENDPOINT
     })
   }
 
@@ -33,11 +41,11 @@ class LLMObsSpanWriter extends BaseWriter {
     telemetry.recordLLMObsSpanSize(event, processedEventSizeBytes, shouldTruncate)
 
     if (this._bufferSize + eventSizeBytes > EVP_PAYLOAD_SIZE_LIMIT) {
-      logger.debug('Flusing queue because queing next event will exceed EvP payload limit')
+      logger.debug('Flushing queue because queuing next event will exceed EvP payload limit')
       this.flush()
     }
 
-    super.append(event, eventSizeBytes)
+    super.append(event, processedEventSizeBytes)
   }
 
   makePayload (events) {