diff --git a/Cargo.lock b/Cargo.lock index f2eeb0b..22d8213 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" @@ -897,7 +897,7 @@ dependencies = [ "futures-core", "http", "js-sys", - "libdd-capabilities", + "libdd-capabilities 2.1.0", "wasm-bindgen", "wasm-bindgen-futures", "wasm-bindgen-test", @@ -914,6 +914,17 @@ dependencies = [ "thiserror", ] +[[package]] +name = "libdd-capabilities" +version = "2.1.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" +dependencies = [ + "anyhow", + "bytes", + "http", + "thiserror", +] + [[package]] name = "libdd-capabilities-impl" version = "2.0.0" @@ -922,8 +933,21 @@ dependencies = [ "bytes", "http", "http-body-util", - "libdd-capabilities", - "libdd-common", + "libdd-capabilities 2.0.0", + "libdd-common 5.0.0", + "tokio", +] + +[[package]] +name = "libdd-capabilities-impl" +version = "3.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" +dependencies = [ + "bytes", + "http", + "http-body-util", + "libdd-capabilities 2.1.0", + "libdd-common 5.1.0", "tokio", ] @@ -962,6 +986,36 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "libdd-common" +version = "5.1.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" +dependencies = [ + "anyhow", + "bytes", + "cc", + "const_format", + "futures", + "futures-core", + "futures-util", + "hex", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "libc", + "nix 0.29.0", + "pin-project", + "regex", + "serde", + "static_assertions", + "thiserror", + "tokio", + "tower-service", + "windows-sys 0.52.0", +] + [[package]] name = "libdd-crashtracker" version = "1.0.0" @@ -974,7 +1028,7 @@ dependencies = [ "errno", "http", "libc", - "libdd-common", + "libdd-common 5.0.0", "libdd-libunwind-sys", "libdd-telemetry", "nix 0.29.0", @@ -997,8 +1051,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", @@ -1008,15 +1062,15 @@ dependencies = [ "getrandom 0.2.17", "http", "http-body-util", - "libdd-capabilities", - "libdd-capabilities-impl", - "libdd-common", - "libdd-ddsketch", + "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", + "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", @@ -1037,15 +1091,23 @@ dependencies = [ "prost", ] +[[package]] +name = "libdd-ddsketch" +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", + "libdd-common 5.1.0", "serde", "tracing", ] @@ -1071,7 +1133,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", @@ -1100,9 +1162,26 @@ dependencies = [ "async-trait", "futures", "futures-util", - "libdd-capabilities", - "libdd-capabilities-impl", - "libdd-common", + "libdd-capabilities 2.0.0", + "libdd-capabilities-impl 2.0.0", + "libdd-common 5.0.0", + "tokio", + "tokio-util", + "tracing", + "wasm-bindgen-futures", +] + +[[package]] +name = "libdd-shared-runtime" +version = "2.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=8cd68ab922fb7aade0f089ccfc91291d874673af#8cd68ab922fb7aade0f089ccfc91291d874673af" +dependencies = [ + "async-trait", + "futures", + "futures-util", + "libdd-capabilities 2.1.0", + "libdd-capabilities-impl 3.0.0", + "libdd-common 5.1.0", "tokio", "tokio-util", "tracing", @@ -1123,9 +1202,9 @@ dependencies = [ "http", "http-body-util", "libc", - "libdd-common", - "libdd-ddsketch", - "libdd-shared-runtime", + "libdd-common 5.0.0", + "libdd-ddsketch 1.0.1", + "libdd-shared-runtime 1.0.0", "serde", "serde_json", "sys-info", @@ -1139,29 +1218,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", - "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", @@ -1172,7 +1251,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", @@ -1181,8 +1260,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", @@ -1191,21 +1270,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", - "libdd-capabilities-impl", - "libdd-common", - "libdd-ddsketch", - "libdd-shared-runtime", + "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", @@ -1216,8 +1297,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", @@ -1229,12 +1310,12 @@ dependencies = [ "http-body", "http-body-util", "indexmap", - "libdd-capabilities", - "libdd-capabilities-impl", - "libdd-common", + "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", @@ -1242,6 +1323,7 @@ dependencies = [ "rmpv", "rustc-hash", "serde", + "serde-transcode", "serde_json", "thin-vec", "tokio", @@ -1720,11 +1802,11 @@ dependencies = [ "http", "js-sys", "libdatadog-nodejs-capabilities", - "libdd-capabilities", - "libdd-common", + "libdd-capabilities 2.1.0", + "libdd-common 5.1.0", "libdd-data-pipeline", - "libdd-shared-runtime", - "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", @@ -1785,7 +1867,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", ] @@ -2034,7 +2116,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2148,6 +2230,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')