Add span-derived primary tags (CSS v1.3.0)

communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java

@@ -139,8 +139,7 @@ protected long getFeaturesDiscoveryMinDelayMillis() {
 
   private synchronized void discoverIfOutdated(final long maxElapsedMs) {
     final long now = System.currentTimeMillis();
-    final State previous = discoveryState;
-    final long elapsed = now - previous.lastTimeDiscovered;
+    final long elapsed = now - discoveryState.lastTimeDiscovered;
     if (elapsed > maxElapsedMs) {
       final State newState = new State();
       doDiscovery(newState);

dd-trace-api/src/main/java/datadog/trace/api/config/GeneralConfig.java

@@ -78,6 +78,7 @@ public final class GeneralConfig {
       "trace.tracer.metrics.ignored.resources";
   public static final String TRACE_STATS_CARDINALITY_LIMITS_ENABLED =
       "trace.stats.cardinality.limits.enabled";
+  public static final String TRACE_STATS_ADDITIONAL_TAGS = "trace.stats.additional.tags";
 
   public static final String AZURE_APP_SERVICES = "azure.app.services";
   public static final String INTERNAL_EXIT_ON_FAILURE = "trace.internal.exit.on.failure";

dd-trace-core/src/jmh/java/datadog/trace/common/metrics/AdditionalTagsMetricsBenchmark.java

@@ -0,0 +1,103 @@
+package datadog.trace.common.metrics;
+
+import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND;
+import static datadog.trace.bootstrap.instrumentation.api.Tags.SPAN_KIND_CLIENT;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import datadog.trace.api.WellKnownTags;
+import datadog.trace.core.CoreSpan;
+import de.thetaphi.forbiddenapis.SuppressForbidden;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+
+/**
+ * Regression benchmark for the additional-tags hot path; {@code limitsEnabled=true} is slower here
+ * because it records more (sentinel collapses), not because limiting is expensive.
+ */
+@State(Scope.Benchmark)
+@Warmup(iterations = 2, time = 15, timeUnit = SECONDS)
+@Measurement(iterations = 5, time = 15, timeUnit = SECONDS)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(SECONDS)
+@Threads(8)
+@Fork(value = 1)
+public class AdditionalTagsMetricsBenchmark {
+
+  private ClientStatsAggregator aggregator;
+  private AdversarialMetricsBenchmark.CountingHealthMetrics health;
+
+  @Param({"false", "true"})
+  public boolean limitsEnabled;
+
+  @State(Scope.Thread)
+  public static class ThreadState {
+    int cursor;
+  }
+
+  @Setup
+  public void setup() {
+    this.health = new AdversarialMetricsBenchmark.CountingHealthMetrics();
+    AdditionalTagsSchema additionalTagsSchema =
+        AdditionalTagsSchema.from(
+            new LinkedHashSet<>(Arrays.asList("region", "tenant_id")), limitsEnabled, this.health);
+    this.aggregator =
+        new ClientStatsAggregator(
+            new WellKnownTags("", "", "", "", "", ""),
+            Collections.emptySet(),
+            additionalTagsSchema,
+            new ClientStatsAggregatorBenchmark.FixedAgentFeaturesDiscovery(
+                Collections.singleton("peer.hostname"), Collections.emptySet()),
+            this.health,
+            new ClientStatsAggregatorBenchmark.NullSink(),
+            2048,
+            2048,
+            false);
+    this.aggregator.start();
+  }
+
+  @TearDown
+  @SuppressForbidden
+  public void tearDown() {
+    aggregator.close();
+    System.err.println("[ADDITIONAL-TAGS] counters (across all threads, single fork):");
+    System.err.println("  onStatsInboxFull         = " + health.inboxFull.sum());
+    System.err.println("  onStatsAggregateDropped  = " + health.aggregateDropped.sum());
+  }
+
+  @Benchmark
+  public void publish(ThreadState ts, Blackhole blackhole) {
+    int idx = ts.cursor++;
+    ThreadLocalRandom rng = ThreadLocalRandom.current();
+
+    int scrambled = idx * 0x9E3779B1;
+    String region = "region-" + ((scrambled >>> 4) & 0xFFFF);
+    String tenant = "tenant-" + ((scrambled >>> 16) & 0xFFFF);
+    long durationNanos = 1L + (rng.nextLong() & 0x3FFFFFFFL);
+
+    SimpleSpan span =
+        new SimpleSpan("svc", "op", "res", "web", true, true, false, 0, durationNanos, 200);
+    span.setTag(SPAN_KIND, SPAN_KIND_CLIENT);
+    span.setTag("region", region);
+    span.setTag("tenant_id", tenant);
+
+    List<CoreSpan<?>> trace = Collections.singletonList(span);
+    blackhole.consume(aggregator.publish(trace));
+  }
+}

dd-trace-core/src/jmh/java/datadog/trace/common/metrics/AdversarialMetricsBenchmark.java

@@ -75,6 +75,7 @@ public void setup() {
         new ClientStatsAggregator(
             new WellKnownTags("", "", "", "", "", ""),
             Collections.emptySet(),
+            AdditionalTagsSchema.EMPTY,
             new ClientStatsAggregatorBenchmark.FixedAgentFeaturesDiscovery(
                 Collections.singleton("peer.hostname"), Collections.emptySet()),
             this.health,

dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorBenchmark.java

@@ -42,6 +42,7 @@ public class ClientStatsAggregatorBenchmark {
       new ClientStatsAggregator(
           new WellKnownTags("", "", "", "", "", ""),
           Collections.emptySet(),
+          AdditionalTagsSchema.EMPTY,
           featuresDiscovery,
           HealthMetrics.NO_OP,
           new NullSink(),

dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorDDSpanBenchmark.java

@@ -62,6 +62,7 @@ public class ClientStatsAggregatorDDSpanBenchmark {
       new ClientStatsAggregator(
           new WellKnownTags("", "", "", "", "", ""),
           Collections.emptySet(),
+          AdditionalTagsSchema.EMPTY,
           featuresDiscovery,
           HealthMetrics.NO_OP,
           new ClientStatsAggregatorBenchmark.NullSink(),

dd-trace-core/src/jmh/java/datadog/trace/common/metrics/HighCardinalityPeerMetricsBenchmark.java

@@ -65,6 +65,7 @@ public void setup() {
         new ClientStatsAggregator(
             new WellKnownTags("", "", "", "", "", ""),
             Collections.emptySet(),
+            AdditionalTagsSchema.EMPTY,
             new ClientStatsAggregatorBenchmark.FixedAgentFeaturesDiscovery(
                 Collections.singleton("peer.hostname"), Collections.emptySet()),
             this.health,

dd-trace-core/src/jmh/java/datadog/trace/common/metrics/HighCardinalityResourceMetricsBenchmark.java

@@ -61,6 +61,7 @@ public void setup() {
         new ClientStatsAggregator(
             new WellKnownTags("", "", "", "", "", ""),
             Collections.emptySet(),
+            AdditionalTagsSchema.EMPTY,
             new ClientStatsAggregatorBenchmark.FixedAgentFeaturesDiscovery(
                 Collections.singleton("peer.hostname"), Collections.emptySet()),
             this.health,

dd-trace-core/src/main/java/datadog/trace/common/metrics/AdditionalTagsSchema.java

@@ -0,0 +1,118 @@
+package datadog.trace.common.metrics;
+
+import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
+import datadog.trace.core.monitor.HealthMetrics;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Immutable schema of configured span-derived primary tag keys; built once at aggregator
+ * construction.
+ */
+final class AdditionalTagsSchema {
+
+  private static final Logger log = LoggerFactory.getLogger(AdditionalTagsSchema.class);
+
+  // Backend pipeline supports ~4 primary tag dimensions by default; drop overflow at startup.
+  static final int MAX_ADDITIONAL_TAG_KEYS = 10;
+
+  /** Singleton empty schema returned when no additional tags are configured. */
+  static final AdditionalTagsSchema EMPTY =
+      new AdditionalTagsSchema(new String[0], new TagCardinalityHandler[0], HealthMetrics.NO_OP);
+
+  final String[] names;
+
+  /** Per-key handlers providing UTF8 caching and per-cycle cardinality limiting. */
+  private final TagCardinalityHandler[] handlers;
+
+  private final HealthMetrics healthMetrics;
+
+  private AdditionalTagsSchema(
+      String[] names, TagCardinalityHandler[] handlers, HealthMetrics healthMetrics) {
+    this.names = names;
+    this.handlers = handlers;
+    this.healthMetrics = healthMetrics;
+  }
+
+  /** Test convenience: uses {@link HealthMetrics#NO_OP} and limits enabled. */
+  static AdditionalTagsSchema from(Set<String> configured) {
+    return from(configured, true, HealthMetrics.NO_OP);
+  }
+
+  static AdditionalTagsSchema from(
+      Set<String> configured, boolean useBlockedSentinel, HealthMetrics healthMetrics) {
+    if (configured == null || configured.isEmpty()) {
+      return EMPTY;
+    }
+    List<String> valid = new ArrayList<>();
+    for (String key : configured) {
+      if (key == null || key.isEmpty()) {
+        log.warn("Ignoring empty additional metric tag key");
+        continue;
+      }
+      if (key.contains(":")) {
+        log.warn("Ignoring additional metric tag key '{}': keys must not contain ':'", key);
+        continue;
+      }
+      valid.add(key);
+    }
+    if (valid.isEmpty()) {
+      return EMPTY;
+    }
+    Collections.sort(valid);
+    // Dedup (sort brings duplicates adjacent)
+    List<String> deduped = new ArrayList<>(valid.size());
+    String prev = null;
+    for (String key : valid) {
+      if (!key.equals(prev)) {
+        deduped.add(key);
+        prev = key;
+      }
+    }
+    if (deduped.size() > MAX_ADDITIONAL_TAG_KEYS) {
+      log.warn(
+          "Configured additional metric tag keys ({}) exceeds the supported limit of {}; "
+              + "dropping extra keys: {}",
+          deduped.size(),
+          MAX_ADDITIONAL_TAG_KEYS,
+          deduped.subList(MAX_ADDITIONAL_TAG_KEYS, deduped.size()));
+      deduped = deduped.subList(0, MAX_ADDITIONAL_TAG_KEYS);
+    }
+    String[] namesArr = deduped.toArray(new String[0]);
+    TagCardinalityHandler[] handlersArr = new TagCardinalityHandler[namesArr.length];
+    for (int i = 0; i < namesArr.length; i++) {
+      handlersArr[i] =
+          new TagCardinalityHandler(
+              namesArr[i], MetricCardinalityLimits.ADDITIONAL_TAG_VALUE, useBlockedSentinel);
+    }
+    return new AdditionalTagsSchema(namesArr, handlersArr, healthMetrics);
+  }
+
+  int size() {
+    return names.length;
+  }
+
+  String name(int i) {
+    return names[i];
+  }
+
+  UTF8BytesString register(int i, String value) {
+    return handlers[i].register(value);
+  }
+
+  void resetHandlers() {
+    for (int i = 0; i < handlers.length; i++) {
+      long blocked = handlers[i].reset();
+      if (blocked > 0) {
+        log.warn(
+            "Cardinality limit reached for additional metric tag '{}'; further values will be reported as blocked_by_tracer",
+            names[i]);
+        healthMetrics.onTagCardinalityBlocked(handlers[i].statsDTag(), blocked);
+      }
+    }
+  }
+}