Exclude long running partial versions from stats

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

@@ -293,6 +293,8 @@ public boolean publish(List<? extends CoreSpan<?>> trace) {
 
   private boolean shouldComputeMetric(CoreSpan<?> span) {
     return (span.isMeasured() || span.isTopLevel() || spanKindEligible(span))
+        && span.getLongRunningVersion()
+            <= 0 // either not long-running or unpublished long-running span
         && span.getDurationNano() > 0;
   }
 

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

@@ -114,4 +114,14 @@ T setSamplingPriority(
    * @return this
    */
   T setMetaStruct(final String field, final Object value);
+
+  /**
+   * Version of a span that can be set by the long running spans feature:
+   * <li>eq 0 -> span is not long running.
+   * <li>lt 0 -> finished span that had running versions previously written.
+   * <li>gt 0 -> long running span and its write version.
+   *
+   * @return the version.
+   */
+  int getLongRunningVersion();
 }

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

@@ -869,4 +869,9 @@ public void copyPropagationAndBaggage(final AgentSpan source) {
       sourceSpanContext.getBaggageItems().forEach(context::setBaggageItem);
     }
   }
+
+  @Override
+  public int getLongRunningVersion() {
+    return longRunningVersion;
+  }
 }

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

@@ -820,6 +820,48 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
     aggregator.close()
   }
 
+  def "should not count partial snapshot(long running)"() {
+    setup:
+    MetricWriter writer = Mock(MetricWriter)
+    Sink sink = Stub(Sink)
+    DDAgentFeaturesDiscovery features = Mock(DDAgentFeaturesDiscovery)
+    features.supportsMetrics() >> true
+    ConflatingMetricsAggregator aggregator = new ConflatingMetricsAggregator(empty,
+      features, HealthMetrics.NO_OP, sink, writer, 10, queueSize, reportingInterval, SECONDS)
+    aggregator.start()
+
+    when:
+    CountDownLatch latch = new CountDownLatch(1)
+    aggregator.publish([
+      new SimpleSpan("service", "operation", "resource", "type", true, true, false, 0, 100, HTTP_OK, true, 12345),
+      new SimpleSpan("service", "operation", "resource", "type", true, true, false, 0, 100, HTTP_OK, true, 0)
+    ])
+    aggregator.report()
+    def latchTriggered = latch.await(2, SECONDS)
+
+    then:
+    latchTriggered
+    1 * writer.startBucket(1, _, _)
+    1 * writer.add(
+      new MetricKey(
+      "resource",
+      "service",
+      "operation",
+      "type",
+      HTTP_OK,
+      false,
+      true,
+      "",
+      []
+      ), { AggregateMetric aggregateMetric ->
+        aggregateMetric.getHitCount() == 1 && aggregateMetric.getTopLevelCount() == 1 && aggregateMetric.getDuration() == 100
+      })
+    1 * writer.finishBucket() >> { latch.countDown() }
+
+    cleanup:
+    aggregator.close()
+  }
+
   def reportAndWaitUntilEmpty(ConflatingMetricsAggregator aggregator) {
     waitUntilEmpty(aggregator)
     aggregator.report()

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

@@ -19,6 +19,7 @@ class SimpleSpan implements CoreSpan<SimpleSpan> {
 
   private final long duration
   private final long startTime
+  private final long longRunningVersion
 
   private final Map<Object, Object> tags = [:]
 
@@ -33,7 +34,8 @@ class SimpleSpan implements CoreSpan<SimpleSpan> {
   long startTime,
   long duration,
   int statusCode,
-  boolean traceRoot = false
+  boolean traceRoot = false,
+  int longRunningVersion = 0
   ) {
     this.serviceName = serviceName
     this.operationName = operationName
@@ -46,6 +48,7 @@ class SimpleSpan implements CoreSpan<SimpleSpan> {
     this.startTime = startTime
     this.duration = duration
     this.statusCode = (short) statusCode
+    this.longRunningVersion = longRunningVersion
   }
 
   @Override
@@ -266,4 +269,9 @@ class SimpleSpan implements CoreSpan<SimpleSpan> {
   SimpleSpan setMetaStruct(String field, Object value) {
     return this
   }
+
+  @Override
+  int getLongRunningVersion() {
+    return longRunningVersion
+  }
 }

dd-trace-core/src/test/groovy/datadog/trace/common/writer/TraceGenerator.groovy

@@ -430,5 +430,10 @@ class TraceGenerator {
       }
       return this
     }
+
+    @Override
+    int getLongRunningVersion() {
+      return 0
+    }
   }
 }

dd-trace-core/src/traceAgentTest/groovy/TraceGenerator.groovy

@@ -401,5 +401,10 @@ class TraceGenerator {
     PojoSpan setMetaStruct(String field, Object value) {
       return this
     }
+
+    @Override
+    int getLongRunningVersion() {
+      return 0
+    }
   }
 }