Apply `_sample_rate` metric to allow dd-agent to do proper scaling of metrics when traces are sampled.

dd-trace-ot/src/main/java/datadog/trace/common/writer/DDAgentWriter.java

@@ -113,7 +113,13 @@ public class DDAgentWriter implements Writer {
   public void write(final List<DDSpan> trace) {
     // We can't add events after shutdown otherwise it will never complete shutting down.
     if (traceProcessingDisruptor.running) {
-      final int representativeCount = traceCount.getAndSet(0) + 1;
+      final int representativeCount;
+      if (trace.isEmpty() || !(trace.get(0).isRootSpan())) {
+        // We don't want to reset the count if we can't correctly report the value.
+        representativeCount = 1;
+      } else {
+        representativeCount = traceCount.getAndSet(0) + 1;
+      }
       final boolean published = traceProcessingDisruptor.publish(trace, representativeCount);
 
       if (published) {

dd-trace-ot/src/main/java/datadog/trace/common/writer/ddagent/TraceProcessingDisruptor.java

@@ -63,6 +63,12 @@ public class TraceProcessingDisruptor extends AbstractDisruptor<List<DDSpan>> {
         final DisruptorEvent<List<DDSpan>> event, final long sequence, final boolean endOfBatch) {
       try {
         if (event.data != null) {
+          if (1 < event.representativeCount && !event.data.isEmpty()) {
+            // attempt to have agent scale the metrics properly
+            ((DDSpan) event.data.get(0).getLocalRootSpan())
+                .context()
+                .setMetric("_sample_rate", 1d / event.representativeCount);
+          }
           try {
             final byte[] serializedTrace = api.serializeTrace(event.data);
             batchWritingDisruptor.publish(serializedTrace, event.representativeCount);

dd-trace-ot/src/test/groovy/datadog/trace/api/writer/DDAgentWriterTest.groovy

@@ -131,7 +131,7 @@ class DDAgentWriterTest extends DDSpecification {
     writer.close()
 
     where:
-    span = [newSpanOf(0, "fixed-thread-name")]
+    span = newSpanOf(0, "fixed-thread-name")
     trace = (0..10000).collect { span }
   }
 
@@ -162,7 +162,7 @@ class DDAgentWriterTest extends DDSpecification {
     writer.close()
 
     where:
-    span = [newSpanOf(0, "fixed-thread-name")]
+    span = newSpanOf(0, "fixed-thread-name")
     trace = (1..10).collect { span }
   }