Add service name source to trace stats (#10653)

Add service name source to stats

Add tests and move postprocessors later

Finalize implementation

Fix agent integration tests

Co-authored-by: devflow.devflow-routing-intake <devflow.devflow-routing-intake@kubernetes.us1.ddbuild.io>

Author: amarziali

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

@@ -331,6 +331,7 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel, CharSequence spanK
             span.getResourceName(),
             SERVICE_NAMES.computeIfAbsent(span.getServiceName(), UTF8_ENCODE),
             span.getOperationName(),
+            span.getServiceNameSource(),
             span.getType(),
             span.getHttpStatusCode(),
             isSynthetic(span),

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

@@ -5,11 +5,13 @@
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 
 /** The aggregation key for tracked metrics. */
 public final class MetricKey {
   private final UTF8BytesString resource;
   private final UTF8BytesString service;
+  private final UTF8BytesString serviceSource;
   private final UTF8BytesString operationName;
   private final UTF8BytesString type;
   private final int httpStatusCode;
@@ -21,35 +23,11 @@ public final class MetricKey {
   private final UTF8BytesString httpMethod;
   private final UTF8BytesString httpEndpoint;
 
-  // Constructor without httpMethod and httpEndpoint for backward compatibility
-  public MetricKey(
-      CharSequence resource,
-      CharSequence service,
-      CharSequence operationName,
-      CharSequence type,
-      int httpStatusCode,
-      boolean synthetics,
-      boolean isTraceRoot,
-      CharSequence spanKind,
-      List<UTF8BytesString> peerTags) {
-    this(
-        resource,
-        service,
-        operationName,
-        type,
-        httpStatusCode,
-        synthetics,
-        isTraceRoot,
-        spanKind,
-        peerTags,
-        null,
-        null);
-  }
-
   public MetricKey(
       CharSequence resource,
       CharSequence service,
       CharSequence operationName,
+      CharSequence serviceSource,
       CharSequence type,
       int httpStatusCode,
       boolean synthetics,
@@ -60,6 +38,7 @@ public MetricKey(
       CharSequence httpEndpoint) {
     this.resource = null == resource ? EMPTY : UTF8BytesString.create(resource);
     this.service = null == service ? EMPTY : UTF8BytesString.create(service);
+    this.serviceSource = null == serviceSource ? null : UTF8BytesString.create(serviceSource);
     this.operationName = null == operationName ? EMPTY : UTF8BytesString.create(operationName);
     this.type = null == type ? EMPTY : UTF8BytesString.create(type);
     this.httpStatusCode = httpStatusCode;
@@ -79,18 +58,34 @@ public MetricKey(
 
     // Only include httpMethod and httpEndpoint in hash if they are not null
     // This ensures backward compatibility when the feature is disabled
-    this.hash =
-        -196_513_505 * Boolean.hashCode(this.isTraceRoot)
-            + -1_807_454_463 * this.spanKind.hashCode()
-            + 887_503_681 * this.peerTags.hashCode()
-            + (this.httpMethod != null ? 28_629_151 * this.httpMethod.hashCode() : 0)
-            + (this.httpEndpoint != null ? 923_521 * this.httpEndpoint.hashCode() : 0)
-            + 29_791 * this.resource.hashCode()
-            + 961 * this.service.hashCode()
-            + 31 * this.operationName.hashCode()
-            + this.type.hashCode()
-            + 31 * httpStatusCode
-            + (this.synthetics ? 1 : 0);
+
+    // Note: all the multiplication got constant folded at compile time (see javap -verbose ...)
+    int tmpHash =
+        (int) (31L * 31 * 31 * 31 * 31 * 31 * 31 * 31) * Boolean.hashCode(this.isTraceRoot) // 8
+            + (int) (31L * 31 * 31 * 31 * 31 * 31 * 31) * this.spanKind.hashCode() // 7
+            + 31 * 31 * 31 * 31 * 31 * 31 * this.peerTags.hashCode() // 6
+            + 31 * 31 * 31 * 31 * 31 * this.resource.hashCode() // 5
+            + 31 * 31 * 31 * 31 * this.service.hashCode() // 4
+            + 31 * 31 * 31 * this.operationName.hashCode() // 3
+            + 31 * 31 * this.type.hashCode() // 2
+            + 31 * this.httpStatusCode // 1
+            + (this.synthetics ? 1 : 0); // 0
+    // optional fields
+    if (this.serviceSource != null) {
+      tmpHash +=
+          (int) (31L * 31 * 31 * 31 * 31 * 31 * 31 * 31 * 31) * this.serviceSource.hashCode(); // 9
+    }
+    if (this.httpEndpoint != null) {
+      tmpHash +=
+          (int) (31L * 31 * 31 * 31 * 31 * 31 * 31 * 31 * 31 * 31)
+              * this.httpEndpoint.hashCode(); // 10
+    }
+    if (this.httpMethod != null) {
+      tmpHash +=
+          (int) (31L * 31 * 31 * 31 * 31 * 31 * 31 * 31 * 31 * 31 * 31)
+              * this.httpMethod.hashCode(); // 11
+    }
+    this.hash = tmpHash;
   }
 
   public UTF8BytesString getResource() {
@@ -101,6 +96,10 @@ public UTF8BytesString getService() {
     return service;
   }
 
+  public UTF8BytesString getServiceSource() {
+    return serviceSource;
+  }
+
   public UTF8BytesString getOperationName() {
     return operationName;
   }
@@ -144,29 +143,19 @@ public boolean equals(Object o) {
     }
     if ((o instanceof MetricKey)) {
       MetricKey metricKey = (MetricKey) o;
-      boolean basicEquals =
-          hash == metricKey.hash
-              && synthetics == metricKey.synthetics
-              && httpStatusCode == metricKey.httpStatusCode
-              && resource.equals(metricKey.resource)
-              && service.equals(metricKey.service)
-              && operationName.equals(metricKey.operationName)
-              && type.equals(metricKey.type)
-              && isTraceRoot == metricKey.isTraceRoot
-              && spanKind.equals(metricKey.spanKind)
-              && peerTags.equals(metricKey.peerTags);
-
-      // Only compare httpMethod and httpEndpoint if at least one of them is not null
-      // This ensures backward compatibility when the feature is disabled
-      boolean thisHasEndpoint = httpMethod != null || httpEndpoint != null;
-      boolean otherHasEndpoint = metricKey.httpMethod != null || metricKey.httpEndpoint != null;
-
-      if (thisHasEndpoint || otherHasEndpoint) {
-        return basicEquals
-            && java.util.Objects.equals(httpMethod, metricKey.httpMethod)
-            && java.util.Objects.equals(httpEndpoint, metricKey.httpEndpoint);
-      }
-      return basicEquals;
+      return hash == metricKey.hash
+          && synthetics == metricKey.synthetics
+          && httpStatusCode == metricKey.httpStatusCode
+          && resource.equals(metricKey.resource)
+          && service.equals(metricKey.service)
+          && operationName.equals(metricKey.operationName)
+          && type.equals(metricKey.type)
+          && isTraceRoot == metricKey.isTraceRoot
+          && spanKind.equals(metricKey.spanKind)
+          && peerTags.equals(metricKey.peerTags)
+          && Objects.equals(serviceSource, metricKey.serviceSource)
+          && Objects.equals(httpMethod, metricKey.httpMethod)
+          && Objects.equals(httpEndpoint, metricKey.httpEndpoint);
     }
     return false;
   }

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

@@ -37,6 +37,7 @@ public final class SerializingMetricWriter implements MetricWriter {
   private static final byte[] PEER_TAGS = "PeerTags".getBytes(ISO_8859_1);
   private static final byte[] HTTP_METHOD = "HTTPMethod".getBytes(ISO_8859_1);
   private static final byte[] HTTP_ENDPOINT = "HTTPEndpoint".getBytes(ISO_8859_1);
+  private static final byte[] SERVICE_SOURCE = "srv_src".getBytes(ISO_8859_1);
 
   // Constant declared here for compile-time folding
   public static final int TRISTATE_TRUE = TriState.TRUE.serialValue;
@@ -109,7 +110,9 @@ public void add(MetricKey key, AggregateMetric aggregate) {
     // Calculate dynamic map size based on optional fields
     final boolean hasHttpMethod = key.getHttpMethod() != null;
     final boolean hasHttpEndpoint = key.getHttpEndpoint() != null;
-    final int mapSize = 15 + (hasHttpMethod ? 1 : 0) + (hasHttpEndpoint ? 1 : 0);
+    final boolean hasServiceSource = key.getServiceSource() != null;
+    final int mapSize =
+        15 + (hasServiceSource ? 1 : 0) + (hasHttpMethod ? 1 : 0) + (hasHttpEndpoint ? 1 : 0);
 
     writer.startMap(mapSize);
 
@@ -145,6 +148,10 @@ public void add(MetricKey key, AggregateMetric aggregate) {
       writer.writeUTF8(peerTag);
     }
 
+    if (hasServiceSource) {
+      writer.writeUTF8(SERVICE_SOURCE);
+      writer.writeUTF8(key.getServiceSource());
+    }
     // Only include HTTPMethod if present
     if (hasHttpMethod) {
       writer.writeUTF8(HTTP_METHOD);

dd-trace-core/src/main/java/datadog/trace/core/CoreSpan.java

@@ -9,6 +9,10 @@ public interface CoreSpan<T extends CoreSpan<T>> {
 
   String getServiceName();
 
+  default CharSequence getServiceNameSource() {
+    return null;
+  }
+
   CharSequence getOperationName();
 
   CharSequence getResourceName();

dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java

@@ -590,6 +590,11 @@ public void setServiceName(@Nonnull String serviceName, @Nonnull CharSequence so
     context.setServiceName(serviceName, source);
   }
 
+  @Override
+  public CharSequence getServiceNameSource() {
+    return context.getServiceNameSource();
+  }
+
   @Override
   public final DDSpan setResourceName(final CharSequence resourceName) {
     return setResourceName(resourceName, ResourceNamePriorities.DEFAULT);

dd-trace-core/src/main/java/datadog/trace/core/tagprocessor/TagsPostProcessorFactory.java

@@ -13,7 +13,7 @@ private static class Lazy {
     private static TagsPostProcessor lazyProcessor = createLazyChain();
 
     private static TagsPostProcessor createEagerChain() {
-      final List<TagsPostProcessor> processors = new ArrayList<>(4);
+      final List<TagsPostProcessor> processors = new ArrayList<>(3);
       processors.add(new PeerServiceCalculator());
       if (addBaseService) {
         processors.add(new BaseServiceAdder(Config.get().getServiceName()));
@@ -23,12 +23,11 @@ private static TagsPostProcessor createEagerChain() {
       if (Config.get().isTraceResourceRenamingEnabled()) {
         processors.add(new HttpEndpointPostProcessor());
       }
-      processors.add(new ServiceNameSourceAdder()); // eager since needed for stats
       return new PostProcessorChain(processors.toArray(new TagsPostProcessor[0]));
     }
 
     private static TagsPostProcessor createLazyChain() {
-      final List<TagsPostProcessor> processors = new ArrayList<>(7);
+      final List<TagsPostProcessor> processors = new ArrayList<>(8);
 
       processors.add(new QueryObfuscator(Config.get().getObfuscationQueryRegexp()));
       if (addRemoteHostname) {
@@ -48,6 +47,7 @@ private static TagsPostProcessor createLazyChain() {
         processors.add(new SpanPointersProcessor());
       }
       processors.add(new IntegrationAdder());
+      processors.add(new ServiceNameSourceAdder());
       return new PostProcessorChain(
           processors.toArray(processors.toArray(new TagsPostProcessor[0])));
     }

dd-trace-core/src/test/groovy/datadog/trace/common/metrics/AggregateMetricTest.groovy

@@ -65,7 +65,7 @@ class AggregateMetricTest extends DDSpecification {
     given:
     AggregateMetric aggregate = new AggregateMetric().recordDurations(3, new AtomicLongArray(0L, 0L, 0L | ERROR_TAG | TOP_LEVEL_TAG))
 
-    Batch batch = new Batch().reset(new MetricKey("foo", "bar", "qux", "type", 0, false, true, "corge", [UTF8BytesString.create("grault:quux")], null, null))
+    Batch batch = new Batch().reset(new MetricKey("foo", "bar", "qux", null, "type", 0, false, true, "corge", [UTF8BytesString.create("grault:quux")], null, null))
     batch.add(0L, 10)
     batch.add(0L, 10)
     batch.add(0L, 10)
@@ -140,7 +140,7 @@ class AggregateMetricTest extends DDSpecification {
   def "consistent under concurrent attempts to read and write"() {
     given:
     AggregateMetric aggregate = new AggregateMetric()
-    MetricKey key = new MetricKey("foo", "bar", "qux", "type", 0, false, true, "corge", [UTF8BytesString.create("grault:quux")], null, null)
+    MetricKey key = new MetricKey("foo", "bar", "qux", null, "type", 0, false, true, "corge", [UTF8BytesString.create("grault:quux")], null, null)
     BlockingDeque<Batch> queue = new LinkedBlockingDeque<>(1000)
     ExecutorService reader = Executors.newSingleThreadExecutor()
     int writerCount = 10