Revert "Lazily obtain feature discovery when starting client metrics (#9548)"

This reverts commit 561eb5e.

Author: amarziali

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

@@ -4,7 +4,6 @@
 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;
@@ -33,25 +32,20 @@
 @OutputTimeUnit(MICROSECONDS)
 @Fork(value = 1)
 public class ConflatingMetricsAggregatorBenchmark {
-  private final SharedCommunicationObjects sco = new SharedCommunicationObjects();
-  private final ConflatingMetricsAggregator aggregator;
+  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 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,8 +98,7 @@ public final class ConflatingMetricsAggregator implements MetricsAggregator, Eve
   private final Aggregator aggregator;
   private final long reportingInterval;
   private final TimeUnit reportingIntervalTimeUnit;
-  private final SharedCommunicationObjects sharedCommunicationObjects;
-  private volatile DDAgentFeaturesDiscovery features;
+  private final DDAgentFeaturesDiscovery features;
   private final HealthMetrics healthMetrics;
 
   private volatile AgentTaskScheduler.Scheduled<?> cancellation;
@@ -111,7 +110,7 @@ public ConflatingMetricsAggregator(
     this(
         config.getWellKnownTags(),
         config.getMetricsIgnoredResources(),
-        sharedCommunicationObjects,
+        sharedCommunicationObjects.featuresDiscovery(config),
         healthMetrics,
         new OkHttpSink(
             sharedCommunicationObjects.okHttpClient,
@@ -127,15 +126,15 @@ public ConflatingMetricsAggregator(
   ConflatingMetricsAggregator(
       WellKnownTags wellKnownTags,
       Set<String> ignoredResources,
-      SharedCommunicationObjects sharedCommunicationObjects,
+      DDAgentFeaturesDiscovery features,
       HealthMetrics healthMetric,
       Sink sink,
       int maxAggregates,
       int queueSize) {
     this(
         wellKnownTags,
         ignoredResources,
-        sharedCommunicationObjects,
+        features,
         healthMetric,
         sink,
         maxAggregates,
@@ -147,7 +146,7 @@ public ConflatingMetricsAggregator(
   ConflatingMetricsAggregator(
       WellKnownTags wellKnownTags,
       Set<String> ignoredResources,
-      SharedCommunicationObjects sharedCommunicationObjects,
+      DDAgentFeaturesDiscovery features,
       HealthMetrics healthMetric,
       Sink sink,
       int maxAggregates,
@@ -156,7 +155,7 @@ public ConflatingMetricsAggregator(
       TimeUnit timeUnit) {
     this(
         ignoredResources,
-        sharedCommunicationObjects,
+        features,
         healthMetric,
         sink,
         new SerializingMetricWriter(wellKnownTags, sink),
@@ -168,7 +167,7 @@ public ConflatingMetricsAggregator(
 
   ConflatingMetricsAggregator(
       Set<String> ignoredResources,
-      SharedCommunicationObjects sharedCommunicationObjects,
+      DDAgentFeaturesDiscovery features,
       HealthMetrics healthMetric,
       Sink sink,
       MetricWriter metricWriter,
@@ -181,7 +180,7 @@ public ConflatingMetricsAggregator(
     this.batchPool = new SpmcArrayQueue<>(maxAggregates);
     this.pending = new NonBlockingHashMap<>(maxAggregates * 4 / 3);
     this.keys = new NonBlockingHashMap<>();
-    this.sharedCommunicationObjects = sharedCommunicationObjects;
+    this.features = features;
     this.healthMetrics = healthMetric;
     this.sink = sink;
     this.aggregator =
@@ -199,18 +198,6 @@ public ConflatingMetricsAggregator(
     this.reportingIntervalTimeUnit = timeUnit;
   }
 
-  private DDAgentFeaturesDiscovery featuresDiscovery() {
-    DDAgentFeaturesDiscovery ret = features;
-    if (ret != null) {
-      return ret;
-    }
-    // no need to synchronise here since it's already done in sharedCommunicationObject.
-    // At worst, we'll assign multiple time the variable but it will be the same object
-    ret = sharedCommunicationObjects.featuresDiscovery(Config.get());
-    features = ret;
-    return ret;
-  }
-
   @Override
   public void start() {
     sink.register(this);
@@ -226,6 +213,13 @@ public void start() {
     log.debug("started metrics aggregator");
   }
 
+  private boolean isMetricsEnabled() {
+    if (features.getMetricsEndpoint() == null) {
+      features.discoverIfOutdated();
+    }
+    return features.supportsMetrics();
+  }
+
   @Override
   public boolean report() {
     boolean published;
@@ -242,7 +236,8 @@ public boolean report() {
 
   @Override
   public Future<Boolean> forceReport() {
-    if (!featuresDiscovery().supportsMetrics()) {
+    // Ensure the feature is enabled
+    if (!isMetricsEnabled()) {
       return CompletableFuture.completedFuture(false);
     }
     // Wait for the thread to start
@@ -278,7 +273,6 @@ public Future<Boolean> forceReport() {
   public boolean publish(List<? extends CoreSpan<?>> trace) {
     boolean forceKeep = false;
     int counted = 0;
-    final DDAgentFeaturesDiscovery features = featuresDiscovery();
     if (features.supportsMetrics()) {
       for (CoreSpan<?> span : trace) {
         boolean isTopLevel = span.isTopLevel();
@@ -289,7 +283,7 @@ public boolean publish(List<? extends CoreSpan<?>> trace) {
             break;
           }
           counted++;
-          forceKeep |= publish(span, isTopLevel, features);
+          forceKeep |= publish(span, isTopLevel);
         }
       }
       healthMetrics.onClientStatTraceComputed(
@@ -311,7 +305,7 @@ private boolean spanKindEligible(CoreSpan<?> span) {
     return spanKind != null && ELIGIBLE_SPAN_KINDS_FOR_METRICS.contains(spanKind.toString());
   }
 
-  private boolean publish(CoreSpan<?> span, boolean isTopLevel, DDAgentFeaturesDiscovery features) {
+  private boolean publish(CoreSpan<?> span, boolean isTopLevel) {
     final CharSequence spanKind = span.getTag(SPAN_KIND, "");
     MetricKey newKey =
         new MetricKey(
@@ -324,7 +318,7 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel, DDAgentFeaturesDis
             span.getParentId() == 0,
             SPAN_KINDS.computeIfAbsent(
                 spanKind, UTF8BytesString::create), // save repeated utf8 conversions
-            getPeerTags(span, spanKind.toString(), features));
+            getPeerTags(span, spanKind.toString()));
     boolean isNewKey = false;
     MetricKey key = keys.putIfAbsent(newKey, newKey);
     if (null == key) {
@@ -359,8 +353,7 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel, DDAgentFeaturesDis
     return isNewKey || span.getError() > 0;
   }
 
-  private List<UTF8BytesString> getPeerTags(
-      CoreSpan<?> span, String spanKind, DDAgentFeaturesDiscovery features) {
+  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()) {
@@ -424,7 +417,8 @@ public void onEvent(EventType eventType, String message) {
     switch (eventType) {
       case DOWNGRADED:
         log.debug("Agent downgrade was detected");
-        AgentTaskScheduler.get().execute(this::disable);
+        disable();
+        healthMetrics.onClientStatDowngraded();
         break;
       case BAD_PAYLOAD:
         log.debug("bad metrics payload sent to trace agent: {}", message);
@@ -440,11 +434,9 @@ public void onEvent(EventType eventType, String message) {
   }
 
   private void disable() {
-    final DDAgentFeaturesDiscovery features = featuresDiscovery();
     features.discover();
     if (!features.supportsMetrics()) {
       log.debug("Disabling metric reporting because an agent downgrade was detected");
-      healthMetrics.onClientStatDowngraded();
       this.pending.clear();
       this.batchPool.clear();
       this.inbox.clear();

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

@@ -789,10 +789,8 @@ private CoreTracer(
     // 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.
-    sharedCommunicationObjects.whenReady(
-        () ->
-            AgentTaskScheduler.get()
-                .scheduleWithJitter(MetricsAggregator::start, metricsAggregator, 1, SECONDS));
+    AgentTaskScheduler.get()
+        .scheduleWithJitter(MetricsAggregator::start, metricsAggregator, 1, SECONDS);
 
     if (dataStreamsMonitoring == null) {
       this.dataStreamsMonitoring =