Add tests and move postprocessors later

Author: amarziali

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

@@ -111,7 +111,8 @@ public void add(MetricKey key, AggregateMetric aggregate) {
     final boolean hasHttpMethod = key.getHttpMethod() != null;
     final boolean hasHttpEndpoint = key.getHttpEndpoint() != null;
     final boolean hasServiceSource = key.getServiceSource() != null;
-    final int mapSize = 15 + (hasServiceSource ? 1 : 0) + (hasHttpMethod ? 1 : 0) + (hasHttpEndpoint ? 1 : 0);
+    final int mapSize =
+        15 + (hasServiceSource ? 1 : 0) + (hasHttpMethod ? 1 : 0) + (hasHttpEndpoint ? 1 : 0);
 
     writer.startMap(mapSize);
 

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

@@ -1,7 +1,5 @@
 package datadog.trace.core;
 
-import static datadog.trace.api.DDTags.DD_SVC_SRC;
-
 import datadog.trace.api.DDTraceId;
 import java.util.Map;
 
@@ -12,15 +10,7 @@ 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();
+    return null;
   }
 
   CharSequence getOperationName();

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/ConflatingMetricAggregatorTest.groovy

@@ -467,7 +467,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       "resource2",
       "service2",
       "operation2",
-        null,
+      null,
       "type",
       HTTP_OK,
       false,
@@ -789,6 +789,52 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
     aggregator.close()
   }
 
+  def "gather the service name source when the span is published"() {
+    setup:
+    MetricWriter writer = Mock(MetricWriter)
+    Sink sink = Stub(Sink)
+    DDAgentFeaturesDiscovery features = Mock(DDAgentFeaturesDiscovery)
+    features.supportsMetrics() >> true
+    features.peerTags() >> []
+    ConflatingMetricsAggregator aggregator = new ConflatingMetricsAggregator(empty,
+      features, HealthMetrics.NO_OP, sink, writer, 10, queueSize, reportingInterval, SECONDS, true)
+    aggregator.start()
+
+    when: "publish span with a service name source"
+    CountDownLatch latch = new CountDownLatch(1)
+    long duration = 100
+    aggregator.publish([
+      new SimpleSpan("service", "operation", "resource", "type", true, true, false, 0, duration, 200, false, 0, "source")
+      .setTag(SPAN_KIND, "server")
+    ])
+    aggregator.report()
+    def latchTriggered = latch.await(2, SECONDS)
+
+    then: "should create the same metric keys for spans with and without s"
+    latchTriggered
+    1 * writer.startBucket(1, _, _)
+    1 * writer.add(new MetricKey(
+      "resource",
+      "service",
+      "operation",
+      "source",
+      "type",
+      200,
+      false,
+      false,
+      "server",
+      [],
+      null,
+      null
+      ), { AggregateMetric value ->
+        value.getHitCount() == 1 && value.getDuration() == duration
+      })
+    1 * writer.finishBucket() >> { latch.countDown() }
+
+    cleanup:
+    aggregator.close()
+  }
+
   def "test least recently written to aggregate flushed when size limit exceeded"() {
     setup:
     int maxAggregates = 10
@@ -821,7 +867,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
         "resource",
         "service" + i,
         "operation",
-          null,
+        null,
         "type",
         HTTP_OK,
         false,
@@ -838,7 +884,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       "resource",
       "service0",
       "operation",
-        null,
+      null,
       "type",
       HTTP_OK,
       false,
@@ -886,7 +932,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
         "resource",
         "service" + i,
         "operation",
-          null,
+        null,
         "type",
         HTTP_OK,
         false,
@@ -920,7 +966,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
         "resource",
         "service" + i,
         "operation",
-          null,
+        null,
         "type",
         HTTP_OK,
         false,
@@ -937,7 +983,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       "resource",
       "service0",
       "operation",
-        null,
+      null,
       "type",
       HTTP_OK,
       false,
@@ -985,7 +1031,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
         "resource",
         "service" + i,
         "operation",
-          null,
+        null,
         "type",
         HTTP_OK,
         false,
@@ -1043,7 +1089,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
         "resource",
         "service" + i,
         "operation",
-          null,
+        null,
         "type",
         HTTP_OK,
         false,

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

@@ -1,22 +1,21 @@
 package datadog.trace.common.metrics
 
+import static datadog.trace.api.config.GeneralConfig.EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED
+import static java.util.concurrent.TimeUnit.MILLISECONDS
+import static java.util.concurrent.TimeUnit.SECONDS
+
 import datadog.metrics.api.Histograms
 import datadog.metrics.impl.DDSketchHistograms
 import datadog.trace.api.Config
+import datadog.trace.api.Pair
 import datadog.trace.api.ProcessTags
 import datadog.trace.api.WellKnownTags
-import datadog.trace.api.Pair
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString
 import datadog.trace.test.util.DDSpecification
-import org.msgpack.core.MessagePack
-import org.msgpack.core.MessageUnpacker
-
 import java.nio.ByteBuffer
 import java.util.concurrent.atomic.AtomicLongArray
-
-import static datadog.trace.api.config.GeneralConfig.EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED
-import static java.util.concurrent.TimeUnit.MILLISECONDS
-import static java.util.concurrent.TimeUnit.SECONDS
+import org.msgpack.core.MessagePack
+import org.msgpack.core.MessageUnpacker
 
 class SerializingMetricWriterTest extends DDSpecification {
 
@@ -55,7 +54,7 @@ class SerializingMetricWriterTest extends DDSpecification {
         "resource1",
         "service1",
         "operation1",
-            null,
+        null,
         "type",
         0,
         false,
@@ -76,7 +75,7 @@ class SerializingMetricWriterTest extends DDSpecification {
         "resource2",
         "service2",
         "operation2",
-            null,
+        null,
         "type2",
         200,
         true,
@@ -97,7 +96,7 @@ class SerializingMetricWriterTest extends DDSpecification {
         "GET /api/users/:id",
         "web-service",
         "http.request",
-            null,
+        null,
         "web",
         200,
         false,
@@ -116,7 +115,7 @@ class SerializingMetricWriterTest extends DDSpecification {
           "resource" + i,
           "service" + i,
           "operation" + i,
-              null,
+          null,
           "type",
           0,
           false,
@@ -280,6 +279,35 @@ class SerializingMetricWriterTest extends DDSpecification {
     }
   }
 
+  def "ServiceSource optional in the payload"() {
+    setup:
+    long startTime = MILLISECONDS.toNanos(System.currentTimeMillis())
+    long duration = SECONDS.toNanos(10)
+    WellKnownTags wellKnownTags = new WellKnownTags("runtimeid", "hostname", "env", "service", "version", "language")
+
+    // Create keys with different combinations of HTTP fields
+    def keyWithNoSource = new MetricKey("resource", "service", "operation", null, "type", 200, false, false, "server", [], "GET", "/api/users")
+    def keyWithSource = new MetricKey("resource", "service", "operation", "source", "type", 200, false, false, "server", [], "POST", null)
+
+    def content = [
+      Pair.of(keyWithNoSource, new AggregateMetric().recordDurations(1, new AtomicLongArray(1L))),
+      Pair.of(keyWithSource, new AggregateMetric().recordDurations(1, new AtomicLongArray(1L))),
+    ]
+
+    ValidatingSink sink = new ValidatingSink(wellKnownTags, startTime, duration, content)
+    SerializingMetricWriter writer = new SerializingMetricWriter(wellKnownTags, sink, 128)
+
+    when:
+    writer.startBucket(content.size(), startTime, duration)
+    for (Pair<MetricKey, AggregateMetric> pair : content) {
+      writer.add(pair.getLeft(), pair.getRight())
+    }
+    writer.finishBucket()
+
+    then:
+    sink.validatedInput()
+  }
+
   def "HTTPMethod and HTTPEndpoint fields are optional in payload"() {
     setup:
     long startTime = MILLISECONDS.toNanos(System.currentTimeMillis())

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

@@ -10,6 +10,7 @@ class SimpleSpan implements CoreSpan<SimpleSpan> {
   private final String serviceName
   private final String operationName
   private final CharSequence resourceName
+  private final CharSequence serviceNameSource
   private final String type
   private final boolean measured
   private final boolean topLevel
@@ -24,22 +25,23 @@ class SimpleSpan implements CoreSpan<SimpleSpan> {
   private final Map<Object, Object> tags = [:]
 
   SimpleSpan(
-      String serviceName,
-      String operationName,
-      CharSequence resourceName,
-      String type,
-      boolean measured,
-      boolean topLevel,
-      boolean error,
-      long startTime,
-      long duration,
-      int statusCode,
-      boolean traceRoot = false,
-      int longRunningVersion = 0
-  ) {
+  String serviceName,
+  String operationName,
+  CharSequence resourceName,
+  String type,
+  boolean measured,
+  boolean topLevel,
+  boolean error,
+  long startTime,
+  long duration,
+  int statusCode,
+  boolean traceRoot = false,
+  int longRunningVersion = 0,
+  CharSequence serviceNameSource = null) {
     this.serviceName = serviceName
     this.operationName = operationName
     this.resourceName = resourceName
+    this.serviceNameSource = serviceNameSource
     this.type = type
     this.measured = measured
     this.topLevel = topLevel
@@ -61,6 +63,11 @@ class SimpleSpan implements CoreSpan<SimpleSpan> {
     return serviceName
   }
 
+  @Override
+  CharSequence getServiceNameSource() {
+    return serviceNameSource
+  }
+
   @Override
   CharSequence getOperationName() {
     return operationName