fix(datadog_traces sink): APM stats payloads are sent independent of …

…trace payloads and at a set interval. (#15084)

- APM stats are now "flushed" in a separate thread, detached from the sink's stream loop.
- The stats flushing thread flushes the oldest bucket every 10 seconds, and caches the last two 10 second buckets.
- When sink is shutting down, the APM stats thread flushes remaining buckets.

scripts/integration/docker-compose.datadog-traces.yml

@@ -2,7 +2,9 @@ version: "3"
 
 services:
   datadog-trace-agent:
-    image: docker.io/datadog/agent:7.39.1
+    # "latest" tag is always the most recent stable release
+    # this is a fallback in case "breaking" changes in the Agent aren't surfaced to Vector
+    image: docker.io/datadog/agent:latest
     networks:
       - backend
     environment:
@@ -20,8 +22,11 @@ services:
       - DD_APM_ERROR_TPS=999999
       - DD_APM_MAX_MEMORY=0
       - DD_APM_MAX_CPU_PERCENT=0
+      - DD_HOSTNAME=datadog-trace-agent
   datadog-trace-agent-to-vector:
-    image: docker.io/datadog/agent:7.39.1
+    # "latest" tag is always the most recent stable release
+    # this is a fallback in case "breaking" changes in the Agent aren't surfaced to Vector
+    image: docker.io/datadog/agent:latest
     networks:
       - backend
     environment:
@@ -39,6 +44,7 @@ services:
       - DD_APM_ERROR_TPS=999999
       - DD_APM_MAX_MEMORY=0
       - DD_APM_MAX_CPU_PERCENT=0
+      - DD_HOSTNAME=datadog-trace-agent-to-vector
   runner:
     build:
       context: ${PWD}

src/internal_events/datadog_traces.rs

@@ -38,3 +38,27 @@ impl InternalEvent for DatadogTracesEncodingError {
         }
     }
 }
+
+#[derive(Debug)]
+pub struct DatadogTracesAPMStatsError<E> {
+    pub error: E,
+}
+
+impl<E: std::fmt::Display> InternalEvent for DatadogTracesAPMStatsError<E> {
+    fn emit(self) {
+        error!(
+            message = "Failed sending APM stats payload.",
+            error = %self.error,
+            error_type = error_type::WRITER_FAILED,
+            stage = error_stage::SENDING,
+            internal_log_rate_limit = true,
+        );
+        counter!(
+            "component_errors_total", 1,
+            "error_type" => error_type::WRITER_FAILED,
+            "stage" => error_stage::SENDING,
+        );
+
+        // No dropped events because APM stats payloads are not considered events.
+    }
+}