Introduce more local counters for tracer health metrics (#6424)

* Introduce more local counters for tracer health metrics
  This will help us report these counts to tracer-flare in the near future.
* Remove unused metric

Author: mcculls

dd-java-agent/instrumentation/opentelemetry/opentelemetry-0.3/src/test/groovy/OpenTelemetryTest.groovy

@@ -215,8 +215,6 @@ class OpenTelemetryTest extends AgentTestRunner {
 
     then:
     tracer.currentSpan.delegate == secondScope.delegate.span()
-    1 * STATS_D_CLIENT.incrementCounter("scope.close.error")
-    1 * STATS_D_CLIENT.incrementCounter("scope.user.close.error")
     _ * TEST_CHECKPOINTER._
     0 * _
 

dd-java-agent/instrumentation/opentracing/api-0.31/src/test/groovy/OpenTracing31Test.groovy

@@ -255,8 +255,6 @@ class OpenTracing31Test extends AgentTestRunner {
 
     then:
     tracer.scopeManager().active().delegate == secondScope.delegate
-    1 * STATS_D_CLIENT.incrementCounter("scope.close.error")
-    1 * STATS_D_CLIENT.incrementCounter("scope.user.close.error")
     _ * TEST_CHECKPOINTER._
     0 * _
 

dd-java-agent/instrumentation/opentracing/api-0.32/src/test/groovy/OpenTracing32Test.groovy

@@ -270,8 +270,6 @@ class OpenTracing32Test extends AgentTestRunner {
 
     then:
     tracer.scopeManager().active().delegate == secondScope.delegate
-    1 * STATS_D_CLIENT.incrementCounter("scope.close.error")
-    1 * STATS_D_CLIENT.incrementCounter("scope.user.close.error")
     _ * TEST_CHECKPOINTER._
     0 * _
 

dd-trace-core/src/main/java/datadog/trace/core/monitor/TracerHealthMetrics.java

@@ -28,6 +28,7 @@ public class TracerHealthMetrics extends HealthMetrics implements AutoCloseable
       httpStatus -> new String[] {"status:" + httpStatus};
 
   private static final String[] NO_TAGS = new String[0];
+  private static final String[] STATUS_OK_TAGS = STATUS_TAGS.apply(200);
   private final RadixTreeCache<String[]> statusTagsCache =
       new RadixTreeCache<>(16, 32, STATUS_TAGS, 200, 400);
 
@@ -62,6 +63,8 @@ public class TracerHealthMetrics extends HealthMetrics implements AutoCloseable
 
   private final FixedSizeStripedLongCounter enqueuedSpans =
       CountersFactory.createFixedSizeStripedCounter(8);
+  private final FixedSizeStripedLongCounter enqueuedBytes =
+      CountersFactory.createFixedSizeStripedCounter(8);
   private final FixedSizeStripedLongCounter singleSpanSampled =
       CountersFactory.createFixedSizeStripedCounter(8);
   private final FixedSizeStripedLongCounter singleSpanUnsampled =
@@ -84,8 +87,6 @@ public class TracerHealthMetrics extends HealthMetrics implements AutoCloseable
       CountersFactory.createFixedSizeStripedCounter(8);
   private final FixedSizeStripedLongCounter unsetPriorityFailedPublishSpanCount =
       CountersFactory.createFixedSizeStripedCounter(8);
-  private final FixedSizeStripedLongCounter sampledSpans =
-      CountersFactory.createFixedSizeStripedCounter(8);
   private final FixedSizeStripedLongCounter manualTraces =
       CountersFactory.createFixedSizeStripedCounter(8);
   private final FixedSizeStripedLongCounter capturedContinuations =
@@ -102,6 +103,8 @@ public class TracerHealthMetrics extends HealthMetrics implements AutoCloseable
       CountersFactory.createFixedSizeStripedCounter(8);
   private final FixedSizeStripedLongCounter partialTraces =
       CountersFactory.createFixedSizeStripedCounter(8);
+  private final FixedSizeStripedLongCounter partialBytes =
+      CountersFactory.createFixedSizeStripedCounter(8);
   private final FixedSizeStripedLongCounter clientSpansWithoutContext =
       CountersFactory.createFixedSizeStripedCounter(8);
   private final FixedSizeStripedLongCounter longRunningTracesWrite =
@@ -111,6 +114,21 @@ public class TracerHealthMetrics extends HealthMetrics implements AutoCloseable
   private final FixedSizeStripedLongCounter longRunningTracesExpired =
       CountersFactory.createFixedSizeStripedCounter(8);
 
+  private final FixedSizeStripedLongCounter apiRequests =
+      CountersFactory.createFixedSizeStripedCounter(8);
+  private final FixedSizeStripedLongCounter apiResponsesOK =
+      CountersFactory.createFixedSizeStripedCounter(8);
+  private final FixedSizeStripedLongCounter apiErrors =
+      CountersFactory.createFixedSizeStripedCounter(8);
+  private final FixedSizeStripedLongCounter flushedTraces =
+      CountersFactory.createFixedSizeStripedCounter(8);
+  private final FixedSizeStripedLongCounter flushedBytes =
+      CountersFactory.createFixedSizeStripedCounter(8);
+  private final FixedSizeStripedLongCounter scopeCloseErrors =
+      CountersFactory.createFixedSizeStripedCounter(8);
+  private final FixedSizeStripedLongCounter userScopeCloseErrors =
+      CountersFactory.createFixedSizeStripedCounter(8);
+
   private final StatsDClient statsd;
   private final long interval;
   private final TimeUnit units;
@@ -216,7 +234,7 @@ public void onFlush(final boolean early) {}
 
   @Override
   public void onPartialFlush(final int sizeInBytes) {
-    statsd.count("span.flushed.partial", sizeInBytes, NO_TAGS);
+    partialBytes.inc(sizeInBytes);
   }
 
   @Override
@@ -233,7 +251,7 @@ public void onSingleSpanUnsampled() {
   public void onSerialize(final int serializedSizeInBytes) {
     // DQH - Because of Java tracer's 2 phase acceptance and serialization scheme, this doesn't
     // map precisely
-    statsd.count("queue.enqueued.bytes", serializedSizeInBytes, NO_TAGS);
+    enqueuedBytes.inc(serializedSizeInBytes);
   }
 
   @Override
@@ -266,9 +284,9 @@ public void onCreateManualTrace() {
 
   @Override
   public void onScopeCloseError(int scopeSource) {
-    statsd.incrementCounter("scope.close.error", NO_TAGS);
+    scopeCloseErrors.inc();
     if (scopeSource == ScopeSource.MANUAL.id()) {
-      statsd.incrementCounter("scope.user.close.error", NO_TAGS);
+      userScopeCloseErrors.inc();
     }
   }
 
@@ -323,19 +341,24 @@ public void onLongRunningUpdate(final int dropped, final int write, final int ex
 
   private void onSendAttempt(
       final int traceCount, final int sizeInBytes, final RemoteApi.Response response) {
-    statsd.incrementCounter("api.requests.total", NO_TAGS);
-    statsd.count("flush.traces.total", traceCount, NO_TAGS);
+    apiRequests.inc();
+    flushedTraces.inc(traceCount);
     // TODO: missing queue.spans (# of spans being sent)
-    statsd.count("flush.bytes.total", sizeInBytes, NO_TAGS);
+    flushedBytes.inc(sizeInBytes);
 
     if (response.exception() != null) {
       // covers communication errors -- both not receiving a response or
       // receiving malformed response (even when otherwise successful)
-      statsd.incrementCounter("api.errors.total", NO_TAGS);
+      apiErrors.inc();
     }
 
-    if (response.status() != null) {
-      statsd.incrementCounter("api.responses.total", statusTagsCache.get(response.status()));
+    Integer status = response.status();
+    if (status != null) {
+      if (200 == status) {
+        apiResponsesOK.inc();
+      } else {
+        statsd.incrementCounter("api.responses.total", statusTagsCache.get(status));
+      }
     }
   }
 
@@ -415,6 +438,7 @@ public void run(TracerHealthMetrics target) {
           target.userDropFailedPublishSpanCount,
           USER_DROP_TAG);
       reportIfChanged(target.statsd, "queue.enqueued.spans", target.enqueuedSpans, NO_TAGS);
+      reportIfChanged(target.statsd, "queue.enqueued.bytes", target.enqueuedBytes, NO_TAGS);
       reportIfChanged(target.statsd, "trace.pending.created", target.createdTraces, NO_TAGS);
       reportIfChanged(target.statsd, "span.pending.created", target.createdSpans, NO_TAGS);
       reportIfChanged(target.statsd, "span.pending.finished", target.finishedSpans, NO_TAGS);
@@ -424,6 +448,7 @@ public void run(TracerHealthMetrics target) {
       reportIfChanged(
           target.statsd, "span.continuations.finished", target.finishedContinuations, NO_TAGS);
       reportIfChanged(target.statsd, "queue.partial.traces", target.partialTraces, NO_TAGS);
+      reportIfChanged(target.statsd, "span.flushed.partial", target.partialBytes, NO_TAGS);
       reportIfChanged(
           target.statsd, "span.client.no-context", target.clientSpansWithoutContext, NO_TAGS);
       reportIfChanged(
@@ -442,6 +467,16 @@ public void run(TracerHealthMetrics target) {
           target.statsd, "long-running.dropped", target.longRunningTracesDropped, NO_TAGS);
       reportIfChanged(
           target.statsd, "long-running.expired", target.longRunningTracesExpired, NO_TAGS);
+
+      reportIfChanged(target.statsd, "api.requests.total", target.apiRequests, NO_TAGS);
+      reportIfChanged(target.statsd, "api.errors.total", target.apiErrors, NO_TAGS);
+      // non-OK responses are reported immediately in onSendAttempt with different status tags
+      reportIfChanged(target.statsd, "api.responses.total", target.apiResponsesOK, STATUS_OK_TAGS);
+      reportIfChanged(target.statsd, "flush.traces.total", target.flushedTraces, NO_TAGS);
+      reportIfChanged(target.statsd, "flush.bytes.total", target.flushedBytes, NO_TAGS);
+      reportIfChanged(target.statsd, "scope.close.error", target.scopeCloseErrors, NO_TAGS);
+      reportIfChanged(
+          target.statsd, "scope.user.close.error", target.userScopeCloseErrors, NO_TAGS);
     }
 
     private void reportIfChanged(

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

@@ -29,6 +29,7 @@ import spock.lang.Timeout
 import spock.util.concurrent.PollingConditions
 
 import java.nio.ByteBuffer
+import java.util.concurrent.CountDownLatch
 import java.util.concurrent.Phaser
 import java.util.concurrent.Semaphore
 import java.util.concurrent.TimeUnit
@@ -692,45 +693,39 @@ class DDAgentWriterCombinedTest extends DDCoreSpecification {
   }
 
   def "statsd comm failure"() {
-    def numRequests = new AtomicInteger(0)
-    def numResponses = new AtomicInteger(0)
-    def numErrors = new AtomicInteger(0)
-
     setup:
     def minimalTrace = createMinimalTrace()
 
     def api = apiWithVersion(agentVersion)
     api.sendSerializedTraces(_) >> RemoteApi.Response.failed(new IOException("comm error"))
 
-    def statsd = Stub(StatsDClient)
-    statsd.incrementCounter("api.requests.total") >> { stat ->
-      numRequests.incrementAndGet()
-    }
-    statsd.incrementCounter("api.responses.total", _) >> { stat, tags ->
-      numResponses.incrementAndGet()
-    }
-    statsd.incrementCounter("api.errors.total", _) >> { stat ->
-      numErrors.incrementAndGet()
-    }
-
-    def healthMetrics = new TracerHealthMetrics(statsd)
+    def latch = new CountDownLatch(2)
+    def statsd = Mock(StatsDClient)
+    def healthMetrics = new TracerHealthMetrics(statsd, 100, TimeUnit.MILLISECONDS)
     def writer = DDAgentWriter.builder()
       .traceAgentV05Enabled(true)
       .agentApi(api).monitoring(monitoring)
       .healthMetrics(healthMetrics).build()
+    healthMetrics.start()
     writer.start()
 
     when:
     writer.write(minimalTrace)
     writer.flush()
+    latch.await(10, TimeUnit.SECONDS)
 
     then:
-    numRequests.get() == 1
-    numResponses.get() == 0
-    numErrors.get() == 1
+    1 * statsd.count("api.requests.total", 1, _) >> {
+      latch.countDown()
+    }
+    0 * statsd.incrementCounter("api.responses.total", _)
+    1 * statsd.count("api.errors.total", 1, _) >> {
+      latch.countDown()
+    }
 
     cleanup:
     writer.close()
+    healthMetrics.close()
 
     where:
     agentVersion << ["v0.3/traces", "v0.4/traces", "v0.5/traces"]

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

@@ -29,6 +29,7 @@ import spock.lang.Timeout
 import spock.util.concurrent.PollingConditions
 
 import java.nio.ByteBuffer
+import java.util.concurrent.CountDownLatch
 import java.util.concurrent.Phaser
 import java.util.concurrent.Semaphore
 import java.util.concurrent.TimeUnit
@@ -669,47 +670,41 @@ class DDIntakeWriterCombinedTest extends DDCoreSpecification {
   }
 
   def "statsd comm failure"() {
-    def numRequests = new AtomicInteger(0)
-    def numResponses = new AtomicInteger(0)
-    def numErrors = new AtomicInteger(0)
-
     setup:
     def minimalTrace = createMinimalTrace()
 
     def api = Mock(DDIntakeApi)
     api.sendSerializedTraces(_) >> RemoteApi.Response.failed(new IOException("comm error"))
 
-    def statsd = Stub(StatsDClient)
-    statsd.incrementCounter("api.requests.total") >> { stat ->
-      numRequests.incrementAndGet()
-    }
-    statsd.incrementCounter("api.responses.total", _) >> { stat, tags ->
-      numResponses.incrementAndGet()
-    }
-    statsd.incrementCounter("api.errors.total", _) >> { stat ->
-      numErrors.incrementAndGet()
-    }
-
-    def healthMetrics = new TracerHealthMetrics(statsd)
+    def latch = new CountDownLatch(2)
+    def statsd = Mock(StatsDClient)
+    def healthMetrics = new TracerHealthMetrics(statsd, 100, TimeUnit.MILLISECONDS)
     def writer = DDIntakeWriter.builder()
       .addTrack(trackType, api)
       .monitoring(monitoring)
       .healthMetrics(healthMetrics)
       .alwaysFlush(false)
       .build()
+    healthMetrics.start()
     writer.start()
 
     when:
     writer.write(minimalTrace)
     writer.flush()
+    latch.await(10, TimeUnit.SECONDS)
 
     then:
-    numRequests.get() == 1
-    numResponses.get() == 0
-    numErrors.get() == 1
+    1 * statsd.count("api.requests.total", 1, _) >> {
+      latch.countDown()
+    }
+    0 * statsd.incrementCounter("api.responses.total", _)
+    1 * statsd.count("api.errors.total", 1, _) >> {
+      latch.countDown()
+    }
 
     cleanup:
     writer.close()
+    healthMetrics.close()
 
     where:
     trackType | apiVersion