Ensure client stat reporter is started when the agent is not available at bootstrap (#9082)

* Ensure metric reporter is started when the agent is not available at bootstrap

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

Co-authored-by: Brice Dutheil <[email protected]>

---------

Co-authored-by: Brice Dutheil <[email protected]>

Author: amarziali

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

@@ -143,20 +143,16 @@ public ConflatingMetricsAggregator(
 
   @Override
   public void start() {
-    if (isMetricsEnabled()) {
-      sink.register(this);
-      thread.start();
-      cancellation =
-          AgentTaskScheduler.INSTANCE.scheduleAtFixedRate(
-              new ReportTask(),
-              this,
-              reportingInterval,
-              reportingInterval,
-              reportingIntervalTimeUnit);
-      log.debug("started metrics aggregator");
-    } else {
-      log.debug("metrics not supported by trace agent");
-    }
+    sink.register(this);
+    thread.start();
+    cancellation =
+        AgentTaskScheduler.INSTANCE.scheduleAtFixedRate(
+            new ReportTask(),
+            this,
+            reportingInterval,
+            reportingInterval,
+            reportingIntervalTimeUnit);
+    log.debug("started metrics aggregator");
   }
 
   private boolean isMetricsEnabled() {

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

@@ -463,6 +463,37 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
     aggregator.close()
   }
 
+  def "should start even if the agent is not available"() {
+    setup:
+    MetricWriter writer = Mock(MetricWriter)
+    Sink sink = Stub(Sink)
+    DDAgentFeaturesDiscovery features = Mock(DDAgentFeaturesDiscovery)
+    features.supportsMetrics() >> false
+    ConflatingMetricsAggregator aggregator = new ConflatingMetricsAggregator(empty,
+      features, sink, writer, 10, queueSize, 200, MILLISECONDS)
+    final spans = [
+      new SimpleSpan("service" , "operation", "resource", "type", false, true, false, 0, 10, HTTP_OK)
+    ]
+    aggregator.start()
+
+    when:
+    aggregator.publish(spans)
+    Thread.sleep(1_000)
+
+    then:
+    0 * writer._
+    when:
+    features.supportsMetrics() >> true
+    aggregator.publish(spans)
+    Thread.sleep(1_000)
+
+    then:
+    (1.._) * writer._
+
+    cleanup:
+    aggregator.close()
+  }
+
   def "force flush should wait for aggregator to start"() {
     setup:
     int maxAggregates = 10