From 11534724b54f8f16a04d96b3e934cef2528f6db4 Mon Sep 17 00:00:00 2001 From: Bryan English Date: Wed, 15 Jul 2026 14:57:40 -0400 Subject: [PATCH 1/2] fix(pipeline): set OTLP instrumentation scope (#173) * fix(pipeline): set OTLP instrumentation scope * fix(pipeline): consume ported libdatadog scope patch Update the temporary libdatadog dependency to the PR head ported onto main. Current libdatadog also changed the stats concentrator API, so adapt the WASM stats wrapper to pass the new cardinality limit argument and encode unobfuscated buckets from FlushResult. * fix(pipeline): return collapsed stats count from flushStats * test(pipeline): widen stats overflow regression --- Cargo.lock | 142 ++++++++++++++++++--------------- crates/capabilities/Cargo.toml | 4 +- crates/pipeline/Cargo.toml | 16 ++-- crates/pipeline/src/lib.rs | 142 +++++++++++++++++++++++---------- crates/pipeline/src/stats.rs | 60 ++++++++++---- test/pipeline.js | 75 +++++++++++++++-- 6 files changed, 299 insertions(+), 140 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 308f24a..70471b0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -34,9 +34,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.102" +version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "arc-swap" @@ -838,7 +838,7 @@ dependencies = [ "futures-core", "http", "js-sys", - "libdd-capabilities 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", + "libdd-capabilities 2.1.0", "wasm-bindgen", "wasm-bindgen-futures", "wasm-bindgen-test", @@ -857,8 +857,8 @@ dependencies = [ [[package]] name = "libdd-capabilities" -version = "2.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "2.1.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "anyhow", "bytes", @@ -874,21 +874,21 @@ dependencies = [ "bytes", "http", "http-body-util", - "libdd-capabilities 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v35.0.0)", + "libdd-capabilities 2.0.0", "libdd-common 4.2.0", "tokio", ] [[package]] name = "libdd-capabilities-impl" -version = "2.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "3.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "bytes", "http", "http-body-util", - "libdd-capabilities 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-common 5.0.0", + "libdd-capabilities 2.1.0", + "libdd-common 5.1.0", "tokio", ] @@ -928,8 +928,8 @@ dependencies = [ [[package]] name = "libdd-common" -version = "5.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "5.1.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "anyhow", "bytes", @@ -991,8 +991,8 @@ dependencies = [ [[package]] name = "libdd-data-pipeline" -version = "6.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "7.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "anyhow", "arc-swap", @@ -1002,15 +1002,15 @@ dependencies = [ "getrandom 0.2.17", "http", "http-body-util", - "libdd-capabilities 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-capabilities-impl 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-common 5.0.0", - "libdd-ddsketch 1.0.1 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", + "libdd-capabilities 2.1.0", + "libdd-capabilities-impl 3.0.0", + "libdd-common 5.1.0", + "libdd-ddsketch 1.1.0", "libdd-dogstatsd-client", - "libdd-shared-runtime 1.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", + "libdd-shared-runtime 2.0.0", "libdd-tinybytes", "libdd-trace-normalization", - "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", + "libdd-trace-protobuf 4.0.0", "libdd-trace-stats", "libdd-trace-utils", "rmp-serde", @@ -1033,21 +1033,21 @@ dependencies = [ [[package]] name = "libdd-ddsketch" -version = "1.0.1" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "1.1.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "prost", ] [[package]] name = "libdd-dogstatsd-client" -version = "3.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "4.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "anyhow", "cadence", "http", - "libdd-common 5.0.0", + "libdd-common 5.1.0", "serde", "tracing", ] @@ -1073,7 +1073,7 @@ source = "git+https://github.com/DataDog/libdatadog.git?rev=7cdeb7896e92d1ba38bd dependencies = [ "anyhow", "libc", - "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?rev=7cdeb7896e92d1ba38bde495934e112dac2eda25)", + "libdd-trace-protobuf 3.0.2", "memfd", "prost", "rand", @@ -1102,8 +1102,8 @@ dependencies = [ "async-trait", "futures", "futures-util", - "libdd-capabilities 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v35.0.0)", - "libdd-capabilities-impl 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v35.0.0)", + "libdd-capabilities 2.0.0", + "libdd-capabilities-impl 2.0.0", "libdd-common 4.2.0", "tokio", "tokio-util", @@ -1113,15 +1113,15 @@ dependencies = [ [[package]] name = "libdd-shared-runtime" -version = "1.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "2.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "async-trait", "futures", "futures-util", - "libdd-capabilities 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-capabilities-impl 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-common 5.0.0", + "libdd-capabilities 2.1.0", + "libdd-capabilities-impl 3.0.0", + "libdd-common 5.1.0", "tokio", "tokio-util", "tracing", @@ -1143,8 +1143,8 @@ dependencies = [ "http-body-util", "libc", "libdd-common 4.2.0", - "libdd-ddsketch 1.0.1 (git+https://github.com/DataDog/libdatadog.git?tag=v35.0.0)", - "libdd-shared-runtime 1.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v35.0.0)", + "libdd-ddsketch 1.0.1", + "libdd-shared-runtime 1.0.0", "serde", "serde_json", "sys-info", @@ -1158,29 +1158,29 @@ dependencies = [ [[package]] name = "libdd-tinybytes" version = "1.1.1" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "serde", ] [[package]] name = "libdd-trace-normalization" -version = "2.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "3.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "anyhow", - "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", + "libdd-trace-protobuf 4.0.0", ] [[package]] name = "libdd-trace-obfuscation" -version = "4.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "5.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "anyhow", "fluent-uri", - "libdd-common 5.0.0", - "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", + "libdd-common 5.1.0", + "libdd-trace-protobuf 4.0.0", "libdd-trace-utils", "log", "percent-encoding", @@ -1191,7 +1191,7 @@ dependencies = [ [[package]] name = "libdd-trace-protobuf" version = "3.0.2" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +source = "git+https://github.com/DataDog/libdatadog.git?rev=7cdeb7896e92d1ba38bde495934e112dac2eda25#7cdeb7896e92d1ba38bde495934e112dac2eda25" dependencies = [ "prost", "serde", @@ -1200,8 +1200,8 @@ dependencies = [ [[package]] name = "libdd-trace-protobuf" -version = "3.0.2" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7cdeb7896e92d1ba38bde495934e112dac2eda25#7cdeb7896e92d1ba38bde495934e112dac2eda25" +version = "4.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "prost", "serde", @@ -1210,21 +1210,23 @@ dependencies = [ [[package]] name = "libdd-trace-stats" -version = "5.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "6.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "anyhow", "arc-swap", "async-trait", + "futures", "hashbrown 0.15.5", "http", - "libdd-capabilities 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-capabilities-impl 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-common 5.0.0", - "libdd-ddsketch 1.0.1 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-shared-runtime 1.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", + "libdd-capabilities 2.1.0", + "libdd-capabilities-impl 3.0.0", + "libdd-common 5.1.0", + "libdd-ddsketch 1.1.0", + "libdd-dogstatsd-client", + "libdd-shared-runtime 2.0.0", "libdd-trace-obfuscation", - "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", + "libdd-trace-protobuf 4.0.0", "libdd-trace-utils", "rmp-serde", "serde", @@ -1235,8 +1237,8 @@ dependencies = [ [[package]] name = "libdd-trace-utils" -version = "8.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0#86b7f5700d3db4e58076794b5e8073a40d780083" +version = "9.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" dependencies = [ "anyhow", "base64", @@ -1248,12 +1250,12 @@ dependencies = [ "http-body", "http-body-util", "indexmap", - "libdd-capabilities 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-capabilities-impl 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-common 5.0.0", + "libdd-capabilities 2.1.0", + "libdd-capabilities-impl 3.0.0", + "libdd-common 5.1.0", "libdd-tinybytes", "libdd-trace-normalization", - "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", + "libdd-trace-protobuf 4.0.0", "prost", "rand", "rmp", @@ -1261,6 +1263,7 @@ dependencies = [ "rmpv", "rustc-hash", "serde", + "serde-transcode", "serde_json", "thin-vec", "tokio", @@ -1739,11 +1742,11 @@ dependencies = [ "http", "js-sys", "libdatadog-nodejs-capabilities", - "libdd-capabilities 2.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-common 5.0.0", + "libdd-capabilities 2.1.0", + "libdd-common 5.1.0", "libdd-data-pipeline", - "libdd-shared-runtime 1.0.0 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", - "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?tag=v37.0.0)", + "libdd-shared-runtime 2.0.0", + "libdd-trace-protobuf 4.0.0", "libdd-trace-stats", "libdd-trace-utils", "rmp-serde", @@ -1804,7 +1807,7 @@ version = "0.1.0" dependencies = [ "anyhow", "libdd-library-config 2.0.0", - "libdd-trace-protobuf 3.0.2 (git+https://github.com/DataDog/libdatadog.git?rev=7cdeb7896e92d1ba38bde495934e112dac2eda25)", + "libdd-trace-protobuf 3.0.2", "napi", "napi-derive", ] @@ -2140,6 +2143,15 @@ dependencies = [ "serde_derive", ] +[[package]] +name = "serde-transcode" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "590c0e25c2a5bb6e85bf5c1bce768ceb86b316e7a01bdf07d2cb4ec2271990e2" +dependencies = [ + "serde", +] + [[package]] name = "serde-wasm-bindgen" version = "0.4.5" diff --git a/crates/capabilities/Cargo.toml b/crates/capabilities/Cargo.toml index 888f566..cbb376e 100644 --- a/crates/capabilities/Cargo.toml +++ b/crates/capabilities/Cargo.toml @@ -15,7 +15,9 @@ http = "1" bytes = "1.4" futures-core = "0.3" anyhow = "1" -libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0" } +# TODO: Replace this temporary libdatadog PR rev with the official release/tag +# that contains DataDog/libdatadog#2235, then regenerate Cargo.lock. +libdd-capabilities = { git = "https://github.com/DataDog/libdatadog.git", rev = "8cd68ab922fb7aade0f089ccfc91291d874673af" } [dev-dependencies] wasm-bindgen-test = "0.3" diff --git a/crates/pipeline/Cargo.toml b/crates/pipeline/Cargo.toml index 7501ad6..f8e318a 100644 --- a/crates/pipeline/Cargo.toml +++ b/crates/pipeline/Cargo.toml @@ -14,13 +14,15 @@ 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", tag = "v37.0.0" } -libdd-common = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0", default-features = false } -libdd-data-pipeline = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0", default-features = false } -libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0", default-features = false, features = ["change-buffer"] } -libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0", default-features = false } -libdd-trace-protobuf = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0", default-features = false } -libdd-shared-runtime = { git = "https://github.com/DataDog/libdatadog.git", tag = "v37.0.0", default-features = false } +# 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 = "8cd68ab922fb7aade0f089ccfc91291d874673af" } +libdd-common = { git = "https://github.com/DataDog/libdatadog.git", rev = "8cd68ab922fb7aade0f089ccfc91291d874673af", default-features = false } +libdd-data-pipeline = { git = "https://github.com/DataDog/libdatadog.git", rev = "8cd68ab922fb7aade0f089ccfc91291d874673af", default-features = false } +libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog.git", rev = "8cd68ab922fb7aade0f089ccfc91291d874673af", default-features = false, features = ["change-buffer"] } +libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog.git", rev = "8cd68ab922fb7aade0f089ccfc91291d874673af", default-features = false } +libdd-trace-protobuf = { git = "https://github.com/DataDog/libdatadog.git", rev = "8cd68ab922fb7aade0f089ccfc91291d874673af", default-features = false } +libdd-shared-runtime = { git = "https://github.com/DataDog/libdatadog.git", rev = "8cd68ab922fb7aade0f089ccfc91291d874673af", default-features = false } rmp-serde = "1" bytes = "1" http = "1" diff --git a/crates/pipeline/src/lib.rs b/crates/pipeline/src/lib.rs index 181589a..31188af 100644 --- a/crates/pipeline/src/lib.rs +++ b/crates/pipeline/src/lib.rs @@ -96,14 +96,16 @@ fn se_read_scalar( 1 => Ok(AttributeArrayValue::Boolean(se_read_u8(buf, idx)? != 0)), 2 => { se_need(buf, *idx, 8)?; - let n = get_num(buf, idx) - .ok_or_else(|| JsValue::from_str("addSpanEvent: truncated span-event attribute buffer"))?; + let n = get_num(buf, idx).ok_or_else(|| { + JsValue::from_str("addSpanEvent: truncated span-event attribute buffer") + })?; Ok(AttributeArrayValue::Integer(n)) } 3 => { se_need(buf, *idx, 8)?; - let n = get_num(buf, idx) - .ok_or_else(|| JsValue::from_str("addSpanEvent: truncated span-event attribute buffer"))?; + let n = get_num(buf, idx).ok_or_else(|| { + JsValue::from_str("addSpanEvent: truncated span-event attribute buffer") + })?; Ok(AttributeArrayValue::Double(n)) } _ => Err(JsValue::from_str( @@ -218,6 +220,21 @@ 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)] @@ -245,6 +262,7 @@ impl WasmSpanState { .set_language(lang) .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 @@ -417,7 +435,8 @@ impl WasmSpanState { return Ok(false); } - self.cbs.borrow_mut() + self.cbs + .borrow_mut() .flush_change_buffer() .map_err(|e| JsValue::from_str(&e.to_string()))?; @@ -432,7 +451,8 @@ impl WasmSpanState { } let spans_vec = self - .cbs.borrow_mut() + .cbs + .borrow_mut() .flush_chunk(&span_ids, first_is_local_root) .map_err(|e| JsValue::from_str(&e.to_string()))?; @@ -526,9 +546,7 @@ impl WasmSpanState { // Unreachable: the block above either set `Some` or returned early. None => return Err(build_failure_error("native exporter unavailable")), }; - let resp = exporter - .send_trace_chunks_async(chunks) - .await; + let resp = exporter.send_trace_chunks_async(chunks).await; let response_str = resp.map(|resp| match resp { AgentResponse::Unchanged => "unchanged".to_string(), AgentResponse::Changed { body } => body, @@ -542,9 +560,11 @@ impl WasmSpanState { /// 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. + /// `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 { + 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 @@ -552,24 +572,26 @@ impl WasmSpanState { // 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 req = { + 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 Ok(false), + None => return stats_flush_result(false, 0), } }; - match req { + let sent = match prepared.request { Some(req) => { stats::StatsCollector::send_request(req) .await .map_err(|e| JsValue::from_str(&e))?; - Ok(true) + true } - None => Ok(false), - } + None => false, + }; + + stats_flush_result(sent, prepared.collapsed_spans) } /// Flush the queued change-buffer operations. On success always returns @@ -577,7 +599,8 @@ impl WasmSpanState { /// flush methods); failures surface as a thrown error. #[wasm_bindgen(js_name = "flushChangeQueue")] pub fn flush_change_queue(&self) -> Result { - self.cbs.borrow_mut() + self.cbs + .borrow_mut() .flush_change_buffer() .map_err(|e| JsValue::from_str(&e.to_string()))?; Ok(true) @@ -605,7 +628,8 @@ impl WasmSpanState { #[wasm_bindgen(js_name = "stringTableInsertOne")] pub fn string_table_insert_one(&self, key: u32, val: &str) { - self.cbs.borrow_mut() + self.cbs + .borrow_mut() .string_table_insert_one(key, val.into()); } @@ -622,7 +646,9 @@ impl WasmSpanState { // the encoded entries must error, not index out of bounds. get_num // does the (overflow-safe) bounds check and returns None past the end. let key: u32 = get_num(buf, &mut index).ok_or_else(|| { - JsValue::from_str("stringTableInsertMany: count exceeds the entries in the input buffer") + JsValue::from_str( + "stringTableInsertMany: count exceeds the entries in the input buffer", + ) })?; let str_slice = &buf[index..]; // Bound the NUL scan to the input slice so a non-terminated string @@ -654,7 +680,9 @@ impl WasmSpanState { pub fn get_service_name(&self, span_id: u64) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - let span = cbs.get_span(span_id).map_err(|e| JsValue::from_str(&e.to_string()))?; + let span = cbs + .get_span(span_id) + .map_err(|e| JsValue::from_str(&e.to_string()))?; Ok(span.service.to_string()) } @@ -662,7 +690,9 @@ impl WasmSpanState { pub fn get_resource_name(&self, span_id: u64) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - let span = cbs.get_span(span_id).map_err(|e| JsValue::from_str(&e.to_string()))?; + let span = cbs + .get_span(span_id) + .map_err(|e| JsValue::from_str(&e.to_string()))?; Ok(span.resource.to_string()) } @@ -670,10 +700,14 @@ impl WasmSpanState { pub fn get_meta_attr(&self, span_id: u64, name: &str) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - let span = cbs.get_span(span_id).map_err(|e| JsValue::from_str(&e.to_string()))?; + let span = cbs + .get_span(span_id) + .map_err(|e| JsValue::from_str(&e.to_string()))?; // VecMap::get accepts &str directly (SpanString: Borrow), so no // SpanString allocation is needed for the lookup. - Ok(span.meta.get(name) + Ok(span + .meta + .get(name) .map(|v| JsValue::from_str(v.0.as_ref())) .unwrap_or(JsValue::NULL)) } @@ -682,8 +716,12 @@ impl WasmSpanState { pub fn get_metric_attr(&self, span_id: u64, name: &str) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - let span = cbs.get_span(span_id).map_err(|e| JsValue::from_str(&e.to_string()))?; - Ok(span.metrics.get(name) + let span = cbs + .get_span(span_id) + .map_err(|e| JsValue::from_str(&e.to_string()))?; + Ok(span + .metrics + .get(name) .map(|v| JsValue::from_f64(*v)) .unwrap_or(JsValue::NULL)) } @@ -692,7 +730,9 @@ impl WasmSpanState { pub fn get_error(&self, span_id: u64) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - let span = cbs.get_span(span_id).map_err(|e| JsValue::from_str(&e.to_string()))?; + let span = cbs + .get_span(span_id) + .map_err(|e| JsValue::from_str(&e.to_string()))?; Ok(span.error) } @@ -703,7 +743,9 @@ impl WasmSpanState { pub fn get_start(&self, span_id: u64) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - let span = cbs.get_span(span_id).map_err(|e| JsValue::from_str(&e.to_string()))?; + let span = cbs + .get_span(span_id) + .map_err(|e| JsValue::from_str(&e.to_string()))?; Ok(span.start) } @@ -711,7 +753,9 @@ impl WasmSpanState { pub fn get_duration(&self, span_id: u64) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - let span = cbs.get_span(span_id).map_err(|e| JsValue::from_str(&e.to_string()))?; + let span = cbs + .get_span(span_id) + .map_err(|e| JsValue::from_str(&e.to_string()))?; Ok(span.duration) } @@ -719,7 +763,9 @@ impl WasmSpanState { pub fn get_type(&self, span_id: u64) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - let span = cbs.get_span(span_id).map_err(|e| JsValue::from_str(&e.to_string()))?; + let span = cbs + .get_span(span_id) + .map_err(|e| JsValue::from_str(&e.to_string()))?; Ok(span.r#type.to_string()) } @@ -727,7 +773,9 @@ impl WasmSpanState { pub fn get_name(&self, span_id: u64) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - let span = cbs.get_span(span_id).map_err(|e| JsValue::from_str(&e.to_string()))?; + let span = cbs + .get_span(span_id) + .map_err(|e| JsValue::from_str(&e.to_string()))?; Ok(span.name.to_string()) } @@ -738,12 +786,7 @@ impl WasmSpanState { // ordering is safe (subsequent ops are applied on the next flush and never // touch meta_struct). #[wasm_bindgen(js_name = "setMetaStruct")] - pub fn set_meta_struct( - &self, - span_id: u64, - key: &str, - value: &[u8], - ) -> Result<(), JsValue> { + pub fn set_meta_struct(&self, span_id: u64, key: &str, value: &[u8]) -> Result<(), JsValue> { self.flush_change_queue()?; let mut cbs = self.cbs.borrow_mut(); let span = cbs @@ -758,9 +801,13 @@ impl WasmSpanState { pub fn get_meta_struct(&self, span_id: u64, key: &str) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - let span = cbs.get_span(span_id).map_err(|e| JsValue::from_str(&e.to_string()))?; + let span = cbs + .get_span(span_id) + .map_err(|e| JsValue::from_str(&e.to_string()))?; // VecMap::get accepts &str directly (SpanString: Borrow). - Ok(span.meta_struct.get(key) + Ok(span + .meta_struct + .get(key) .map(|v| JsValue::from(js_sys::Uint8Array::from(v.0.as_slice()))) .unwrap_or(JsValue::NULL)) } @@ -816,7 +863,8 @@ impl WasmSpanState { pub fn get_trace_meta_attr(&self, segment_id: u64, name: &str) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - Ok(cbs.get_segment(&segment_id) + Ok(cbs + .get_segment(&segment_id) .and_then(|s| s.meta.get(name)) .map(|v| JsValue::from_str(v.0.as_ref())) .unwrap_or(JsValue::NULL)) @@ -826,7 +874,8 @@ impl WasmSpanState { pub fn get_trace_metric_attr(&self, segment_id: u64, name: &str) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - Ok(cbs.get_segment(&segment_id) + Ok(cbs + .get_segment(&segment_id) .and_then(|s| s.metrics.get(name)) .map(|v| JsValue::from_f64(*v)) .unwrap_or(JsValue::NULL)) @@ -836,7 +885,8 @@ impl WasmSpanState { pub fn get_trace_origin(&self, segment_id: u64) -> Result { self.flush_change_queue()?; let cbs = self.cbs.borrow(); - Ok(cbs.get_segment(&segment_id) + Ok(cbs + .get_segment(&segment_id) .and_then(|s| s.origin.as_ref()) .map(|v| JsValue::from_str(v.0.as_ref())) .unwrap_or(JsValue::NULL)) @@ -870,8 +920,12 @@ pub fn get_op_codes() -> JsValue { ("SetTraceOrigin", 12), ]; for (name, val) in entries { - js_sys::Reflect::set(&obj, &JsValue::from_str(name), &JsValue::from_f64(*val as f64)) - .expect("Reflect::set on a freshly created object cannot fail"); + js_sys::Reflect::set( + &obj, + &JsValue::from_str(name), + &JsValue::from_f64(*val as f64), + ) + .expect("Reflect::set on a freshly created object cannot fail"); } obj.into() } diff --git a/crates/pipeline/src/stats.rs b/crates/pipeline/src/stats.rs index 3141036..127a21e 100644 --- a/crates/pipeline/src/stats.rs +++ b/crates/pipeline/src/stats.rs @@ -17,11 +17,11 @@ fn now() -> SystemTime { } 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 libdatadog_nodejs_capabilities::WasmHttpClient; use crate::trace_data::WasmTraceData; @@ -38,6 +38,12 @@ pub struct StatsMeta { 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, @@ -60,6 +66,7 @@ impl StatsCollector { "consumer".to_string(), ], Vec::new(), + None, ), meta, agent_url, @@ -77,21 +84,38 @@ impl StatsCollector { } } - /// Drain aggregated stats into a ready-to-send request, **synchronously**. + /// Drain aggregated stats into a ready-to-send request plus flush metadata, + /// **synchronously**. /// - /// Returns `Ok(None)` when there is nothing to flush. The concentrator is - /// drained and the sequence advanced as part of this call, so the 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>, String> { - let buckets = self.concentrator.flush(now(), force); - if buckets.is_empty() { - return Ok(None); + /// 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) @@ -114,8 +138,11 @@ impl StatsCollector { 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}"))?); + 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() @@ -127,7 +154,10 @@ impl StatsCollector { .body(Bytes::from(body)) .map_err(|e| format!("failed to build stats request: {e}"))?; - Ok(Some(req)) + Ok(PreparedStatsFlush { + request: Some(req), + collapsed_spans, + }) } /// Send a prepared stats request to the agent. Does **not** borrow the diff --git a/test/pipeline.js b/test/pipeline.js index 18ececa..bc6eabf 100644 --- a/test/pipeline.js +++ b/test/pipeline.js @@ -945,6 +945,7 @@ describe('pipeline', { skip }, () => { url: req.url, ct: req.headers['content-type'], len: Buffer.concat(chunks).length, + body: Buffer.concat(chunks).toString(), }) res.writeHead(200, { 'content-type': 'application/json' }) res.end('{}') @@ -952,7 +953,10 @@ describe('pipeline', { skip }, () => { }) 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}` }) + const ns = new NativeSpansInterface({ + agentUrl: `http://127.0.0.1:${port}`, + tracerVersion: '7.0.0-pre', + }) ns.state.setOtlpEndpoint(`http://127.0.0.1:${port}/v1/traces`) const span = ns.createSpan() span.name = 'otlp-span' @@ -968,6 +972,9 @@ describe('pipeline', { skip }, () => { // No setOtlpProtocol call — pins the default wire protocol (http/json). assert.match(req.ct || '', /json/) assert.ok(req.len > 0, 'OTLP body is non-empty') + const body = JSON.parse(req.body) + assert.strictEqual(body.resourceSpans[0].scopeSpans[0].scope.name, 'dd-trace-js') + assert.strictEqual(body.resourceSpans[0].scopeSpans[0].scope.version, '7.0.0-pre') } finally { server.closeAllConnections?.() server.close() @@ -1106,24 +1113,72 @@ describe('pipeline', { skip }, () => { try { await ns.flushSpans(span) - const sent = await ns.state.flushStats(true) - assert.strictEqual(sent, true, 'flushStats reported a send') + 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.strictEqual(await ns.state.flushStats(true), false, 'second flush has nothing to send') + 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 returns false when stats are disabled', async () => { + it('flushStats reports no send when stats are disabled', async () => { const ns = new NativeSpansInterface({ statsEnabled: false }) - assert.strictEqual(await ns.state.flushStats(true), 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 () => { @@ -1167,8 +1222,12 @@ describe('pipeline', { skip }, () => { try { await ns.flushSpans(span) - const sent = await ns.state.flushStats(true) - assert.strictEqual(sent, true, 'flushStats reported a send over the socket') + 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') From 54a172b9ccdff99dc359e9af5a2f86f7f6bd72c7 Mon Sep 17 00:00:00 2001 From: Bryan English Date: Wed, 15 Jul 2026 15:01:12 -0400 Subject: [PATCH 2/2] v0.17.0 --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 4012867..50f90e4 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@datadog/libdatadog", - "version": "0.16.0", + "version": "0.17.0", "description": "Node.js binding for libdatadog", "main": "index.js", "scripts": {