diff --git a/Cargo.lock b/Cargo.lock index b69d0b47..5aa7b6bf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -917,7 +917,7 @@ dependencies = [ [[package]] name = "libdd-capabilities" version = "2.1.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "anyhow", "bytes", @@ -941,7 +941,7 @@ dependencies = [ [[package]] name = "libdd-capabilities-impl" version = "3.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "anyhow", "bytes", @@ -990,7 +990,7 @@ dependencies = [ [[package]] name = "libdd-common" version = "5.1.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "anyhow", "bytes", @@ -1053,7 +1053,7 @@ dependencies = [ [[package]] name = "libdd-data-pipeline" version = "7.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "anyhow", "arc-swap", @@ -1067,13 +1067,13 @@ dependencies = [ "libdd-capabilities 2.1.0", "libdd-capabilities-impl 3.0.0", "libdd-common 5.1.0", - "libdd-ddsketch 1.1.0", + "libdd-ddsketch 1.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec)", "libdd-dogstatsd-client", "libdd-shared-runtime 2.0.0", "libdd-telemetry 6.0.0", "libdd-tinybytes", "libdd-trace-normalization", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec)", "libdd-trace-stats", "libdd-trace-utils", "rmp-serde", @@ -1095,6 +1095,14 @@ dependencies = [ "prost", ] +[[package]] +name = "libdd-ddsketch" +version = "1.1.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" +dependencies = [ + "prost", +] + [[package]] name = "libdd-ddsketch" version = "1.1.0" @@ -1106,13 +1114,16 @@ dependencies = [ [[package]] name = "libdd-dogstatsd-client" version = "4.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "anyhow", + "async-trait", "cadence", "http", "libdd-common 5.1.0", + "libdd-shared-runtime 2.0.0", "serde", + "tokio", "tracing", ] @@ -1164,7 +1175,7 @@ dependencies = [ [[package]] name = "libdd-shared-runtime" version = "2.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "async-trait", "futures", @@ -1208,7 +1219,7 @@ dependencies = [ [[package]] name = "libdd-telemetry" version = "6.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "anyhow", "async-trait", @@ -1221,7 +1232,7 @@ dependencies = [ "libc", "libdd-capabilities 2.1.0", "libdd-common 5.1.0", - "libdd-ddsketch 1.1.0", + "libdd-ddsketch 1.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec)", "libdd-shared-runtime 2.0.0", "serde", "serde_json", @@ -1237,7 +1248,7 @@ dependencies = [ [[package]] name = "libdd-tinybytes" version = "1.1.1" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "serde", ] @@ -1245,21 +1256,21 @@ dependencies = [ [[package]] name = "libdd-trace-normalization" version = "3.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "anyhow", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec)", ] [[package]] name = "libdd-trace-obfuscation" version = "5.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "anyhow", "fluent-uri", "libdd-common 5.1.0", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec)", "libdd-trace-utils", "log", "percent-encoding", @@ -1280,7 +1291,7 @@ dependencies = [ [[package]] name = "libdd-trace-protobuf" version = "4.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "prost", "serde", @@ -1290,7 +1301,7 @@ dependencies = [ [[package]] name = "libdd-trace-stats" version = "6.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "anyhow", "arc-swap", @@ -1301,12 +1312,12 @@ dependencies = [ "libdd-capabilities 2.1.0", "libdd-capabilities-impl 3.0.0", "libdd-common 5.1.0", - "libdd-ddsketch 1.1.0", + "libdd-ddsketch 1.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec)", "libdd-dogstatsd-client", "libdd-shared-runtime 2.0.0", "libdd-telemetry 6.0.0", "libdd-trace-obfuscation", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec)", "libdd-trace-utils", "rmp-serde", "serde", @@ -1319,7 +1330,7 @@ dependencies = [ [[package]] name = "libdd-trace-utils" version = "9.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f#3081603d3c74f209be4e3be951f78a1a7469397f" +source = "git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec#15899dfe754d12186ce7db72f0ff41c1920d52ec" dependencies = [ "anyhow", "base64", @@ -1337,7 +1348,7 @@ dependencies = [ "libdd-common 5.1.0", "libdd-tinybytes", "libdd-trace-normalization", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", + "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=15899dfe754d12186ce7db72f0ff41c1920d52ec)", "prost", "rand", "rmp", @@ -1818,27 +1829,21 @@ checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" name = "pipeline" version = "0.1.0" dependencies = [ - "bytes", "console_error_panic_hook", "getrandom 0.2.17", - "http", "js-sys", "libdatadog-nodejs-capabilities", "libdd-capabilities 2.1.0", "libdd-common 5.1.0", "libdd-data-pipeline", "libdd-shared-runtime 2.0.0", - "libdd-trace-protobuf 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", - "libdd-trace-stats", "libdd-trace-utils", - "rmp-serde", "serde", "serde_json", "uuid", "wasm-bindgen", "wasm-bindgen-futures", "wasm-bindgen-test", - "web-time", ] [[package]] @@ -2368,7 +2373,7 @@ checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214" name = "sketches" version = "0.1.0" dependencies = [ - "libdd-ddsketch 1.1.0", + "libdd-ddsketch 1.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=3081603d3c74f209be4e3be951f78a1a7469397f)", "wasm-bindgen", ] diff --git a/crates/capabilities/Cargo.toml b/crates/capabilities/Cargo.toml index edb40adc..371b3f4c 100644 --- a/crates/capabilities/Cargo.toml +++ b/crates/capabilities/Cargo.toml @@ -15,7 +15,7 @@ http = "1" bytes = "1.4" futures-core = "0.3" anyhow = "1" -libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f" } +libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", rev = "15899dfe754d12186ce7db72f0ff41c1920d52ec" } [dev-dependencies] wasm-bindgen-test = "0.3" diff --git a/crates/pipeline/Cargo.toml b/crates/pipeline/Cargo.toml index 8056816e..81f1c15b 100644 --- a/crates/pipeline/Cargo.toml +++ b/crates/pipeline/Cargo.toml @@ -14,17 +14,13 @@ js-sys = "0.3" serde = { version = "1.0", features = ["derive"] } serde_json = "1" libdatadog-nodejs-capabilities = { path = "../capabilities" } -libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f" } -libdd-common = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false } -libdd-data-pipeline = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false, features = ["telemetry"] } -libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false, features = ["change-buffer"] } -libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false } -libdd-trace-protobuf = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false } -libdd-shared-runtime = { git = "https://github.com/DataDog/libdatadog.git", rev = "3081603d3c74f209be4e3be951f78a1a7469397f", default-features = false } -rmp-serde = "1" -bytes = "1" -http = "1" -web-time = "1" +# TODO: Replace these temporary libdatadog PR revs with the official release/tag +# that contains DataDog/libdatadog#2235, then regenerate Cargo.lock. +libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", rev = "15899dfe754d12186ce7db72f0ff41c1920d52ec" } +libdd-common = { git = "https://github.com/DataDog/libdatadog.git", rev = "15899dfe754d12186ce7db72f0ff41c1920d52ec", default-features = false } +libdd-data-pipeline = { git = "https://github.com/DataDog/libdatadog.git", rev = "15899dfe754d12186ce7db72f0ff41c1920d52ec", default-features = false, features = ["telemetry"] } +libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog.git", rev = "15899dfe754d12186ce7db72f0ff41c1920d52ec", default-features = false, features = ["change-buffer"] } +libdd-shared-runtime = { git = "https://github.com/DataDog/libdatadog.git", rev = "15899dfe754d12186ce7db72f0ff41c1920d52ec", default-features = false } console_error_panic_hook = "0.1" [target.'cfg(target_arch = "wasm32")'.dependencies] diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index eb25866e..b3108f15 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -18,8 +18,6 @@ mod span_bytes; mod trace_data; use trace_data::*; -mod stats; - use libdd_trace_utils::change_buffer::{ChangeBuffer, ChangeBufferState}; use libdd_trace_utils::span::v04::{AttributeAnyValue, AttributeArrayValue, SpanEvent}; use span_string::SpanString; @@ -170,7 +168,6 @@ pub struct WasmSpanState { exporter: UnsafeCell>>, builder: UnsafeCell>>, cbs: RefCell>, - stats_collector: RefCell>, /// Chunks staged by `prepareChunk`, one per trace (segment), sent together by /// `sendPreparedChunk` as a single multi-trace request. The exporter groups a /// flush batch by trace and calls `prepareChunk` once per trace so each chunk @@ -224,21 +221,6 @@ impl Drop for InFlightGuard<'_> { } } -fn stats_flush_result(sent: bool, collapsed_spans: u64) -> Result { - let result = js_sys::Object::new(); - js_sys::Reflect::set( - &result, - &JsValue::from_str("sent"), - &JsValue::from_bool(sent), - )?; - js_sys::Reflect::set( - &result, - &JsValue::from_str("collapsedSpans"), - &JsValue::from_f64(collapsed_spans as f64), - )?; - Ok(result.into()) -} - #[wasm_bindgen] impl WasmSpanState { #[wasm_bindgen(constructor)] @@ -267,10 +249,7 @@ impl WasmSpanState { .set_language_version(lang_version) .set_language_interpreter(lang_interpreter) .set_otlp_instrumentation_scope("dd-trace-js", tracer_version) - // Populate the payload-level TracerMetadata (service/env/hostname/ - // app_version) the agent receives. These values are already passed - // in for the stats collector; without these calls the trace - // payload's tracer metadata is sent empty. + // Without these setters the payload-level TracerMetadata is empty. .set_service(tracer_service) .set_env(env) .set_hostname(hostname) @@ -278,16 +257,14 @@ impl WasmSpanState { .set_runtime_id(runtime_id) .enable_agent_rates_payload_version(); - // Advertise `Datadog-Client-Computed-Stats` so the agent skips its own - // APM stats/sampling for these traces. This is required in two cases: - // - `stats_enabled`: we build a StatsCollector and send client-side - // stats, so the agent MUST NOT also compute them (double counting); - // - `client_computed_stats`: set independently for APM-standalone - // (apmTracingEnabled=false), where the agent should skip APM stats - // even though we don't compute them client-side. - // Enabling stats therefore always implies the header, so OR the flags - // rather than relying on the caller to keep them in sync. - if client_computed_stats || stats_enabled { + // `enable_stats` makes libdatadog stamp the client-computed-stats + // header itself when stats run, so the two flags are disjoint: + // `set_client_computed_stats` is only for APM-standalone + // (`apmTracingEnabled=false`), where the header is advertised + // without any stats actually being computed. + if stats_enabled { + builder.enable_stats(Duration::from_secs(10)); + } else if client_computed_stats { builder.set_client_computed_stats(); } @@ -311,31 +288,12 @@ impl WasmSpanState { pid, ); - let stats_collector = if stats_enabled { - Some(stats::StatsCollector::new( - Duration::from_secs(10), - url.to_string(), - stats::StatsMeta { - hostname: hostname.to_string(), - env: env.to_string(), - version: app_version.to_string(), - lang: lang.to_string(), - tracer_version: tracer_version.to_string(), - runtime_id: runtime_id.to_string(), - service: tracer_service.to_string(), - }, - )) - } else { - None - }; - Ok(WasmSpanState { change_queue, string_table_input: vec![0u8; string_table_input_size as usize], exporter: UnsafeCell::new(None), builder: UnsafeCell::new(Some(builder)), cbs: RefCell::new(change_buffer_state), - stats_collector: RefCell::new(stats_collector), prepared_spans: RefCell::new(Vec::new()), sending: Cell::new(false), use_v05: Cell::new(false), @@ -439,8 +397,8 @@ impl WasmSpanState { self.string_table_input.len() as u32 } - /// Prepare a chunk of spans for sending. Flushes the change buffer, - /// extracts spans, feeds stats. Returns `true` if a chunk was prepared + /// Prepare a chunk of spans for sending. Flushes the change buffer and + /// extracts spans. Returns `true` if a chunk was prepared /// (there are spans to send) and `false` if there was nothing to send. /// Must be followed by `sendPreparedChunk()` to actually send. #[wasm_bindgen(js_name = "prepareChunk")] @@ -486,10 +444,6 @@ impl WasmSpanState { .flush_chunk(&span_ids, first_is_local_root) .map_err(|e| JsValue::from_str(&e.to_string()))?; - if let Some(collector) = self.stats_collector.borrow_mut().as_mut() { - collector.add_spans(&spans_vec); - } - // Stage this trace's chunk for the subsequent sendPreparedChunk call. // Multiple prepareChunk calls (one per trace) accumulate here and are // sent together as one multi-trace request. An empty result (e.g. every @@ -590,43 +544,6 @@ impl WasmSpanState { .map_err(|e| JsValue::from_str(&format!("{:?}", e))) } - /// Flush aggregated stats to the agent's /v0.6/stats endpoint. - /// - /// Should be called periodically (e.g. every 10s) from JS, and with - /// `force=true` on shutdown. Returns `{ sent, collapsedSpans }` so JS can - /// emit the same collapsed-span health metric as libdd-trace-stats's native - /// exporter. - #[wasm_bindgen(js_name = "flushStats")] - pub async fn flush_stats(&self, force: bool) -> Result { - // Build the stats request under a brief *synchronous* borrow, then drop - // the borrow BEFORE the async send. The collector therefore stays in - // `stats_collector`, so a concurrent `prepareChunk` during the in-flight - // send still reaches `add_spans` and those spans are counted. (Taking - // the collector out for the whole await would silently drop them from - // client-side stats.) No borrow is held across the await, so there is - // no double-borrow hazard from overlapping calls. - let prepared = { - let mut guard = self.stats_collector.borrow_mut(); - match guard.as_mut() { - Some(collector) => collector - .prepare_request(force) - .map_err(|e| JsValue::from_str(&e))?, - None => return stats_flush_result(false, 0), - } - }; - let sent = match prepared.request { - Some(req) => { - stats::StatsCollector::send_request(req) - .await - .map_err(|e| JsValue::from_str(&e))?; - true - } - None => false, - }; - - stats_flush_result(sent, prepared.collapsed_spans) - } - /// Flush the queued change-buffer operations. On success always returns /// `true` (the bool exists only for signature symmetry with the other /// flush methods); failures surface as a thrown error. @@ -639,6 +556,35 @@ impl WasmSpanState { Ok(true) } + /// Gracefully shut down the exporter and stop the background workers. + /// Should be awaited during tracer shutdown to avoid losing in-flight + /// data. Consumes the exporter — subsequent sends will error. + /// + /// `timeoutMs` bounds the wait; on timeout returns an error and workers + /// may still be finishing. `None` waits indefinitely. + #[wasm_bindgen(js_name = "shutdown")] + pub async fn shutdown(&self, timeout_ms: Option) -> Result<(), JsValue> { + // `sendPreparedChunk` holds an `&mut` on the exporter across awaits; + // taking it out from under an in-flight send would alias. + if self.sending.get() { + return Err(JsValue::from_str("shutdown: sendPreparedChunk in flight")); + } + self.sending.set(true); + let _in_flight = InFlightGuard(&self.sending); + + // SAFETY: `sending` guard prevents overlapping access; WASM is single-threaded. + let exporter_slot = unsafe { &mut *self.exporter.get() }; + // Idempotent: never-built or already-shut-down state is not an error. + let Some(exporter) = exporter_slot.take() else { + return Ok(()); + }; + let timeout = timeout_ms.map(|ms| Duration::from_millis(u64::from(ms))); + exporter + .shutdown_async(timeout) + .await + .map_err(|e| JsValue::from_str(&format!("shutdown: {e:?}"))) + } + /// Set default meta tags applied to every new span. /// Takes a flat array of key-value pairs: [key1, val1, key2, val2, ...] #[wasm_bindgen(js_name = "setDefaultMeta")] diff --git a/crates/pipeline/src/stats.rs b/crates/pipeline/src/stats.rs deleted file mode 100644 index 3e65cb46..00000000 --- a/crates/pipeline/src/stats.rs +++ /dev/null @@ -1,201 +0,0 @@ -// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ -// SPDX-License-Identifier: Apache-2.0 - -//! Native stats collection for the pipeline WASM module. -//! -//! Wraps `SpanConcentrator` from `libdd-trace-stats` and provides encoding + -//! HTTP transport for flushing stats to the Datadog agent's `/v0.6/stats` -//! endpoint. - -use web_time::{Duration, SystemTime}; - -/// Wall-clock now() for wasm. Delegates to `web_time::SystemTime::now()`, -/// which routes to JS `Date.now()` on `wasm32-unknown-unknown` (native -/// `std::time::SystemTime::now()` is unimplemented on that target and traps). -fn now() -> SystemTime { - SystemTime::now() -} - -use bytes::Bytes; -use libdatadog_nodejs_capabilities::WasmHttpClient; -use libdd_capabilities::http::HttpClientCapability; -use libdd_common::parse_uri; -use libdd_trace_protobuf::pb; -use libdd_trace_stats::span_concentrator::SpanConcentrator; - -use crate::trace_data::WasmTraceData; - -const STATS_ENDPOINT_PATH: &str = "/v0.6/stats"; - -/// Metadata for the stats payload envelope. -pub struct StatsMeta { - pub hostname: String, - pub env: String, - pub version: String, - pub lang: String, - pub tracer_version: String, - pub runtime_id: String, - pub service: String, -} - -/// Stats data prepared by a synchronous concentrator flush. -pub struct PreparedStatsFlush { - pub request: Option>, - pub collapsed_spans: u64, -} - -/// Manages stats aggregation and flushing. -pub struct StatsCollector { - concentrator: SpanConcentrator, - meta: StatsMeta, - agent_url: String, - sequence: u64, -} - -impl StatsCollector { - /// Create a new stats collector. - pub fn new(bucket_size: Duration, agent_url: String, meta: StatsMeta) -> Self { - StatsCollector { - concentrator: SpanConcentrator::new( - bucket_size, - now(), - vec![ - "client".to_string(), - "server".to_string(), - "producer".to_string(), - "consumer".to_string(), - ], - Vec::new(), - None, - Vec::new(), - ), - meta, - agent_url, - sequence: 0, - } - } - - /// Add spans to the concentrator for stats aggregation. - /// - /// The spans should already have `_dd.top_level` and `_dd.measured` metrics - /// set (done by `ChangeBufferState::flush_chunk`). - pub fn add_spans(&mut self, spans: &[libdd_trace_utils::span::v04::Span]) { - for span in spans { - self.concentrator.add_span(span); - } - } - - /// Drain aggregated stats into a ready-to-send request plus flush metadata, - /// **synchronously**. - /// - /// Returns `request: None` when there is no stats payload to send. The - /// concentrator is drained and the sequence advanced as part of this call, - /// so a returned request must be sent (see `send_request`). Kept - /// synchronous and separate from the send so a caller can build the request - /// under a brief borrow and release the collector *before* the async send — - /// leaving it available for `add_spans` while the stats request is in - /// flight. - pub fn prepare_request(&mut self, force: bool) -> Result { - let mut flush = self.concentrator.flush(now(), force); - let collapsed_spans = flush.collapsed_spans; - if !flush.obfuscated_buckets.is_empty() { - // TODO: stats obfuscation is currently disabled. Obfuscated stats - // require the datadog-obfuscation-version header, which - // prepare_request doesn't emit yet. Add that header before enabling - // stats-obfuscation. - return Err( - "stats flush produced obfuscated buckets without obfuscation header support" - .to_string(), - ); - } - if flush.unobfuscated_buckets.is_empty() { - return Ok(PreparedStatsFlush { - request: None, - collapsed_spans, - }); - } - - self.sequence += 1; - let buckets = std::mem::take(&mut flush.unobfuscated_buckets); - let payload = encode_stats_payload(&buckets, &self.meta, self.sequence); - - let body = rmp_serde::encode::to_vec_named(&payload) - .map_err(|e| format!("stats msgpack encode error: {e}"))?; - - // Build the base agent URI exactly like the trace exporter does, via - // libdatadog's `parse_uri`. For a `unix://` / `windows:` agent URL that - // hex-encodes the socket path into the URI *authority* (there is no - // standard URL form for socket paths), which the WASM HTTP client's - // `decode_socket_path` reverses to route over the socket. A raw parse - // instead leaves the socket path in the URI *path* with an empty/invalid - // authority, so the stats request never reaches the socket — client stats - // silently never arrive over UDS (dd-trace-js #9139, uds-express4). - let base = parse_uri(&self.agent_url).map_err(|e| format!("invalid agent URL: {e}"))?; - // Append `/v0.6/stats` to the base path while preserving the (hex) - // authority, mirroring libdd-data-pipeline's `add_path`. For `unix://` - // the base path is "/" and the authority holds the hex socket path; for - // TCP it's `http://host:port/`. Trim a trailing slash so the path is - // exactly `/v0.6/stats` (a double slash makes the agent miss the request). - let base_path = base.path().strip_suffix('/').unwrap_or_else(|| base.path()); - let new_path_and_query = format!("{base_path}{STATS_ENDPOINT_PATH}"); - let mut parts = base.into_parts(); - parts.path_and_query = Some( - new_path_and_query - .parse() - .map_err(|e| format!("invalid stats path: {e}"))?, - ); - let uri = http::Uri::from_parts(parts).map_err(|e| format!("invalid stats URL: {e}"))?; - - let req = http::Request::builder() - .method(http::Method::PUT) - .uri(uri) - .header("Content-Type", "application/msgpack") - .header("Datadog-Meta-Lang", &self.meta.lang) - .header("Datadog-Meta-Tracer-Version", &self.meta.tracer_version) - .body(Bytes::from(body)) - .map_err(|e| format!("failed to build stats request: {e}"))?; - - Ok(PreparedStatsFlush { - request: Some(req), - collapsed_spans, - }) - } - - /// Send a prepared stats request to the agent. Does **not** borrow the - /// collector, so trace export (`add_spans`) can proceed during the await. - pub async fn send_request(req: http::Request) -> Result<(), String> { - let client = WasmHttpClient::new_client(); - client - .request(req) - .await - .map_err(|e| format!("stats send error: {e:?}"))?; - Ok(()) - } -} - -/// Encode flushed stats buckets into a `ClientStatsPayload` for msgpack -/// serialization. -fn encode_stats_payload( - buckets: &[pb::ClientStatsBucket], - meta: &StatsMeta, - sequence: u64, -) -> pb::ClientStatsPayload { - pb::ClientStatsPayload { - hostname: meta.hostname.clone(), - env: meta.env.clone(), - version: meta.version.clone(), - lang: meta.lang.clone(), - tracer_version: meta.tracer_version.clone(), - runtime_id: meta.runtime_id.clone(), - sequence, - stats: buckets.to_vec(), - service: meta.service.clone(), - container_id: String::new(), - tags: Vec::new(), - agent_aggregation: String::new(), - git_commit_sha: String::new(), - image_tag: String::new(), - process_tags: String::new(), - process_tags_hash: 0, - } -} diff --git a/test/pipeline.js b/test/pipeline.js index 59511fef..08032c07 100644 --- a/test/pipeline.js +++ b/test/pipeline.js @@ -1080,172 +1080,6 @@ describe('pipeline', { skip }, () => { assert.ok(sawTraces, 'expected a POST to /v0.4/traces') assert.strictEqual(header, undefined) }) - - it('sends the header when stats are enabled (client-side stats imply it)', async () => { - // Enabling client-side stats without clientComputedStats must still send - // the header, otherwise the agent double-counts APM stats. - const { header, sawTraces } = await captureTraceHeader({ statsEnabled: true, clientComputedStats: false }) - assert.ok(sawTraces, 'expected a POST to /v0.4/traces') - assert.strictEqual(header, 'true') - }) - }) - - describe('client-side stats', () => { - it('aggregates and flushes stats to /v0.6/stats', async () => { - const http = require('node:http') - const seen = [] - const server = http.createServer((req, res) => { - const chunks = [] - req.on('data', c => chunks.push(c)) - req.on('end', () => { - seen.push({ method: req.method, url: req.url, len: Buffer.concat(chunks).length }) - res.writeHead(200, { 'content-type': 'application/json' }) - res.end('{}') - }) - }) - await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) - const { port } = server.address() - - // statsEnabled:true builds the StatsCollector; prepareChunk feeds spans - // into it, and flushStats(true) force-flushes to /v0.6/stats. - const ns = new NativeSpansInterface({ agentUrl: `http://127.0.0.1:${port}`, statsEnabled: true }) - const span = ns.createSpan() - span.name = 'stats-span' - span.service = 'stats-svc' - span.resource = '/stats' - span.type = 'web' - span.duration = 5_000_000n - - try { - await ns.flushSpans(span) - const result = await ns.state.flushStats(true) - assert.deepStrictEqual(result, { sent: true, collapsedSpans: 0 }, 'flushStats reported a send') - const statsReq = seen.find(r => r.url === '/v0.6/stats') - assert.ok(statsReq, 'agent received a /v0.6/stats request') - assert.strictEqual(statsReq.method, 'PUT') - assert.ok(statsReq.len > 0, 'stats payload is non-empty') - - // Nothing new aggregated -> a second forced flush is a no-op. - assert.deepStrictEqual(await ns.state.flushStats(true), { sent: false, collapsedSpans: 0 }, 'second flush has nothing to send') - } finally { - server.closeAllConnections?.() - server.close() - } - }) - - it('returns collapsed span count when stats cardinality overflows', async () => { - const http = require('node:http') - const seen = [] - const server = http.createServer((req, res) => { - const chunks = [] - req.on('data', c => chunks.push(c)) - req.on('end', () => { - seen.push({ method: req.method, url: req.url, len: Buffer.concat(chunks).length }) - res.writeHead(200, { 'content-type': 'application/json' }) - res.end('{}') - }) - }) - await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) - const { port } = server.address() - - const ns = new NativeSpansInterface({ agentUrl: `http://127.0.0.1:${port}`, statsEnabled: true }) - let batch = [] - - try { - for (let i = 0; i < 15_000; i++) { - const span = ns.createSpan() - span.name = 'stats-span' - span.service = 'stats-svc' - span.resource = `/stats/${i}` - span.type = 'web' - span.setTag('span.kind', 'server') - span.duration = 5_000_000n - ns.flushChangeQueue() - batch.push(span) - if (batch.length === 500) { - await ns.flushSpans(...batch) - batch = [] - } - } - if (batch.length > 0) { - await ns.flushSpans(...batch) - } - - const result = await ns.state.flushStats(true) - assert.strictEqual(result.sent, true) - assert.ok(result.collapsedSpans > 0, 'stats cardinality overflow reported collapsed spans') - assert.ok(seen.some(r => r.url === '/v0.6/stats'), 'agent received a /v0.6/stats request') - } finally { - server.closeAllConnections?.() - server.close() - } - }) - - it('flushStats reports no send when stats are disabled', async () => { - const ns = new NativeSpansInterface({ statsEnabled: false }) - assert.deepStrictEqual(await ns.state.flushStats(true), { sent: false, collapsedSpans: 0 }) - }) - - it('flushes stats to /v0.6/stats over a Unix domain socket', { skip: process.platform === 'win32' }, async () => { - // A `unix://` agent URL must route /v0.6/stats over the socket, like - // traces do. parse_uri hex-encodes the socket path into the URI authority - // (which the transport's decode_socket_path reverses); a raw parse would - // leave the path in the URI path and never reach the socket. - const http = require('node:http') - const os = require('node:os') - const fs = require('node:fs') - const nodePath = require('node:path') - // Keep the path short — AF_UNIX paths are capped (~104 bytes on macOS). - const sockPath = nodePath.join(os.tmpdir(), `dd-st-${process.pid}.sock`) - try { - fs.unlinkSync(sockPath) - } catch { - // not present - } - const seen = [] - const server = http.createServer((req, res) => { - const chunks = [] - req.on('data', c => chunks.push(c)) - req.on('end', () => { - seen.push({ method: req.method, url: req.url, len: Buffer.concat(chunks).length }) - res.writeHead(200, { 'content-type': 'application/json' }) - res.end('{}') - }) - }) - await new Promise((resolve, reject) => { - server.once('error', reject) - server.listen(sockPath, resolve) - }) - - const ns = new NativeSpansInterface({ agentUrl: `unix://${sockPath}`, statsEnabled: true }) - const span = ns.createSpan() - span.name = 'stats-span' - span.service = 'stats-svc' - span.resource = '/stats' - span.type = 'web' - span.duration = 5_000_000n - - try { - await ns.flushSpans(span) - const result = await ns.state.flushStats(true) - assert.deepStrictEqual( - result, - { sent: true, collapsedSpans: 0 }, - 'flushStats reported a send over the socket', - ) - const statsReq = seen.find(r => r.url === '/v0.6/stats') - assert.ok(statsReq, 'agent received a /v0.6/stats request over the socket') - assert.ok(statsReq.len > 0, 'stats payload is non-empty') - } finally { - server.closeAllConnections?.() - server.close() - try { - fs.unlinkSync(sockPath) - } catch { - // already gone - } - } - }) }) describe('send re-entrancy', () => {