Add service name source to stats

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

@@ -10,6 +10,8 @@
 public final class MetricKey {
   private final UTF8BytesString resource;
   private final UTF8BytesString service;
+  // note: it won't participate to equals/hashcode since not part of aggregation
+  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;
@@ -101,6 +80,10 @@ public UTF8BytesString getService() {
     return service;
   }
 
+  public UTF8BytesString getServiceSource() {
+    return serviceSource;
+  }
+
   public UTF8BytesString getOperationName() {
     return operationName;
   }

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 = "ServiceSource".getBytes(ISO_8859_1);
 
   // Constant declared here for compile-time folding
   public static final int TRISTATE_TRUE = TriState.TRUE.serialValue;
@@ -109,7 +110,8 @@ 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 +147,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

@@ -1,5 +1,7 @@
 package datadog.trace.core;
 
+import static datadog.trace.api.DDTags.DD_SVC_SRC;
+
 import datadog.trace.api.DDTraceId;
 import java.util.Map;
 
@@ -9,6 +11,18 @@ public interface CoreSpan<T extends CoreSpan<T>> {
 
   String getServiceName();
 
+  default CharSequence getServiceNameSource() {
+    // overridden in DDSpan for a better implementation
+    final Object obj = getTag(DD_SVC_SRC);
+    if (obj == null) {
+      return null;
+    }
+    if (obj instanceof CharSequence) {
+      return (CharSequence) obj;
+    }
+    return obj.toString();
+  }
+
   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/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