feat: add trace stats computation

LICENSE-3rdparty.csv

@@ -123,6 +123,8 @@ libdd-common,https://github.com/DataDog/libdatadog/tree/main/datadog-common,Apac
 libdd-data-pipeline,https://github.com/DataDog/libdatadog/tree/main/libdd-data-pipeline,Apache-2.0,The libdd-data-pipeline Authors
 libdd-ddsketch,https://github.com/DataDog/libdatadog/tree/main/libdd-ddsketch,Apache-2.0,The libdd-ddsketch Authors
 libdd-dogstatsd-client,https://github.com/DataDog/libdatadog/tree/main/libdd-dogstatsd-client,Apache-2.0,The libdd-dogstatsd-client Authors
+libdd-library-config,https://github.com/DataDog/libdatadog/tree/main/libdd-library-config,Apache-2.0,The libdd-library-config Authors
+libdd-shared-runtime,https://github.com/DataDog/libdatadog/tree/main/libdd-shared-runtime,Apache-2.0,The libdd-shared-runtime Authors
 libdd-telemetry,https://github.com/DataDog/libdatadog/tree/main/libdd-telemetry,Apache-2.0,The libdd-telemetry Authors
 libdd-tinybytes,https://github.com/DataDog/libdatadog/tree/main/libdd-tinybytes,Apache-2.0,The libdd-tinybytes Authors
 libdd-trace-normalization,https://github.com/DataDog/libdatadog/tree/main/libdd-trace-normalization,Apache-2.0,David Lee <david.lee@datadoghq.com>
@@ -139,6 +141,7 @@ lru,https://github.com/jeromefroe/lru-rs,MIT,Jerome Froelich <jeromefroelic@hotm
 lru-slab,https://github.com/Ralith/lru-slab,MIT OR Apache-2.0 OR Zlib,Benjamin Saunders <ben.e.saunders@gmail.com>
 matchers,https://github.com/hawkw/matchers,MIT,Eliza Weisman <eliza@buoyant.io>
 memchr,https://github.com/BurntSushi/memchr,Unlicense OR MIT,"Andrew Gallant <jamslam@gmail.com>, bluss"
+memfd,https://github.com/lucab/memfd-rs,MIT OR Apache-2.0,"Luca Bruno <lucab@lucabruno.net>, Simonas Kazlauskas <memfd@kazlauskas.me>"
 mime,https://github.com/hyperium/mime,MIT OR Apache-2.0,Sean McArthur <sean@seanmonstar.com>
 minimal-lexical,https://github.com/Alexhuszagh/minimal-lexical,MIT OR Apache-2.0,Alex Huszagh <ahuszagh@gmail.com>
 miniz_oxide,https://github.com/Frommi/miniz_oxide/tree/master/miniz_oxide,MIT OR Zlib OR Apache-2.0,"Frommi <daniil.liferenko@gmail.com>, oyvindln <oyvindln@users.noreply.github.com>, Rich Geldreich richgel99@gmail.com"

crates/datadog-agent-config/Cargo.toml

@@ -6,8 +6,8 @@ license.workspace = true
 
 [dependencies]
 figment = { version = "0.10", default-features = false, features = ["yaml", "env"] }
-libdd-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "27aa92cfeeca073d8730a8b4974bd3fdef7ddf3a" }
-libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "27aa92cfeeca073d8730a8b4974bd3fdef7ddf3a" }
+libdd-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "e4d93e5274f176e110dbacd745e922268ac31ccd" }
+libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "e4d93e5274f176e110dbacd745e922268ac31ccd" }
 log = { version = "0.4", default-features = false }
 serde = { version = "1.0", default-features = false, features = ["derive"] }
 serde-aux = { version = "4.7", default-features = false }

crates/datadog-serverless-compat/Cargo.toml

@@ -12,7 +12,7 @@ windows-pipes = ["datadog-trace-agent/windows-pipes", "dogstatsd/windows-pipes"]
 [dependencies]
 datadog-logs-agent = { path = "../datadog-logs-agent" }
 datadog-trace-agent = { path = "../datadog-trace-agent" }
-libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "27aa92cfeeca073d8730a8b4974bd3fdef7ddf3a" }
+libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "e4d93e5274f176e110dbacd745e922268ac31ccd" }
 datadog-fips = { path = "../datadog-fips", default-features = false }
 dogstatsd = { path = "../dogstatsd", default-features = true }
 reqwest = { version = "0.12.4", default-features = false }

crates/datadog-serverless-compat/src/main.rs

@@ -18,7 +18,8 @@ use zstd::zstd_safe::CompressionLevel;
 
 use datadog_trace_agent::{
     aggregator::TraceAggregator,
-    config, env_verifier, mini_agent, proxy_flusher, stats_flusher, stats_processor,
+    config, env_verifier, mini_agent, proxy_flusher, stats_concentrator_service, stats_flusher,
+    stats_processor,
     trace_flusher::{self, TraceFlusher},
     trace_processor,
 };
@@ -49,6 +50,11 @@ const DEFAULT_DOGSTATSD_PORT: u16 = 8125;
 const DEFAULT_LOG_INTAKE_PORT: u16 = 10517;
 const AGENT_HOST: &str = "0.0.0.0";
 
+struct StatsConcentratorComponents {
+    handle: stats_concentrator_service::StatsConcentratorHandle,
+    service_handle: tokio::task::JoinHandle<()>,
+}
+
 #[tokio::main]
 pub async fn main() {
     let log_level = env::var("DD_LOG_LEVEL")
@@ -119,6 +125,11 @@ pub async fn main() {
         .ok()
         .and_then(|v| v.parse::<u16>().ok())
         .unwrap_or(DEFAULT_LOG_INTAKE_PORT);
+
+    let dd_agent_stats_computation_enabled = env::var("DD_AGENT_STATS_COMPUTATION_ENABLED")
+        .map(|val| val.to_lowercase() != "false")
+        .unwrap_or(true);
+
     debug!("Starting serverless trace mini agent");
 
     let env_filter = format!("h2=off,hyper=off,rustls=off,{}", log_level);
@@ -144,11 +155,6 @@ pub async fn main() {
 
     let env_verifier = Arc::new(env_verifier::ServerlessEnvVerifier::default());
 
-    let trace_processor = Arc::new(trace_processor::ServerlessTraceProcessor {});
-
-    let stats_flusher = Arc::new(stats_flusher::ServerlessStatsFlusher {});
-    let stats_processor = Arc::new(stats_processor::ServerlessStatsProcessor {});
-
     let config = match config::Config::new() {
         Ok(c) => Arc::new(c),
         Err(e) => {
@@ -157,6 +163,28 @@ pub async fn main() {
         }
     };
 
+    let stats_concentrator = if dd_agent_stats_computation_enabled {
+        info!("agent stats computation enabled");
+        let (service, handle) =
+            stats_concentrator_service::StatsConcentratorService::new(config.clone());
+        Some(StatsConcentratorComponents {
+            service_handle: tokio::spawn(service.run()),
+            handle,
+        })
+    } else {
+        info!("agent stats computation disabled");
+        None
+    };
+
+    let trace_processor = Arc::new(trace_processor::ServerlessTraceProcessor {
+        stats_concentrator: stats_concentrator.as_ref().map(|c| c.handle.clone()),
+    });
+
+    let stats_flusher = Arc::new(stats_flusher::ServerlessStatsFlusher {
+        stats_concentrator: stats_concentrator.as_ref().map(|c| c.handle.clone()),
+    });
+    let stats_processor = Arc::new(stats_processor::ServerlessStatsProcessor {});
+
     let trace_aggregator = Arc::new(TokioMutex::new(TraceAggregator::default()));
     let trace_flusher = Arc::new(trace_flusher::ServerlessTraceFlusher::new(
         trace_aggregator,
@@ -175,8 +203,11 @@ pub async fn main() {
         proxy_flusher,
     });
 
+    let (_shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
     tokio::spawn(async move {
-        let res = mini_agent.start_mini_agent().await;
+        let res = mini_agent
+            .start_mini_agent(shutdown_rx, stats_concentrator.map(|c| c.service_handle))
+            .await;
         if let Err(e) = res {
             error!("Error when starting serverless trace mini agent: {e:?}");
         }