fix(trace-agent): adapt stats concentrator to libdatadog v32.0.0

libdatadog v32.0.0 dropped the Eq/Hash derives on TracerMetadata
(serializable-only), so it can no longer be used directly as a HashMap
key. Project the discriminating fields (tracer language, service name,
service env, service version, container id) into a new ConcentratorKey,
and store the originating Arc<TracerMetadata> alongside the
SpanConcentrator so flushed payloads still carry the full tracer
context.

Author: duncanista

crates/datadog-trace-agent/src/stats_concentrator_service.rs

@@ -70,9 +70,37 @@ fn new_concentrator() -> SpanConcentrator {
     )
 }
 
+/// Hashable key derived from `TracerMetadata`.
+///
+/// `TracerMetadata` itself does not implement `Eq`/`Hash` (libdatadog v32
+/// removed those derives), so we project the fields used to discriminate
+/// concentrators into a separate key type.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+struct ConcentratorKey {
+    tracer_language: String,
+    service_name: Option<String>,
+    service_env: Option<String>,
+    service_version: Option<String>,
+    container_id: Option<String>,
+}
+
+impl ConcentratorKey {
+    fn from_metadata(metadata: &TracerMetadata) -> Self {
+        Self {
+            tracer_language: metadata.tracer_language.clone(),
+            service_name: metadata.service_name.clone(),
+            service_env: metadata.service_env.clone(),
+            service_version: metadata.service_version.clone(),
+            container_id: metadata.container_id.clone(),
+        }
+    }
+}
+
 pub struct StatsConcentratorService {
-    /// One concentrator per unique TracerMetadata.
-    concentrators: HashMap<Arc<TracerMetadata>, SpanConcentrator>,
+    /// One concentrator per unique TracerMetadata. Keyed by the projected
+    /// fields, with the originating metadata kept alongside the concentrator
+    /// so flushed payloads can carry the full tracer context.
+    concentrators: HashMap<ConcentratorKey, (Arc<TracerMetadata>, SpanConcentrator)>,
     rx: mpsc::UnboundedReceiver<ConcentratorCommand>,
     config: Arc<Config>,
 }
@@ -99,13 +127,14 @@ impl StatsConcentratorService {
                     // the base service and omits the version. A separate concentrator is kept per
                     // unique metadata so that each payload's stats are flushed with the metadata
                     // from the originating payload.
-                    let concentrator = self
+                    let key = ConcentratorKey::from_metadata(&metadata);
+                    let entry = self
                         .concentrators
-                        .entry(Arc::clone(&metadata))
-                        .or_insert_with(new_concentrator);
+                        .entry(key)
+                        .or_insert_with(|| (Arc::clone(&metadata), new_concentrator()));
 
                     for span in &chunk.spans {
-                        concentrator.add_span(span);
+                        entry.1.add_span(span);
                     }
                 }
                 ConcentratorCommand::Flush(force_flush, response_tx) => {
@@ -123,7 +152,7 @@ impl StatsConcentratorService {
         let payloads = self
             .concentrators
             .iter_mut()
-            .filter_map(|(metadata, concentrator)| {
+            .filter_map(|(_key, (metadata, concentrator))| {
                 let stats_buckets = concentrator.flush(SystemTime::now(), force_flush);
                 if stats_buckets.is_empty() {
                     return None;