feat: [Trace Stats] Do not set ClientStatsPayload.hostname (#905)

# This PR
As APM Stats team suggested, `ClientStatsPayload.hostname` should be
left empty so the trace stats backend can aggregate stats properly.

# Testing
Still some undercounting. Expecting 5000 for the `highvolume` (with
concurrency) tests, but seeing 4xxx for some runtimes. Will keep
investigating.
<img width="1457" height="381" alt="Screenshot 2025-10-23 at 3 28 04 PM"
src="https://github.com/user-attachments/assets/6bb1e231-49de-4924-a476-826fe1e645ff"
/>
<img width="1453" height="369" alt="Screenshot 2025-10-23 at 3 29 21 PM"
src="https://github.com/user-attachments/assets/91dbd642-a637-41e3-be3f-88f83b4a022c"
/>

# Notes
#688
https://datadoghq.atlassian.net/browse/SVLS-6907

Author: lym953

bottlecap/src/bin/bottlecap/main.rs

@@ -992,7 +992,7 @@ fn start_trace_agent(
 ) {
     // Stats
     let (stats_concentrator_service, stats_concentrator_handle) =
-        StatsConcentratorService::new(Arc::clone(config), tags_provider.clone());
+        StatsConcentratorService::new(Arc::clone(config));
     tokio::spawn(stats_concentrator_service.run());
     let stats_aggregator: Arc<TokioMutex<StatsAggregator>> = Arc::new(TokioMutex::new(
         StatsAggregator::new_with_concentrator(stats_concentrator_handle.clone()),

bottlecap/src/traces/stats_aggregator.rs

@@ -91,22 +91,14 @@ impl StatsAggregator {
 #[allow(clippy::unwrap_used)]
 mod tests {
     use super::*;
-    use crate::LAMBDA_RUNTIME_SLUG;
     use crate::config::Config;
-    use crate::tags::provider::Provider as TagProvider;
     use crate::traces::stats_concentrator_service::StatsConcentratorService;
-    use std::collections::HashMap;
     use std::sync::Arc;
 
     #[test]
     fn test_add() {
         let config = Arc::new(Config::default());
-        let tags_provider = Arc::new(TagProvider::new(
-            config.clone(),
-            LAMBDA_RUNTIME_SLUG.to_string(),
-            &HashMap::new(),
-        ));
-        let (_, concentrator) = StatsConcentratorService::new(config, tags_provider);
+        let (_, concentrator) = StatsConcentratorService::new(config);
         let mut aggregator = StatsAggregator::new_with_concentrator(concentrator);
         let payload = ClientStatsPayload {
             hostname: "hostname".to_string(),
@@ -135,12 +127,7 @@ mod tests {
     #[tokio::test]
     async fn test_get_batch() {
         let config = Arc::new(Config::default());
-        let tags_provider = Arc::new(TagProvider::new(
-            config.clone(),
-            LAMBDA_RUNTIME_SLUG.to_string(),
-            &HashMap::new(),
-        ));
-        let (_, concentrator) = StatsConcentratorService::new(config, tags_provider);
+        let (_, concentrator) = StatsConcentratorService::new(config);
         let mut aggregator = StatsAggregator::new_with_concentrator(concentrator);
         let payload = ClientStatsPayload {
             hostname: "hostname".to_string(),
@@ -169,12 +156,7 @@ mod tests {
     #[tokio::test]
     async fn test_get_batch_full_entries() {
         let config = Arc::new(Config::default());
-        let tags_provider = Arc::new(TagProvider::new(
-            config.clone(),
-            LAMBDA_RUNTIME_SLUG.to_string(),
-            &HashMap::new(),
-        ));
-        let (_, concentrator) = StatsConcentratorService::new(config, tags_provider);
+        let (_, concentrator) = StatsConcentratorService::new(config);
         let mut aggregator = StatsAggregator::new(704, concentrator);
         // Payload below is 352 bytes
         let payload = ClientStatsPayload {

bottlecap/src/traces/stats_concentrator_service.rs

@@ -1,7 +1,6 @@
 use tokio::sync::{mpsc, oneshot};
 
 use crate::config::Config;
-use crate::tags::provider::Provider as TagProvider;
 use datadog_trace_protobuf::pb;
 use datadog_trace_protobuf::pb::{ClientStatsPayload, TracerPayload};
 use datadog_trace_stats::span_concentrator::SpanConcentrator;
@@ -104,18 +103,14 @@ pub struct StatsConcentratorService {
     concentrator: SpanConcentrator,
     rx: mpsc::UnboundedReceiver<ConcentratorCommand>,
     tracer_metadata: TracerMetadata,
-    hostname: String,
     config: Arc<Config>,
 }
 
 // A service that handles add() and flush() requests in the same queue,
 // to avoid using mutex, which may cause lock contention.
 impl StatsConcentratorService {
     #[must_use]
-    pub fn new(
-        config: Arc<Config>,
-        tags_provider: Arc<TagProvider>,
-    ) -> (Self, StatsConcentratorHandle) {
+    pub fn new(config: Arc<Config>) -> (Self, StatsConcentratorHandle) {
         let (tx, rx) = mpsc::unbounded_channel();
         let handle = StatsConcentratorHandle::new(tx);
         // TODO: set span_kinds_stats_computed and peer_tag_keys
@@ -125,13 +120,11 @@ impl StatsConcentratorService {
             vec![],
             vec![],
         );
-        let hostname = tags_provider.get_canonical_id().unwrap_or_default();
         let service: StatsConcentratorService = Self {
             concentrator,
             rx,
             // To be set when the first trace is received
             tracer_metadata: TracerMetadata::default(),
-            hostname,
             config,
         };
         (service, handle)
@@ -161,7 +154,8 @@ impl StatsConcentratorService {
             None
         } else {
             Some(ClientStatsPayload {
-                hostname: self.hostname.clone(),
+                // Do not set hostname so the trace stats backend can aggregate stats properly
+                hostname: String::new(),
                 env: self.config.env.clone().unwrap_or("unknown-env".to_string()),
                 // Version is not in the trace payload. Need to read it from config.
                 version: self.config.version.clone().unwrap_or_default(),