fix(llmobs): aggregate span.finished telemetry counts per interval

dd-java-agent/agent-llmobs/src/test/groovy/datadog/trace/llmobs/domain/DDLLMObsSpanTest.groovy

@@ -46,6 +46,7 @@ class DDLLMObsSpanTest  extends DDSpecification{
   void setup() {
     assert TEST_TRACER.activeSpan() == null: "Span is active before test has started: " + TEST_TRACER.activeSpan()
     TEST_TRACER.flush()
+    LLMObsMetricCollector.get().resetForTesting()
   }
 
   void cleanup() {
@@ -626,10 +627,10 @@ class DDLLMObsSpanTest  extends DDSpecification{
   def "finish records span.finished telemetry when LLMObs enabled"() {
     setup:
     LLMObsMetricCollector collector = LLMObsMetricCollector.get()
-    collector.drain()
 
     when:
     llmObsSpan(Tags.LLMOBS_WORKFLOW_SPAN_KIND, "workflow-span").finish()
+    collector.prepareMetrics()
 
     then:
     def metrics = collector.drain()
@@ -650,12 +651,12 @@ class DDLLMObsSpanTest  extends DDSpecification{
   def "finish records span.finished telemetry for non-root span when LLMObs enabled"() {
     setup:
     LLMObsMetricCollector collector = LLMObsMetricCollector.get()
-    collector.drain()
 
     when:
     runUnderTrace("parent") {
       llmObsSpan(Tags.LLMOBS_LLM_SPAN_KIND, "child-llm").finish()
     }
+    collector.prepareMetrics()
 
     then:
     def metrics = collector.drain()
@@ -676,10 +677,10 @@ class DDLLMObsSpanTest  extends DDSpecification{
   def "span has expected session tag and telemetry has #expectedHasSessionIdTag"() {
     setup:
     LLMObsMetricCollector collector = LLMObsMetricCollector.get()
-    collector.drain()
 
     when:
     llmObsSpan(Tags.LLMOBS_WORKFLOW_SPAN_KIND, "workflow-span", sessionId).finish()
+    collector.prepareMetrics()
 
     then:
     def metrics = collector.drain()

internal-api/src/main/java/datadog/trace/api/telemetry/LLMObsMetricCollector.java

@@ -2,17 +2,27 @@
 
 import datadog.trace.api.cache.DDCache;
 import datadog.trace.api.cache.DDCaches;
+import datadog.trace.api.internal.VisibleForTesting;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.LongAdder;
 import javax.annotation.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * Collects telemetry metrics for LLM Observability spans.
+ *
+ * <p>Counts are aggregated per tag combination in-process and emitted as one point per metrics
+ * interval.
+ */
 public final class LLMObsMetricCollector
     implements MetricCollector<LLMObsMetricCollector.LLMObsMetric> {
   private static final String METRIC_NAMESPACE = "mlobs";
@@ -41,19 +51,38 @@ public static LLMObsMetricCollector get() {
   private static final String HAS_SESSION_ID_TRUE = "has_session_id:1";
   private static final String HAS_SESSION_ID_FALSE = "has_session_id:0";
 
+  /**
+   * Upper bound on the number of distinct tag combinations tracked. Tag values are drawn from
+   * bounded sets (integrations, span kinds, and four booleans), so legitimate cardinality is in the
+   * low hundreds at the 8-integration scale the tag caches are sized for; this only guards against
+   * an unexpected high-cardinality source. Entries are never removed (see {@link
+   * #prepareMetrics()}), so this is a lifetime ceiling, not a concurrent one.
+   */
+  static final int MAX_TAG_COMBINATIONS = 512;
+
   private final BlockingQueue<LLMObsMetric> metricsQueue;
   private final DDCache<String, String> integrationTagCache;
   private final DDCache<String, String> spanKindTagCache;
 
+  /**
+   * Counter per tag combination, aggregated in-process and flushed once per metrics interval by
+   * {@link #prepareMetrics()}.
+   */
+  private final ConcurrentHashMap<List<String>, LongAdder> spanFinishedCounters;
+
   private LLMObsMetricCollector() {
     this.metricsQueue = new ArrayBlockingQueue<>(RAW_QUEUE_SIZE);
     this.integrationTagCache = DDCaches.newFixedSizeCache(8);
     this.spanKindTagCache = DDCaches.newFixedSizeCache(8);
+    this.spanFinishedCounters = new ConcurrentHashMap<>();
   }
 
   /**
    * Record a span finished metric for LLMObs telemetry.
    *
+   * <p>This only increments an in-process counter. The counter is converted into a single telemetry
+   * metric per tag combination by {@link #prepareMetrics()}, once per metrics interval.
+   *
    * @param integration the integration name (e.g., "openai")
    * @param spanKind the span kind (e.g., "llm", "embedding")
    * @param isRootSpan whether this is a root span
@@ -80,11 +109,23 @@ public void recordSpanFinished(
             isAutoInstrumented ? AUTOINSTRUMENTED_TRUE : AUTOINSTRUMENTED_FALSE,
             hasError ? ERROR_TRUE : ERROR_FALSE,
             hasSessionId ? HAS_SESSION_ID_TRUE : HAS_SESSION_ID_FALSE);
-    LLMObsMetric metric =
-        new LLMObsMetric(METRIC_NAMESPACE, true, SPAN_FINISHED_METRIC, COUNT_METRIC_TYPE, 1L, tags);
-    if (!metricsQueue.offer(metric)) {
-      log.debug("Unable to add telemetry metric {} for {}", SPAN_FINISHED_METRIC, integration);
+
+    LongAdder counter = spanFinishedCounters.get(tags);
+    if (counter == null) {
+      // Soft bound: concurrent recorders may overshoot slightly, which is fine for a guard.
+      if (spanFinishedCounters.size() >= MAX_TAG_COMBINATIONS) {
+        if (log.isDebugEnabled()) {
+          log.debug(
+              "Dropping telemetry metric {} for {}: tag combination limit ({}) reached",
+              SPAN_FINISHED_METRIC,
+              integration,
+              MAX_TAG_COMBINATIONS);
+        }
+        return;
+      }
+      counter = spanFinishedCounters.computeIfAbsent(tags, key -> new LongAdder());
     }
+    counter.increment();
   }
 
   /**
@@ -140,7 +181,30 @@ public void recordFeedbackSubmitted(
 
   @Override
   public void prepareMetrics() {
-    // metrics are added directly via recordSpanFinished; no additional preparation needed
+    // Entries are never removed: a recorder thread may already hold a reference to a LongAdder, so
+    // removing it here would silently drop a concurrent increment. Tag values come from bounded
+    // sets, so retaining idle combinations costs at most MAX_TAG_COMBINATIONS entries.
+    for (Map.Entry<List<String>, LongAdder> entry : spanFinishedCounters.entrySet()) {
+      long value = entry.getValue().sumThenReset();
+      if (value == 0) {
+        continue;
+      }
+      LLMObsMetric metric =
+          new LLMObsMetric(
+              METRIC_NAMESPACE,
+              true,
+              SPAN_FINISHED_METRIC,
+              COUNT_METRIC_TYPE,
+              value,
+              entry.getKey());
+      if (!metricsQueue.offer(metric)) {
+        // Queue is full; give the count back to the counter so it is reported in a later interval
+        // instead of being lost, and stop staging for now.
+        entry.getValue().add(value);
+        log.debug("Unable to add telemetry metric {}: queue is full", SPAN_FINISHED_METRIC);
+        break;
+      }
+    }
   }
 
   @Override
@@ -153,6 +217,13 @@ public Collection<LLMObsMetric> drain() {
     return drained;
   }
 
+  /** Clears all staged counters and metrics. Visible for testing only. */
+  @VisibleForTesting
+  public void resetForTesting() {
+    spanFinishedCounters.clear();
+    metricsQueue.clear();
+  }
+
   public static class LLMObsMetric extends MetricCollector.Metric {
     public LLMObsMetric(
         String namespace,