Lazily obtain feature discovery when starting client metrics

Author: amarziali

dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ConflatingMetricsAggregatorBenchmark.java

@@ -4,6 +4,7 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 
 import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
+import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.communication.monitor.Monitoring;
 import datadog.trace.api.WellKnownTags;
 import datadog.trace.core.CoreSpan;
@@ -32,20 +33,25 @@
 @OutputTimeUnit(MICROSECONDS)
 @Fork(value = 1)
 public class ConflatingMetricsAggregatorBenchmark {
-  private final DDAgentFeaturesDiscovery featuresDiscovery =
-      new FixedAgentFeaturesDiscovery(
-          Collections.singleton("peer.hostname"), Collections.emptySet());
-  private final ConflatingMetricsAggregator aggregator =
-      new ConflatingMetricsAggregator(
-          new WellKnownTags("", "", "", "", "", ""),
-          Collections.emptySet(),
-          featuresDiscovery,
-          HealthMetrics.NO_OP,
-          new NullSink(),
-          2048,
-          2048);
+  private final SharedCommunicationObjects sco = new SharedCommunicationObjects();
+  private final ConflatingMetricsAggregator aggregator;
   private final List<CoreSpan<?>> spans = generateTrace(64);
 
+  public ConflatingMetricsAggregatorBenchmark() {
+    sco.setFeaturesDiscovery(
+        new FixedAgentFeaturesDiscovery(
+            Collections.singleton("peer.hostname"), Collections.emptySet()));
+    aggregator =
+        new ConflatingMetricsAggregator(
+            new WellKnownTags("", "", "", "", "", ""),
+            Collections.emptySet(),
+            sco,
+            HealthMetrics.NO_OP,
+            new NullSink(),
+            2048,
+            2048);
+  }
+
   static List<CoreSpan<?>> generateTrace(int len) {
     final List<CoreSpan<?>> trace = new ArrayList<>();
     for (int i = 0; i < len; i++) {

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

@@ -98,7 +98,8 @@ public final class ConflatingMetricsAggregator implements MetricsAggregator, Eve
   private final Aggregator aggregator;
   private final long reportingInterval;
   private final TimeUnit reportingIntervalTimeUnit;
-  private final DDAgentFeaturesDiscovery features;
+  private final SharedCommunicationObjects sharedCommunicationObjects;
+  private DDAgentFeaturesDiscovery features;
   private final HealthMetrics healthMetrics;
 
   private volatile AgentTaskScheduler.Scheduled<?> cancellation;
@@ -110,7 +111,7 @@ public ConflatingMetricsAggregator(
     this(
         config.getWellKnownTags(),
         config.getMetricsIgnoredResources(),
-        sharedCommunicationObjects.featuresDiscovery(config),
+        sharedCommunicationObjects,
         healthMetrics,
         new OkHttpSink(
             sharedCommunicationObjects.okHttpClient,
@@ -126,15 +127,15 @@ public ConflatingMetricsAggregator(
   ConflatingMetricsAggregator(
       WellKnownTags wellKnownTags,
       Set<String> ignoredResources,
-      DDAgentFeaturesDiscovery features,
+      SharedCommunicationObjects sharedCommunicationObjects,
       HealthMetrics healthMetric,
       Sink sink,
       int maxAggregates,
       int queueSize) {
     this(
         wellKnownTags,
         ignoredResources,
-        features,
+        sharedCommunicationObjects,
         healthMetric,
         sink,
         maxAggregates,
@@ -146,7 +147,7 @@ public ConflatingMetricsAggregator(
   ConflatingMetricsAggregator(
       WellKnownTags wellKnownTags,
       Set<String> ignoredResources,
-      DDAgentFeaturesDiscovery features,
+      SharedCommunicationObjects sharedCommunicationObjects,
       HealthMetrics healthMetric,
       Sink sink,
       int maxAggregates,
@@ -155,7 +156,7 @@ public ConflatingMetricsAggregator(
       TimeUnit timeUnit) {
     this(
         ignoredResources,
-        features,
+        sharedCommunicationObjects,
         healthMetric,
         sink,
         new SerializingMetricWriter(wellKnownTags, sink),
@@ -167,7 +168,7 @@ public ConflatingMetricsAggregator(
 
   ConflatingMetricsAggregator(
       Set<String> ignoredResources,
-      DDAgentFeaturesDiscovery features,
+      SharedCommunicationObjects sharedCommunicationObjects,
       HealthMetrics healthMetric,
       Sink sink,
       MetricWriter metricWriter,
@@ -180,7 +181,7 @@ public ConflatingMetricsAggregator(
     this.batchPool = new SpmcArrayQueue<>(maxAggregates);
     this.pending = new NonBlockingHashMap<>(maxAggregates * 4 / 3);
     this.keys = new NonBlockingHashMap<>();
-    this.features = features;
+    this.sharedCommunicationObjects = sharedCommunicationObjects;
     this.healthMetrics = healthMetric;
     this.sink = sink;
     this.aggregator =
@@ -200,6 +201,7 @@ public ConflatingMetricsAggregator(
 
   @Override
   public void start() {
+    features = sharedCommunicationObjects.featuresDiscovery(Config.get());
     sink.register(this);
     thread.start();
     cancellation =
@@ -214,10 +216,13 @@ public void start() {
   }
 
   private boolean isMetricsEnabled() {
-    if (features.getMetricsEndpoint() == null) {
-      features.discoverIfOutdated();
+    if (features != null) {
+      if (features.getMetricsEndpoint() == null) {
+        features.discoverIfOutdated();
+      }
+      return features.supportsMetrics();
     }
-    return features.supportsMetrics();
+    return false;
   }
 
   @Override
@@ -273,7 +278,7 @@ public Future<Boolean> forceReport() {
   public boolean publish(List<? extends CoreSpan<?>> trace) {
     boolean forceKeep = false;
     int counted = 0;
-    if (features.supportsMetrics()) {
+    if (features != null && features.supportsMetrics()) {
       for (CoreSpan<?> span : trace) {
         boolean isTopLevel = span.isTopLevel();
         if (shouldComputeMetric(span)) {
@@ -354,31 +359,34 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel) {
   }
 
   private List<UTF8BytesString> getPeerTags(CoreSpan<?> span, String spanKind) {
-    if (ELIGIBLE_SPAN_KINDS_FOR_PEER_AGGREGATION.contains(spanKind)) {
-      List<UTF8BytesString> peerTags = new ArrayList<>();
-      for (String peerTag : features.peerTags()) {
-        Object value = span.getTag(peerTag);
-        if (value != null) {
+    if (features != null) {
+      if (ELIGIBLE_SPAN_KINDS_FOR_PEER_AGGREGATION.contains(spanKind)) {
+        List<UTF8BytesString> peerTags = new ArrayList<>();
+        for (String peerTag : features.peerTags()) {
+          Object value = span.getTag(peerTag);
+          if (value != null) {
+            final Pair<DDCache<String, UTF8BytesString>, Function<String, UTF8BytesString>>
+                cacheAndCreator = PEER_TAGS_CACHE.computeIfAbsent(peerTag, PEER_TAGS_CACHE_ADDER);
+            peerTags.add(
+                cacheAndCreator
+                    .getLeft()
+                    .computeIfAbsent(value.toString(), cacheAndCreator.getRight()));
+          }
+        }
+        return peerTags;
+      } else if (SPAN_KIND_INTERNAL.equals(spanKind)) {
+        // in this case only the base service should be aggregated if present
+        final Object baseService = span.getTag(BASE_SERVICE);
+        if (baseService != null) {
           final Pair<DDCache<String, UTF8BytesString>, Function<String, UTF8BytesString>>
-              cacheAndCreator = PEER_TAGS_CACHE.computeIfAbsent(peerTag, PEER_TAGS_CACHE_ADDER);
-          peerTags.add(
+              cacheAndCreator =
+                  PEER_TAGS_CACHE.computeIfAbsent(BASE_SERVICE, PEER_TAGS_CACHE_ADDER);
+          return Collections.singletonList(
               cacheAndCreator
                   .getLeft()
-                  .computeIfAbsent(value.toString(), cacheAndCreator.getRight()));
+                  .computeIfAbsent(baseService.toString(), cacheAndCreator.getRight()));
         }
       }
-      return peerTags;
-    } else if (SPAN_KIND_INTERNAL.equals(spanKind)) {
-      // in this case only the base service should be aggregated if present
-      final Object baseService = span.getTag(BASE_SERVICE);
-      if (baseService != null) {
-        final Pair<DDCache<String, UTF8BytesString>, Function<String, UTF8BytesString>>
-            cacheAndCreator = PEER_TAGS_CACHE.computeIfAbsent(BASE_SERVICE, PEER_TAGS_CACHE_ADDER);
-        return Collections.singletonList(
-            cacheAndCreator
-                .getLeft()
-                .computeIfAbsent(baseService.toString(), cacheAndCreator.getRight()));
-      }
     }
     return Collections.emptyList();
   }
@@ -434,6 +442,7 @@ public void onEvent(EventType eventType, String message) {
   }
 
   private void disable() {
+    // note: disable is called only if started so we're not nullchecking before accessing features
     features.discover();
     if (!features.supportsMetrics()) {
       log.debug("Disabling metric reporting because an agent downgrade was detected");

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

@@ -100,7 +100,6 @@
 import datadog.trace.core.traceinterceptor.LatencyTraceInterceptor;
 import datadog.trace.lambda.LambdaHandler;
 import datadog.trace.relocate.api.RatelimitedLogger;
-import datadog.trace.util.AgentTaskScheduler;
 import java.io.IOException;
 import java.lang.ref.WeakReference;
 import java.math.BigInteger;
@@ -790,11 +789,8 @@ private CoreTracer(
 
     metricsAggregator =
         createMetricsAggregator(config, sharedCommunicationObjects, this.healthMetrics);
-    // Schedule the metrics aggregator to begin reporting after a random delay of 1 to 10 seconds
-    // (using milliseconds granularity.) This avoids a fleet of traced applications starting at the
-    // same time from sending metrics in sync.
-    AgentTaskScheduler.get()
-        .scheduleWithJitter(MetricsAggregator::start, metricsAggregator, 1, SECONDS);
+    // the jitter is brought implicitly by the callback that can vary
+    sharedCommunicationObjects.whenReady(metricsAggregator::start);
 
     if (dataStreamsMonitoring == null) {
       this.dataStreamsMonitoring =