diff --git a/Cargo.lock b/Cargo.lock index 7d9e6e6..59abb26 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1206,6 +1206,12 @@ checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" name = "libdatadog" version = "0.1.0" dependencies = [ + "anyhow", + "bytes", + "futures", + "http 1.5.0", + "libdatadog-data-pipeline", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", "libdd-ddsketch 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", "napi 3.12.2", "napi-build", @@ -1213,6 +1219,16 @@ dependencies = [ "zstd", ] +[[package]] +name = "libdatadog-data-pipeline" +version = "0.1.0" +dependencies = [ + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", + "libdd-data-pipeline-core", + "libdd-trace-utils 10.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", + "thiserror 1.0.69", +] + [[package]] name = "libdatadog-nodejs-capabilities" version = "0.1.0" @@ -1222,7 +1238,7 @@ dependencies = [ "futures-core", "http 1.5.0", "js-sys", - "libdd-capabilities 3.0.0", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "wasm-bindgen", "wasm-bindgen-futures", "wasm-bindgen-test", @@ -1232,10 +1248,17 @@ dependencies = [ name = "libdatadog-wasm" version = "0.1.0" dependencies = [ + "anyhow", + "bytes", + "futures", "getrandom 0.2.17", + "http 1.5.0", "js-sys", + "libdatadog-data-pipeline", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", "libdd-ddsketch 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", "wasm-bindgen", + "wasm-bindgen-futures", "zrip", ] @@ -1263,6 +1286,19 @@ dependencies = [ "thiserror 1.0.69", ] +[[package]] +name = "libdd-capabilities" +version = "3.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0#cbe142c07c649fb24faa6eba0f294c2a31726cd0" +dependencies = [ + "anyhow", + "bytes", + "futures-channel", + "futures-util", + "http 1.5.0", + "thiserror 1.0.69", +] + [[package]] name = "libdd-capabilities-impl" version = "2.0.0" @@ -1285,8 +1321,8 @@ dependencies = [ "bytes", "http 1.5.0", "http-body-util", - "libdd-capabilities 3.0.0", - "libdd-common 5.2.0", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "tokio", ] @@ -1360,6 +1396,31 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "libdd-common" +version = "5.2.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0#cbe142c07c649fb24faa6eba0f294c2a31726cd0" +dependencies = [ + "anyhow", + "bytes", + "cc", + "const_format", + "futures", + "futures-core", + "futures-util", + "hex", + "http 1.5.0", + "libc", + "nix 0.29.0", + "pin-project", + "regex", + "regex-lite", + "serde", + "static_assertions", + "thiserror 1.0.69", + "windows-sys 0.52.0", +] + [[package]] name = "libdd-crashtracker" version = "1.0.0" @@ -1407,18 +1468,18 @@ dependencies = [ "getrandom 0.2.17", "http 1.5.0", "http-body-util", - "libdd-capabilities 3.0.0", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-capabilities-impl 4.0.0", - "libdd-common 5.2.0", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-ddsketch 1.1.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-dogstatsd-client", "libdd-shared-runtime 3.0.0", "libdd-telemetry 7.0.0", - "libdd-tinybytes", - "libdd-trace-normalization", - "libdd-trace-protobuf 4.0.1", + "libdd-tinybytes 1.1.2 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-trace-normalization 3.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-trace-protobuf 4.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-trace-stats", - "libdd-trace-utils", + "libdd-trace-utils 10.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "rmp-serde", "serde", "serde_json", @@ -1430,6 +1491,19 @@ dependencies = [ "web-time", ] +[[package]] +name = "libdd-data-pipeline-core" +version = "1.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0#cbe142c07c649fb24faa6eba0f294c2a31726cd0" +dependencies = [ + "http 1.5.0", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", + "libdd-trace-utils 10.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", + "serde_json", + "thiserror 1.0.69", +] + [[package]] name = "libdd-ddsketch" version = "1.0.1" @@ -1464,7 +1538,7 @@ dependencies = [ "async-trait", "cadence", "http 1.5.0", - "libdd-common 5.2.0", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-shared-runtime 3.0.0", "serde", "tokio", @@ -1514,10 +1588,10 @@ dependencies = [ "hashbrown 0.15.5", "http 1.5.0", "http-body-util", - "libdd-capabilities 3.0.0", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-capabilities-impl 4.0.0", - "libdd-common 5.2.0", - "libdd-trace-protobuf 4.0.1", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-trace-protobuf 4.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-tuf", "manual_future", "prost", @@ -1562,9 +1636,9 @@ dependencies = [ "async-trait", "futures", "futures-util", - "libdd-capabilities 3.0.0", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-capabilities-impl 4.0.0", - "libdd-common 5.2.0", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "tokio", "tokio-util", "tracing", @@ -1612,8 +1686,8 @@ dependencies = [ "hashbrown 0.15.5", "http 1.5.0", "libc", - "libdd-capabilities 3.0.0", - "libdd-common 5.2.0", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-ddsketch 1.1.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-shared-runtime 3.0.0", "serde", @@ -1637,13 +1711,30 @@ dependencies = [ "serde", ] +[[package]] +name = "libdd-tinybytes" +version = "1.1.2" +source = "git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0#cbe142c07c649fb24faa6eba0f294c2a31726cd0" +dependencies = [ + "serde", +] + [[package]] name = "libdd-trace-normalization" version = "3.0.1" source = "git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75#966f921c226b63a1ba84c2e53e3d0f4625f90c75" dependencies = [ "anyhow", - "libdd-trace-protobuf 4.0.1", + "libdd-trace-protobuf 4.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", +] + +[[package]] +name = "libdd-trace-normalization" +version = "3.0.1" +source = "git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0#cbe142c07c649fb24faa6eba0f294c2a31726cd0" +dependencies = [ + "anyhow", + "libdd-trace-protobuf 4.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", ] [[package]] @@ -1653,9 +1744,9 @@ source = "git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84 dependencies = [ "anyhow", "fluent-uri", - "libdd-common 5.2.0", - "libdd-trace-protobuf 4.0.1", - "libdd-trace-utils", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-trace-protobuf 4.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-trace-utils 10.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "log", "percent-encoding", "serde", @@ -1682,6 +1773,16 @@ dependencies = [ "serde_bytes", ] +[[package]] +name = "libdd-trace-protobuf" +version = "4.0.1" +source = "git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0#cbe142c07c649fb24faa6eba0f294c2a31726cd0" +dependencies = [ + "prost", + "serde", + "serde_bytes", +] + [[package]] name = "libdd-trace-stats" version = "7.0.0" @@ -1693,16 +1794,16 @@ dependencies = [ "futures", "hashbrown 0.15.5", "http 1.5.0", - "libdd-capabilities 3.0.0", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-capabilities-impl 4.0.0", - "libdd-common 5.2.0", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-ddsketch 1.1.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-dogstatsd-client", "libdd-shared-runtime 3.0.0", "libdd-telemetry 7.0.0", "libdd-trace-obfuscation", - "libdd-trace-protobuf 4.0.1", - "libdd-trace-utils", + "libdd-trace-protobuf 4.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-trace-utils 10.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "rmp-serde", "serde", "tokio", @@ -1727,12 +1828,12 @@ dependencies = [ "http-body-util", "indexmap 2.14.0", "itoa 1.0.18", - "libdd-capabilities 3.0.0", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-capabilities-impl 4.0.0", - "libdd-common 5.2.0", - "libdd-tinybytes", - "libdd-trace-normalization", - "libdd-trace-protobuf 4.0.1", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-tinybytes 1.1.2 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-trace-normalization 3.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-trace-protobuf 4.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "prost", "rand", "rmp", @@ -1747,6 +1848,42 @@ dependencies = [ "tracing", ] +[[package]] +name = "libdd-trace-utils" +version = "10.1.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0#cbe142c07c649fb24faa6eba0f294c2a31726cd0" +dependencies = [ + "anyhow", + "base64", + "bytes", + "futures", + "getrandom 0.2.17", + "hex", + "http 1.5.0", + "http-body", + "http-body-util", + "indexmap 2.14.0", + "itoa 1.0.18", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", + "libdd-tinybytes 1.1.2 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", + "libdd-trace-normalization 3.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", + "libdd-trace-protobuf 4.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=cbe142c07c649fb24faa6eba0f294c2a31726cd0)", + "prost", + "rand", + "rmp", + "rmp-serde", + "rmpv", + "rustc-hash", + "serde", + "serde-transcode", + "serde_json", + "thin-vec", + "tracing", + "zrip", + "zstd", +] + [[package]] name = "libdd-tuf" version = "0.3.1" @@ -1939,6 +2076,7 @@ dependencies = [ "napi-sys 3.3.0", "nohash-hasher", "rustc-hash", + "tokio", ] [[package]] @@ -2337,13 +2475,13 @@ dependencies = [ "http 1.5.0", "js-sys", "libdatadog-nodejs-capabilities", - "libdd-capabilities 3.0.0", - "libdd-common 5.2.0", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-data-pipeline", "libdd-shared-runtime 3.0.0", - "libdd-trace-protobuf 4.0.1", + "libdd-trace-protobuf 4.0.1 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-trace-stats", - "libdd-trace-utils", + "libdd-trace-utils 10.1.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "rmp-serde", "serde", "serde_json", @@ -2533,6 +2671,12 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "regex-lite" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973" + [[package]] name = "regex-syntax" version = "0.8.11" @@ -2546,7 +2690,7 @@ dependencies = [ "console_error_panic_hook", "js-sys", "libdatadog-nodejs-capabilities", - "libdd-common 5.2.0", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-remote-config", "serde", "serde-wasm-bindgen", diff --git a/README.md b/README.md index f8f8d83..1c6689e 100644 --- a/README.md +++ b/README.md @@ -18,8 +18,8 @@ published as two packages with different availability guarantees. The universal package is maintained under [`packages/libdatadog`](packages/libdatadog). Its matching native and WASM -backends expose Zstandard compression and DDSketch from a single native or -WASM artifact. +backends expose the agentless data pipeline, Zstandard compression, and +DDSketch from a single native or WASM artifact. The package tries a platform-native napi-rs backend first. Native artifacts are published as optional dependencies using napi-rs-compatible package names such diff --git a/crates/libdatadog-data-pipeline/Cargo.toml b/crates/libdatadog-data-pipeline/Cargo.toml new file mode 100644 index 0000000..bb9748b --- /dev/null +++ b/crates/libdatadog-data-pipeline/Cargo.toml @@ -0,0 +1,28 @@ +[package] +name = "libdatadog-data-pipeline" +version = "0.1.0" +edition = "2021" +publish = false + +[dependencies] +thiserror = "1" + +[dependencies.libdd-capabilities] +git = "https://github.com/DataDog/libdatadog.git" +rev = "cbe142c07c649fb24faa6eba0f294c2a31726cd0" +default-features = false + +[dependencies.libdd-data-pipeline-core] +git = "https://github.com/DataDog/libdatadog.git" +rev = "cbe142c07c649fb24faa6eba0f294c2a31726cd0" +default-features = false + +[dependencies.libdd-trace-utils] +git = "https://github.com/DataDog/libdatadog.git" +rev = "cbe142c07c649fb24faa6eba0f294c2a31726cd0" +default-features = false + +[features] +default = [] +compression = ["libdd-data-pipeline-core/compression"] +regex-lite = ["libdd-data-pipeline-core/regex-lite"] diff --git a/crates/libdatadog-data-pipeline/src/lib.rs b/crates/libdatadog-data-pipeline/src/lib.rs new file mode 100644 index 0000000..c4dded4 --- /dev/null +++ b/crates/libdatadog-data-pipeline/src/lib.rs @@ -0,0 +1,32 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use libdd_capabilities::{HttpClientCapability, SleepCapability}; +use thiserror::Error; + +pub use libdd_data_pipeline_core::{ + AgentlessTraceConfig, TracerMetadata, DEFAULT_AGENTLESS_TIMEOUT, +}; + +#[derive(Debug, Error)] +pub enum SendAgentlessV04Error { + #[error("failed to decode v0.4 traces: {0}")] + Deserialization(libdd_trace_utils::msgpack_decoder::decode::error::DecodeError), + #[error(transparent)] + Agentless(#[from] libdd_data_pipeline_core::AgentlessError), +} + +pub async fn send_agentless_v04( + capabilities: &C, + payload: &[u8], + metadata: &TracerMetadata, + config: &AgentlessTraceConfig, +) -> Result<(), SendAgentlessV04Error> +where + C: HttpClientCapability + SleepCapability, +{ + let (traces, _) = libdd_trace_utils::msgpack_decoder::v04::from_slice(payload) + .map_err(SendAgentlessV04Error::Deserialization)?; + libdd_data_pipeline_core::send_agentless_traces(capabilities, traces, metadata, config).await?; + Ok(()) +} diff --git a/crates/libdatadog-wasm/Cargo.toml b/crates/libdatadog-wasm/Cargo.toml index 1648b8c..e915a93 100644 --- a/crates/libdatadog-wasm/Cargo.toml +++ b/crates/libdatadog-wasm/Cargo.toml @@ -10,10 +10,27 @@ name = "libdatadog_wasm" [dependencies] js-sys = "0.3" wasm-bindgen = "0.2" +wasm-bindgen-futures = "0.4" +anyhow = "1" +bytes = "1" +futures = "0.3" +http = "1" +# The agentless v0.4 path does not compile or evaluate user-provided patterns. +# Use the smaller regex engine so the full engine and its Unicode tables are +# removed by the WASM linker. libdd-ddsketch = "1.1.1" # Keep this aligned with the pure-Rust WASM encoder used by libdatadog. zrip = { version = "=0.6.0", default-features = false, features = ["alloc"] } +[dependencies.libdd-capabilities] +git = "https://github.com/DataDog/libdatadog.git" +rev = "cbe142c07c649fb24faa6eba0f294c2a31726cd0" +default-features = false + +[dependencies.libdatadog-data-pipeline] +path = "../libdatadog-data-pipeline" +features = ["compression", "regex-lite"] + [target.'cfg(target_arch = "wasm32")'.dependencies] getrandom = { version = "0.2", features = ["js"] } diff --git a/crates/libdatadog-wasm/src/data_pipeline/mod.rs b/crates/libdatadog-wasm/src/data_pipeline/mod.rs new file mode 100644 index 0000000..3ec8bf7 --- /dev/null +++ b/crates/libdatadog-wasm/src/data_pipeline/mod.rs @@ -0,0 +1,363 @@ +use std::cell::{Cell, RefCell}; +use std::collections::HashMap; +use std::fmt; +use std::rc::Rc; +use std::time::Duration; + +use bytes::Bytes; +use futures::future::{AbortHandle, Abortable}; +use js_sys::{Array, Function, Object, Promise, Reflect, Uint8Array}; +use libdatadog_data_pipeline::{ + send_agentless_v04, AgentlessTraceConfig, SendAgentlessV04Error, TracerMetadata, + DEFAULT_AGENTLESS_TIMEOUT, +}; +use libdd_capabilities::{HttpClientCapability, HttpError, SleepCapability}; +use wasm_bindgen::prelude::*; +use wasm_bindgen_futures::JsFuture; + +#[derive(Clone)] +struct AgentlessExporterOptions { + endpoint: String, + api_key: String, + hostname: Option, + env: Option, + service: Option, + version: Option, + runtime_id: Option, + container_id: Option, + tracer_version: String, + language_version: String, + language_interpreter: String, + timeout_ms: Option, +} + +#[derive(Clone)] +struct HostCapabilities { + request: Function, + cancel_request: Function, + sleep: Function, + cancel_sleep: Function, + next_call_id: Rc>, +} + +impl fmt::Debug for HostCapabilities { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("HostCapabilities") + } +} + +impl HostCapabilities { + fn next_call_id(&self) -> u32 { + let id = self.next_call_id.get(); + self.next_call_id.set(id.wrapping_add(1)); + id + } +} + +impl HttpClientCapability for HostCapabilities { + fn new_client() -> Self { + panic!("host capabilities must be constructed with JavaScript functions") + } + + fn new_without_connection_pooling() -> Self { + Self::new_client() + } + + async fn request( + &self, + request: http::Request, + ) -> Result, HttpError> { + let id = self.next_call_id(); + let plan = request_value(id, request)?; + let mut guard = CancelGuard::new(id, self.cancel_request.clone()); + let promise = self + .request + .call1(&JsValue::UNDEFINED, &plan) + .map(|value| Promise::resolve(&value)) + .map_err(network_error)?; + let response = JsFuture::from(promise).await.map_err(network_error)?; + guard.disarm(); + + let status = required_number(&response, "status")?; + let status = + u16::try_from(status).map_err(|error| HttpError::InvalidRequest(error.into()))?; + let body = Reflect::get(&response, &JsValue::from_str("body")).map_err(network_error)?; + http::Response::builder() + .status(status) + .body(Bytes::from(Uint8Array::new(&body).to_vec())) + .map_err(|error| HttpError::InvalidRequest(error.into())) + } +} + +impl SleepCapability for HostCapabilities { + fn new() -> Self { + panic!("host capabilities must be constructed with JavaScript functions") + } + + async fn sleep(&self, duration: Duration) { + let id = self.next_call_id(); + let milliseconds = duration_millis(duration); + let mut guard = CancelGuard::new(id, self.cancel_sleep.clone()); + if let Ok(value) = self.sleep.call2( + &JsValue::UNDEFINED, + &JsValue::from_f64(f64::from(id)), + &JsValue::from_f64(f64::from(milliseconds)), + ) { + let _ = JsFuture::from(Promise::resolve(&value)).await; + } + guard.disarm(); + } +} + +struct CancelGuard { + id: u32, + cancel: Function, + armed: bool, +} + +impl CancelGuard { + fn new(id: u32, cancel: Function) -> Self { + Self { + id, + cancel, + armed: true, + } + } + + fn disarm(&mut self) { + self.armed = false; + } +} + +impl Drop for CancelGuard { + fn drop(&mut self) { + if self.armed { + let _ = self + .cancel + .call1(&JsValue::UNDEFINED, &JsValue::from_f64(f64::from(self.id))); + } + } +} + +#[wasm_bindgen] +pub struct AgentlessExporter { + metadata: TracerMetadata, + config: AgentlessTraceConfig, + capabilities: HostCapabilities, + in_flight: Rc>>, + next_operation_id: Cell, +} + +#[wasm_bindgen] +impl AgentlessExporter { + #[wasm_bindgen(constructor)] + pub fn new( + value: JsValue, + request: Function, + cancel_request: Function, + sleep: Function, + cancel_sleep: Function, + ) -> Result { + let options = AgentlessExporterOptions { + endpoint: required_string(&value, "endpoint")?, + api_key: required_string(&value, "apiKey")?, + hostname: optional_string(&value, "hostname")?, + env: optional_string(&value, "env")?, + service: optional_string(&value, "service")?, + version: optional_string(&value, "version")?, + runtime_id: optional_string(&value, "runtimeId")?, + container_id: optional_string(&value, "containerId")?, + tracer_version: required_string(&value, "tracerVersion")?, + language_version: required_string(&value, "languageVersion")?, + language_interpreter: required_string(&value, "languageInterpreter")?, + timeout_ms: optional_number(&value, "timeoutMs")?, + }; + let metadata = TracerMetadata { + hostname: options.hostname.unwrap_or_default(), + env: options.env.unwrap_or_default(), + app_version: options.version.unwrap_or_default(), + runtime_id: options.runtime_id.unwrap_or_default(), + service: options.service.unwrap_or_default(), + tracer_version: options.tracer_version, + language: "nodejs".to_string(), + language_version: options.language_version, + language_interpreter: options.language_interpreter, + container_id: options.container_id.unwrap_or_default(), + ..Default::default() + }; + let timeout = options + .timeout_ms + .map(|timeout_ms| Duration::from_millis(u64::from(timeout_ms))) + .unwrap_or(DEFAULT_AGENTLESS_TIMEOUT); + let config = AgentlessTraceConfig { + endpoint_url: options.endpoint, + api_key: options.api_key, + timeout, + }; + let capabilities = HostCapabilities { + request, + cancel_request, + sleep, + cancel_sleep, + next_call_id: Rc::new(Cell::new(1)), + }; + + Ok(Self { + metadata, + config, + capabilities, + in_flight: Rc::new(RefCell::new(HashMap::new())), + next_operation_id: Cell::new(1), + }) + } + + #[wasm_bindgen(js_name = sendV04)] + pub async fn send_v04(&self, payload: &[u8]) -> Result<(), JsValue> { + let operation_id = self.next_operation_id.get(); + self.next_operation_id.set(operation_id.wrapping_add(1)); + let (abort, registration) = AbortHandle::new_pair(); + self.in_flight.borrow_mut().insert(operation_id, abort); + let _guard = OperationGuard { + id: operation_id, + in_flight: self.in_flight.clone(), + }; + let send = send_agentless_v04(&self.capabilities, payload, &self.metadata, &self.config); + + match Abortable::new(send, registration).await { + Ok(Ok(_)) => Ok(()), + Ok(Err(error)) => Err(send_error(error)), + Err(_) => Err(JsValue::from_str("data-pipeline export was cancelled")), + } + } + + #[wasm_bindgen(js_name = cancelAll)] + pub fn cancel_all(&self) { + for abort in self.in_flight.borrow().values() { + abort.abort(); + } + } +} + +impl Drop for AgentlessExporter { + fn drop(&mut self) { + self.cancel_all(); + } +} + +struct OperationGuard { + id: u32, + in_flight: Rc>>, +} + +impl Drop for OperationGuard { + fn drop(&mut self) { + self.in_flight.borrow_mut().remove(&self.id); + } +} + +fn request_value(id: u32, request: http::Request) -> Result { + let (parts, body) = request.into_parts(); + let headers = Array::new(); + for (name, value) in &parts.headers { + let value = value + .to_str() + .map_err(|error| HttpError::InvalidRequest(error.into()))?; + let header = Object::new(); + Reflect::set( + &header, + &JsValue::from_str("name"), + &JsValue::from_str(name.as_str()), + ) + .map_err(network_error)?; + Reflect::set( + &header, + &JsValue::from_str("value"), + &JsValue::from_str(value), + ) + .map_err(network_error)?; + headers.push(&header); + } + + let plan = Object::new(); + set(&plan, "id", &JsValue::from_f64(f64::from(id)))?; + set(&plan, "url", &JsValue::from_str(&parts.uri.to_string()))?; + set(&plan, "method", &JsValue::from_str(parts.method.as_str()))?; + set(&plan, "headers", &headers)?; + set(&plan, "body", &Uint8Array::from(body.as_ref()))?; + Ok(plan.into()) +} + +fn set(object: &Object, key: &str, value: &JsValue) -> Result<(), HttpError> { + Reflect::set(object, &JsValue::from_str(key), value) + .map(|_| ()) + .map_err(network_error) +} + +fn duration_millis(duration: Duration) -> u32 { + u32::try_from(duration.as_millis()).unwrap_or(u32::MAX) +} + +fn network_error(error: JsValue) -> HttpError { + HttpError::Network(anyhow::anyhow!(js_error_message(error))) +} + +fn send_error(error: SendAgentlessV04Error) -> JsValue { + JsValue::from_str(&format!("failed to send data-pipeline export: {error}")) +} + +fn js_error_message(value: JsValue) -> String { + Reflect::get(&value, &JsValue::from_str("message")) + .ok() + .and_then(|message| message.as_string()) + .or_else(|| value.as_string()) + .unwrap_or_else(|| "JavaScript transport error".to_string()) +} + +fn required_string(value: &JsValue, key: &str) -> Result { + optional_string(value, key)? + .ok_or_else(|| JsValue::from_str(&format!("{key} must be a string"))) +} + +fn optional_string(value: &JsValue, key: &str) -> Result, JsValue> { + let value = Reflect::get(value, &JsValue::from_str(key))?; + if value.is_null() || value.is_undefined() { + return Ok(None); + } + value + .as_string() + .ok_or_else(|| JsValue::from_str(&format!("{key} must be a string"))) + .map(Some) +} + +fn optional_number(value: &JsValue, key: &str) -> Result, JsValue> { + let value = Reflect::get(value, &JsValue::from_str(key))?; + if value.is_null() || value.is_undefined() { + return Ok(None); + } + let number = value + .as_f64() + .filter(|number| { + number.is_finite() + && *number >= 0.0 + && *number <= f64::from(u32::MAX) + && number.fract() == 0.0 + }) + .ok_or_else(|| JsValue::from_str(&format!("{key} must be an unsigned integer")))?; + Ok(Some(u32::try_from(number as u64).map_err(|_| { + JsValue::from_str(&format!("{key} must be an unsigned integer")) + })?)) +} + +fn required_number(value: &JsValue, key: &str) -> Result { + let number = Reflect::get(value, &JsValue::from_str(key)) + .map_err(network_error)? + .as_f64() + .filter(|number| { + number.is_finite() + && *number >= 0.0 + && *number <= f64::from(u32::MAX) + && number.fract() == 0.0 + }) + .ok_or_else(|| HttpError::InvalidRequest(anyhow::anyhow!("{key} must be an integer")))?; + Ok(number as u32) +} diff --git a/crates/libdatadog-wasm/src/lib.rs b/crates/libdatadog-wasm/src/lib.rs index 259fcac..e133c0a 100644 --- a/crates/libdatadog-wasm/src/lib.rs +++ b/crates/libdatadog-wasm/src/lib.rs @@ -1,2 +1,3 @@ +mod data_pipeline; mod sketches; mod zstd; diff --git a/crates/libdatadog/Cargo.toml b/crates/libdatadog/Cargo.toml index f15da88..ba36a37 100644 --- a/crates/libdatadog/Cargo.toml +++ b/crates/libdatadog/Cargo.toml @@ -8,10 +8,23 @@ crate-type = ["cdylib"] name = "libdatadog" [dependencies] -napi = { version = "3", default-features = false, features = ["napi8"] } +napi = { version = "3", default-features = false, features = ["async", "napi8"] } napi-derive = "3" +anyhow = "1" +bytes = "1" +futures = "0.3" +http = "1" libdd-ddsketch = "1.1.1" zstd = { version = "0.13.3", default-features = false, features = ["thin"] } +[dependencies.libdd-capabilities] +git = "https://github.com/DataDog/libdatadog.git" +rev = "cbe142c07c649fb24faa6eba0f294c2a31726cd0" +default-features = false + +[dependencies.libdatadog-data-pipeline] +path = "../libdatadog-data-pipeline" +features = ["compression"] + [build-dependencies] napi-build = "2" diff --git a/crates/libdatadog/src/data_pipeline/mod.rs b/crates/libdatadog/src/data_pipeline/mod.rs new file mode 100644 index 0000000..1ab402f --- /dev/null +++ b/crates/libdatadog/src/data_pipeline/mod.rs @@ -0,0 +1,335 @@ +use std::collections::HashMap; +use std::fmt; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use bytes::Bytes; +use futures::future::{AbortHandle, Abortable}; +use libdatadog_data_pipeline::{ + send_agentless_v04, AgentlessTraceConfig, SendAgentlessV04Error, TracerMetadata, + DEFAULT_AGENTLESS_TIMEOUT, +}; +use libdd_capabilities::{HttpClientCapability, HttpError, SleepCapability}; +use napi::bindgen_prelude::*; +use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode}; +use napi::Status; +use napi_derive::napi; + +type RequestFunction = ThreadsafeFunction< + AgentlessRequest, + Promise, + AgentlessRequest, + Status, + false, + true, +>; +type SleepArgs = FnArgs<(u32, u32)>; +type SleepFunction = ThreadsafeFunction, SleepArgs, Status, false, true>; +type CancelFunction = ThreadsafeFunction; + +#[napi(object)] +pub struct AgentlessExporterOptions { + pub endpoint: String, + pub api_key: String, + pub hostname: Option, + pub env: Option, + pub service: Option, + pub version: Option, + pub runtime_id: Option, + pub container_id: Option, + pub tracer_version: String, + pub language_version: String, + pub language_interpreter: String, + pub timeout_ms: Option, +} + +#[napi(object)] +pub struct AgentlessRequestHeader { + pub name: String, + pub value: String, +} + +#[napi(object)] +pub struct AgentlessRequest { + pub id: u32, + pub url: String, + pub method: String, + pub headers: Vec, + pub body: Buffer, +} + +#[napi(object)] +pub struct AgentlessResponse { + pub status: u16, + pub body: Buffer, +} + +#[derive(Clone)] +struct HostCapabilities { + request: Arc, + cancel_request: Arc, + sleep: Arc, + cancel_sleep: Arc, + next_call_id: Arc, +} + +impl fmt::Debug for HostCapabilities { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("HostCapabilities") + } +} + +impl HostCapabilities { + fn next_call_id(&self) -> u32 { + self.next_call_id.fetch_add(1, Ordering::Relaxed) + } +} + +impl HttpClientCapability for HostCapabilities { + fn new_client() -> Self { + panic!("host capabilities must be constructed with JavaScript functions") + } + + fn new_without_connection_pooling() -> Self { + Self::new_client() + } + + async fn request( + &self, + request: http::Request, + ) -> std::result::Result, HttpError> { + let id = self.next_call_id(); + let (parts, body) = request.into_parts(); + let headers = parts + .headers + .iter() + .map(|(name, value)| { + value + .to_str() + .map(|value| AgentlessRequestHeader { + name: name.to_string(), + value: value.to_string(), + }) + .map_err(|error| HttpError::InvalidRequest(error.into())) + }) + .collect::, _>>()?; + let request = AgentlessRequest { + id, + url: parts.uri.to_string(), + method: parts.method.to_string(), + headers, + body: body.to_vec().into(), + }; + let mut guard = CancelGuard::new(id, self.cancel_request.clone()); + let promise = self + .request + .call_async(request) + .await + .map_err(network_error)?; + let response = promise.await.map_err(network_error)?; + guard.disarm(); + + http::Response::builder() + .status(response.status) + .body(Bytes::from(response.body.to_vec())) + .map_err(|error| HttpError::InvalidRequest(error.into())) + } +} + +impl SleepCapability for HostCapabilities { + fn new() -> Self { + panic!("host capabilities must be constructed with JavaScript functions") + } + + async fn sleep(&self, duration: Duration) { + let id = self.next_call_id(); + let milliseconds = duration_millis(duration); + let mut guard = CancelGuard::new(id, self.cancel_sleep.clone()); + if let Ok(promise) = self.sleep.call_async((id, milliseconds).into()).await { + let _ = promise.await; + } + guard.disarm(); + } +} + +struct CancelGuard { + id: u32, + cancel: Arc, + armed: bool, +} + +impl CancelGuard { + fn new(id: u32, cancel: Arc) -> Self { + Self { + id, + cancel, + armed: true, + } + } + + fn disarm(&mut self) { + self.armed = false; + } +} + +impl Drop for CancelGuard { + fn drop(&mut self) { + if self.armed { + self.cancel + .call(self.id, ThreadsafeFunctionCallMode::NonBlocking); + } + } +} + +#[napi] +pub struct AgentlessExporter { + metadata: TracerMetadata, + config: AgentlessTraceConfig, + capabilities: HostCapabilities, + in_flight: Arc>>, + next_operation_id: Arc, +} + +#[napi] +impl AgentlessExporter { + #[napi(constructor)] + pub fn new( + options: AgentlessExporterOptions, + request: Function<'_, AgentlessRequest, Promise>, + cancel_request: Function<'_, u32, ()>, + sleep: Function<'_, FnArgs<(u32, u32)>, Promise<()>>, + cancel_sleep: Function<'_, u32, ()>, + ) -> Result { + let timeout = options + .timeout_ms + .map(|timeout_ms| Duration::from_millis(u64::from(timeout_ms))) + .unwrap_or(DEFAULT_AGENTLESS_TIMEOUT); + let metadata = TracerMetadata { + hostname: options.hostname.unwrap_or_default(), + env: options.env.unwrap_or_default(), + app_version: options.version.unwrap_or_default(), + runtime_id: options.runtime_id.unwrap_or_default(), + service: options.service.unwrap_or_default(), + tracer_version: options.tracer_version, + language: "nodejs".to_string(), + language_version: options.language_version, + language_interpreter: options.language_interpreter, + container_id: options.container_id.unwrap_or_default(), + ..Default::default() + }; + let config = AgentlessTraceConfig { + endpoint_url: options.endpoint, + api_key: options.api_key, + timeout, + }; + let capabilities = HostCapabilities { + request: Arc::new( + request + .build_threadsafe_function::() + .weak::() + .build()?, + ), + cancel_request: Arc::new( + cancel_request + .build_threadsafe_function::() + .weak::() + .build()?, + ), + sleep: Arc::new( + sleep + .build_threadsafe_function::() + .weak::() + .build()?, + ), + cancel_sleep: Arc::new( + cancel_sleep + .build_threadsafe_function::() + .weak::() + .build()?, + ), + next_call_id: Arc::new(AtomicU32::new(1)), + }; + + Ok(Self { + metadata, + config, + capabilities, + in_flight: Arc::new(Mutex::new(HashMap::new())), + next_operation_id: Arc::new(AtomicU32::new(1)), + }) + } + + #[napi] + pub fn send_v04<'env>(&self, env: &'env Env, payload: Buffer) -> Result> { + let operation_id = self.next_operation_id.fetch_add(1, Ordering::Relaxed); + let (abort, registration) = AbortHandle::new_pair(); + lock(&self.in_flight).insert(operation_id, abort); + let capabilities = self.capabilities.clone(); + let metadata = self.metadata.clone(); + let config = self.config.clone(); + let in_flight = self.in_flight.clone(); + let cleanup = in_flight.clone(); + let future = async move { + let _guard = OperationGuard { + id: operation_id, + in_flight, + }; + let send = send_agentless_v04(&capabilities, payload.as_ref(), &metadata, &config); + + match Abortable::new(send, registration).await { + Ok(Ok(_)) => Ok(()), + Ok(Err(error)) => Err(send_error(error)), + Err(_) => Err(Error::from_reason("data-pipeline export was cancelled")), + } + }; + + match env.spawn_future(future) { + Ok(promise) => Ok(promise), + Err(error) => { + lock(&cleanup).remove(&operation_id); + Err(error) + } + } + } + + #[napi] + pub fn cancel_all(&self) { + for abort in lock(&self.in_flight).values() { + abort.abort(); + } + } +} + +impl Drop for AgentlessExporter { + fn drop(&mut self) { + self.cancel_all(); + } +} + +struct OperationGuard { + id: u32, + in_flight: Arc>>, +} + +impl Drop for OperationGuard { + fn drop(&mut self) { + lock(&self.in_flight).remove(&self.id); + } +} + +fn lock(mutex: &Mutex) -> std::sync::MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(|error| error.into_inner()) +} + +fn duration_millis(duration: Duration) -> u32 { + u32::try_from(duration.as_millis()).unwrap_or(u32::MAX) +} + +fn network_error(error: napi::Error) -> HttpError { + HttpError::Network(anyhow::anyhow!(error.reason)) +} + +fn send_error(error: SendAgentlessV04Error) -> Error { + Error::from_reason(format!("failed to send data-pipeline export: {error}")) +} diff --git a/crates/libdatadog/src/lib.rs b/crates/libdatadog/src/lib.rs index 259fcac..e133c0a 100644 --- a/crates/libdatadog/src/lib.rs +++ b/crates/libdatadog/src/lib.rs @@ -1,2 +1,3 @@ +mod data_pipeline; mod sketches; mod zstd; diff --git a/eslint.config.js b/eslint.config.js index b59926f..c362da0 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -91,6 +91,7 @@ module.exports = [ files: [ 'packages/libdatadog/lib/wasm.js', 'packages/libdatadog/test/bundlers.test.js', + 'packages/libdatadog/test/exporter.test.js', ], rules: { 'n/no-missing-require': 'off', diff --git a/packages/libdatadog/README.md b/packages/libdatadog/README.md index 8406c78..2e710cf 100644 --- a/packages/libdatadog/README.md +++ b/packages/libdatadog/README.md @@ -10,8 +10,11 @@ are maintained in the root `crates/libdatadog` and `crates/libdatadog-wasm` workspace crates. Optional libdatadog functionality is published separately as `@datadog/libdatadog-extras`. -Both backends expose Zstandard compression and DDSketch from a single native -or WASM artifact. +Both backends expose the agentless data pipeline, Zstandard compression, and +DDSketch from a single native or WASM artifact. + +The package accepts Datadog v0.4 MessagePack payloads and exports them to an +agentless intake. The package tries a platform-native napi-rs backend first and falls back to a wasm-bindgen backend with the WebAssembly bytes embedded in JavaScript. The diff --git a/packages/libdatadog/index.d.ts b/packages/libdatadog/index.d.ts index 9e02614..6aec92a 100644 --- a/packages/libdatadog/index.d.ts +++ b/packages/libdatadog/index.d.ts @@ -1,3 +1,24 @@ +export interface AgentlessExporterOptions { + endpoint: string + apiKey: string + hostname?: string + env?: string + service?: string + version?: string + runtimeId?: string + containerId?: string + tracerVersion: string + languageVersion: string + languageInterpreter: string + timeoutMs?: number +} + +export interface AgentlessExporter { + sendV04(payload: Uint8Array): Promise + close(): void | Promise +} + +export function createAgentlessExporter(options: AgentlessExporterOptions): AgentlessExporter export function backend(): 'native' | 'wasm' export function zstd_compress(data: Uint8Array, level: number): Uint8Array diff --git a/packages/libdatadog/index.mjs b/packages/libdatadog/index.mjs index 9c1a9a9..17e4656 100644 --- a/packages/libdatadog/index.mjs +++ b/packages/libdatadog/index.mjs @@ -2,6 +2,7 @@ import libdatadog from './index.js' export const { backend, + createAgentlessExporter, DDSketch, zstd_compress, } = libdatadog diff --git a/packages/libdatadog/lib/agentless-transport.js b/packages/libdatadog/lib/agentless-transport.js new file mode 100644 index 0000000..daef09b --- /dev/null +++ b/packages/libdatadog/lib/agentless-transport.js @@ -0,0 +1,78 @@ +'use strict' + +function createHostTransport () { + const requests = new Map() + const timers = new Map() + + async function request ({ id, url, method, headers: headerList, body }) { + const target = new URL(url) + const client = target.protocol === 'https:' ? require('node:https') : require('node:http') + const headers = Object.fromEntries(headerList.map(({ name, value }) => [name, value])) + + return new Promise((resolve, reject) => { + let settled = false + const finish = (callback, value) => { + if (settled) return + settled = true + requests.delete(id) + callback(value) + } + const outgoing = client.request(target, { + agent: false, + headers: { ...headers, connection: 'close' }, + method, + }, (response) => { + const chunks = [] + response.on('data', chunk => chunks.push(chunk)) + response.once('aborted', () => finish(reject, new Error('response aborted'))) + response.once('error', error => finish(reject, error)) + response.on('end', () => finish(resolve, { + status: response.statusCode, + body: Buffer.concat(chunks), + })) + }) + + requests.set(id, { + cancel: () => { + const error = new Error('agentless request was cancelled') + finish(reject, error) + outgoing.destroy(error) + }, + }) + outgoing.once('error', error => finish(reject, error)) + outgoing.end(body) + }) + } + + function cancelRequest (id) { + requests.get(id)?.cancel() + } + + // TODO(libdd-capabilities): Make host-backed capability futures cancel their + // underlying operation when dropped. Then sleep can return a cancellable + // operation directly, removing timer IDs, the timers map, and cancelSleep. + function sleep (id, milliseconds) { + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + timers.delete(id) + resolve() + }, milliseconds) + timeout.unref?.() + timers.set(id, { + cancel: () => { + clearTimeout(timeout) + timers.delete(id) + reject(new Error('agentless timer was cancelled')) + }, + }) + }) + } + + function cancelSleep (id) { + timers.get(id)?.cancel() + } + + return { request, cancelRequest, sleep, cancelSleep } +} + +module.exports = { createHostTransport } diff --git a/packages/libdatadog/lib/agentless.js b/packages/libdatadog/lib/agentless.js new file mode 100644 index 0000000..5d82302 --- /dev/null +++ b/packages/libdatadog/lib/agentless.js @@ -0,0 +1,54 @@ +'use strict' + +const { randomUUID } = require('node:crypto') + +const { createHostTransport } = require('./agentless-transport') + +class AgentlessExporter { + #binding + #closed = false + #inFlight = new Set() + + constructor (binding, options) { + const runtimeId = options.runtimeId ?? randomUUID() + const transport = createHostTransport() + this.#binding = new binding.AgentlessExporter( + { ...options, runtimeId }, + transport.request, + transport.cancelRequest, + transport.sleep, + transport.cancelSleep, + ) + } + + sendV04 (payload) { + if (this.#closed) { + return Promise.reject(new Error('data-pipeline exporter is closed')) + } + + let operation + try { + operation = this.#binding.sendV04(payload) + } catch (error) { + operation = Promise.reject(error) + } + this.#inFlight.add(operation) + operation.then( + () => this.#inFlight.delete(operation), + () => this.#inFlight.delete(operation), + ) + return operation + } + + async close () { + this.#closed = true + this.#binding.cancelAll() + await Promise.allSettled(this.#inFlight) + } +} + +function createAgentlessExporter (binding, options) { + return new AgentlessExporter(binding, options) +} + +module.exports = { createAgentlessExporter } diff --git a/packages/libdatadog/lib/native.js b/packages/libdatadog/lib/native.js index 86b430f..02e8a92 100644 --- a/packages/libdatadog/lib/native.js +++ b/packages/libdatadog/lib/native.js @@ -3,6 +3,8 @@ const path = require('node:path') const os = require('node:os') +const { createAgentlessExporter } = require('./agentless') + const target = getNativeTarget() const binding = loadBinding(target) @@ -45,5 +47,6 @@ function loadBinding (target) { module.exports = { backend: () => 'native', DDSketch: binding.DDSketch, + createAgentlessExporter: options => createAgentlessExporter(binding, options), zstd_compress: binding.zstd_compress, } diff --git a/packages/libdatadog/lib/wasm.js b/packages/libdatadog/lib/wasm.js index 8274b43..d77de2c 100644 --- a/packages/libdatadog/lib/wasm.js +++ b/packages/libdatadog/lib/wasm.js @@ -2,8 +2,11 @@ const binding = require('@datadog/libdatadog-wasm') +const { createAgentlessExporter } = require('./agentless') + module.exports = { backend: () => 'wasm', DDSketch: binding.DDSketch, + createAgentlessExporter: options => createAgentlessExporter(binding, options), zstd_compress: binding.zstd_compress, } diff --git a/packages/libdatadog/package-lock.json b/packages/libdatadog/package-lock.json index 5d8940d..fe5079e 100644 --- a/packages/libdatadog/package-lock.json +++ b/packages/libdatadog/package-lock.json @@ -15,6 +15,7 @@ "@datadog/libdatadog-wasm": "0.1.0-pre" }, "devDependencies": { + "@msgpack/msgpack": "^3.1.3", "@napi-rs/cli": "^3.5.1", "esbuild": "^0.28.2", "typescript": "^6.0.3", @@ -935,6 +936,16 @@ "@jridgewell/sourcemap-codec": "^1.4.14" } }, + "node_modules/@msgpack/msgpack": { + "version": "3.1.3", + "resolved": "https://registry.npmjs.org/@msgpack/msgpack/-/msgpack-3.1.3.tgz", + "integrity": "sha512-47XIizs9XZXvuJgoaJUIE2lFoID8ugvc0jzSHP+Ptfk8nTbnR8g788wv48N03Kx0UkAv559HWRQ3yzOgzlRNUA==", + "dev": true, + "license": "ISC", + "engines": { + "node": ">= 18" + } + }, "node_modules/@napi-rs/cli": { "version": "3.8.6", "resolved": "https://registry.npmjs.org/@napi-rs/cli/-/cli-3.8.6.tgz", diff --git a/packages/libdatadog/package.json b/packages/libdatadog/package.json index f26db5e..8303ac8 100644 --- a/packages/libdatadog/package.json +++ b/packages/libdatadog/package.json @@ -69,6 +69,7 @@ "@datadog/libdatadog-linux-x64-musl": "0.1.0-pre" }, "devDependencies": { + "@msgpack/msgpack": "^3.1.3", "@napi-rs/cli": "^3.5.1", "esbuild": "^0.28.2", "typescript": "^6.0.3", diff --git a/packages/libdatadog/test/exporter.test.js b/packages/libdatadog/test/exporter.test.js new file mode 100644 index 0000000..c13fcc8 --- /dev/null +++ b/packages/libdatadog/test/exporter.test.js @@ -0,0 +1,361 @@ +'use strict' + +const assert = require('node:assert/strict') +const { spawnSync } = require('node:child_process') +const fs = require('node:fs') +const http = require('node:http') +const path = require('node:path') +const { test } = require('node:test') +const { Worker } = require('node:worker_threads') +const { zstdDecompressSync } = require('node:zlib') + +const { encode } = require('@msgpack/msgpack') + +const zstdMagic = Buffer.from([0x28, 0xB5, 0x2F, 0xFD]) + +const packageRoot = path.join(__dirname, '..') +const nativeDirectory = path.join(packageRoot, 'dist', 'native') +const nativeArtifact = fs.existsSync(nativeDirectory) + ? fs.readdirSync(nativeDirectory).find(file => file.startsWith('libdatadog.') && file.endsWith('.node')) + : undefined +const wasmArtifact = path.join(packageRoot, 'wasm', 'dist', 'libdatadog_wasm.js') + +test('native backend compresses agentless v0.4 exports with Zstandard', { + skip: !nativeArtifact, +}, async () => { + process.env.DD_LIBDATADOG_NATIVE_PATH = path.join(nativeDirectory, nativeArtifact) + await assertExport( + require('../lib/native'), + 'native', + ) + delete process.env.DD_LIBDATADOG_NATIVE_PATH +}) + +test('inline-WASM backend compresses agentless v0.4 exports with Zstandard', { + skip: !fs.existsSync(wasmArtifact), +}, async () => { + await assertExport( + require('../wasm'), + 'wasm', + ) +}) + +test('inline-WASM backend validates optional values', { + skip: !fs.existsSync(wasmArtifact), +}, async () => { + const pipeline = require('../wasm') + const options = { + endpoint: 'http://127.0.0.1:8126/api/v2/spans', + apiKey: 'test-api-key', + tracerVersion: '0.1.0', + languageVersion: process.version, + languageInterpreter: 'v8', + } + + const exporter = pipeline.createAgentlessExporter({ + ...options, + hostname: null, + env: null, + service: null, + version: null, + runtimeId: null, + containerId: null, + timeoutMs: null, + }) + await exporter.close() + + assert.throws( + () => pipeline.createAgentlessExporter({ ...options, timeoutMs: 1.5 }), + /timeoutMs must be an unsigned integer/, + ) +}) + +const backends = [ + { + name: 'native', + skip: !nativeArtifact, + load: loadNativePipeline, + }, + { + name: 'inline-WASM', + skip: !fs.existsSync(wasmArtifact), + load: () => require('../wasm'), + }, +] + +for (const backend of backends) { + test(`${backend.name} retries in Rust until the third attempt succeeds`, { + skip: backend.skip, + }, async () => { + const pipeline = backend.load() + let requests = 0 + const server = http.createServer((incoming, response) => { + incoming.resume() + incoming.once('end', () => { + requests++ + response.writeHead(requests < 3 ? 500 : 202) + response.end() + }) + }) + + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const exporter = createExporter(pipeline, server) + try { + await exporter.sendV04(tracePayload()) + assert.strictEqual(requests, 3) + } finally { + await exporter.close() + await new Promise(resolve => server.close(resolve)) + } + }) + + test(`${backend.name} applies Rust timeouts and retry policy`, { + skip: backend.skip, + }, async () => { + const pipeline = backend.load() + let requests = 0 + const server = http.createServer((incoming) => { + requests++ + incoming.resume() + }) + + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const exporter = createExporter(pipeline, server, { timeoutMs: 100 }) + try { + await assert.rejects(exporter.sendV04(tracePayload()), /Request timed out/) + assert.strictEqual(requests, 3) + } finally { + await exporter.close() + await new Promise(resolve => server.close(resolve)) + } + }) + + test(`${backend.name} close cancels an active HTTP request`, { + skip: backend.skip, + }, async () => { + const pipeline = backend.load() + let resolveRequest + const request = new Promise((resolve) => { + resolveRequest = resolve + }) + const server = http.createServer((incoming) => { + incoming.resume() + incoming.once('end', resolveRequest) + }) + + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const exporter = createExporter(pipeline, server) + try { + const send = exporter.sendV04(tracePayload()) + await request + await exporter.close() + await assert.rejects(send, /export was cancelled/) + } finally { + await exporter.close() + await new Promise(resolve => server.close(resolve)) + } + }) + + test(`${backend.name} close cancels retry backoff`, { + skip: backend.skip, + }, async () => { + const pipeline = backend.load() + let requests = 0 + let resolveResponse + const responseSent = new Promise((resolve) => { + resolveResponse = resolve + }) + const server = http.createServer((incoming, response) => { + incoming.resume() + incoming.once('end', () => { + requests++ + response.writeHead(500) + response.end(resolveResponse) + }) + }) + + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const exporter = createExporter(pipeline, server) + try { + const send = exporter.sendV04(tracePayload()) + await responseSent + await new Promise(resolve => setTimeout(resolve, 50)) + await exporter.close() + await assert.rejects(send, /export was cancelled/) + await new Promise(resolve => setTimeout(resolve, 1100)) + assert.strictEqual(requests, 1) + } finally { + await exporter.close() + await new Promise(resolve => server.close(resolve)) + } + }) +} + +test('an idle NAPI exporter does not keep Node alive', { + skip: !nativeArtifact, +}, () => { + const nativePath = path.join(nativeDirectory, nativeArtifact) + const modulePath = path.join(packageRoot, 'lib', 'native.js') + const script = ` + const pipeline = require(${JSON.stringify(modulePath)}) + pipeline.createAgentlessExporter({ + endpoint: 'http://127.0.0.1:1/api/v2/spans', + apiKey: 'test-api-key', + tracerVersion: '0.1.0', + languageVersion: process.version, + languageInterpreter: 'v8', + }) + ` + const result = spawnSync(process.execPath, ['-e', script], { + env: { ...process.env, DD_LIBDATADOG_NATIVE_PATH: nativePath }, + encoding: 'utf8', + timeout: 3000, + }) + + assert.notStrictEqual(result.error?.code, 'ETIMEDOUT') + assert.strictEqual(result.status, 0, result.stderr) +}) + +test('NAPI exporter survives worker teardown during an HTTP request', { + skip: !nativeArtifact, +}, async () => { + let resolveRequest + const request = new Promise((resolve) => { + resolveRequest = resolve + }) + const server = http.createServer((incoming) => { + incoming.resume() + incoming.once('end', resolveRequest) + }) + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const { port } = server.address() + const worker = new Worker(` + const { workerData } = require('node:worker_threads') + process.env.DD_LIBDATADOG_NATIVE_PATH = workerData.nativePath + const pipeline = require(workerData.modulePath) + const exporter = pipeline.createAgentlessExporter({ + endpoint: workerData.endpoint, + apiKey: 'test-api-key', + tracerVersion: '0.1.0', + languageVersion: process.version, + languageInterpreter: 'v8', + }) + exporter.sendV04(Buffer.from(workerData.payload)).catch(() => {}) + `, { + eval: true, + workerData: { + nativePath: path.join(nativeDirectory, nativeArtifact), + modulePath: path.join(packageRoot, 'lib', 'native.js'), + endpoint: `http://127.0.0.1:${port}/api/v2/spans`, + payload: tracePayload(), + }, + }) + const workerError = new Promise((_, reject) => { + worker.once('error', reject) + }) + + try { + await Promise.race([request, workerError]) + const exitCode = await worker.terminate() + assert.strictEqual(exitCode, 1) + } finally { + await worker.terminate() + await new Promise(resolve => server.close(resolve)) + } +}) + +async function assertExport (pipeline, expectedBackend) { + const received = await withIntake(async (endpoint) => { + const exporter = pipeline.createAgentlessExporter({ + endpoint, + apiKey: 'test-api-key', + tracerVersion: '0.1.0', + languageVersion: process.version, + languageInterpreter: 'v8', + service: 'service', + containerId: 'container-id', + }) + + try { + await exporter.sendV04(tracePayload()) + } finally { + await exporter.close() + } + }) + + assert.strictEqual(pipeline.backend(), expectedBackend) + assert.strictEqual(received.headers['dd-api-key'], 'test-api-key') + assert.strictEqual(received.headers['datadog-container-id'], 'container-id') + assert.match(received.headers['content-type'], /^application\/json/) + assert.strictEqual(received.headers['content-encoding'], 'zstd') + assert.deepStrictEqual(received.body.subarray(0, zstdMagic.length), zstdMagic) + if (zstdDecompressSync) { + const body = JSON.parse(zstdDecompressSync(received.body).toString()) + assert.strictEqual(body.traces[0].spans[0].name, 'operation') + assert.strictEqual(body.traces[0].spans[0].service, 'service') + } +} + +function tracePayload () { + return encode([[{ + service: 'service', + name: 'operation', + resource: 'resource', + trace_id: 1n, + span_id: 2n, + parent_id: 0n, + start: 1, + duration: 1, + error: 0, + meta: {}, + metrics: {}, + }]], { useBigInt64: true }) +} + +function loadNativePipeline () { + process.env.DD_LIBDATADOG_NATIVE_PATH = path.join(nativeDirectory, nativeArtifact) + try { + return require('../lib/native') + } finally { + delete process.env.DD_LIBDATADOG_NATIVE_PATH + } +} + +function createExporter (pipeline, server, options = {}) { + const { port } = server.address() + return pipeline.createAgentlessExporter({ + endpoint: `http://127.0.0.1:${port}/api/v2/spans`, + apiKey: 'test-api-key', + tracerVersion: '0.1.0', + languageVersion: process.version, + languageInterpreter: 'v8', + ...options, + }) +} + +async function withIntake (send) { + let resolveRequest + const request = new Promise((resolve) => { + resolveRequest = resolve + }) + const server = http.createServer((incoming, response) => { + const chunks = [] + incoming.on('data', chunk => chunks.push(chunk)) + incoming.on('end', () => { + resolveRequest({ + headers: incoming.headers, + body: Buffer.concat(chunks), + }) + response.end() + }) + }) + + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + try { + const { port } = server.address() + await send(`http://127.0.0.1:${port}/api/v2/spans`) + return await request + } finally { + await new Promise(resolve => server.close(resolve)) + } +} diff --git a/packages/libdatadog/test/package-contents.test.js b/packages/libdatadog/test/package-contents.test.js index 8c96051..fdc2d4a 100644 --- a/packages/libdatadog/test/package-contents.test.js +++ b/packages/libdatadog/test/package-contents.test.js @@ -184,22 +184,26 @@ function assertEsmImports (installRoot, environment, expectedBackend) { import assert from 'node:assert/strict' import libdatadog, { backend, + createAgentlessExporter, DDSketch, zstd_compress, } from '@datadog/libdatadog' import wasm, { backend as wasmBackend, + createAgentlessExporter as createWasmAgentlessExporter, DDSketch as WasmDDSketch, zstd_compress as wasmCompress, } from '@datadog/libdatadog/wasm' assert.strictEqual(backend(), ${JSON.stringify(expectedBackend)}) assert.strictEqual(libdatadog.backend, backend) + assert.strictEqual(libdatadog.createAgentlessExporter, createAgentlessExporter) assert(zstd_compress(new Uint8Array(16), 3) instanceof Uint8Array) assert.strictEqual(new DDSketch().count(), 0) assert.strictEqual(wasmBackend(), 'wasm') assert.strictEqual(wasm.backend, wasmBackend) + assert.strictEqual(wasm.createAgentlessExporter, createWasmAgentlessExporter) assert(wasmCompress(new Uint8Array(16), 3) instanceof Uint8Array) assert.strictEqual(new WasmDDSketch().count(), 0) ` diff --git a/packages/libdatadog/test/transport.test.js b/packages/libdatadog/test/transport.test.js new file mode 100644 index 0000000..6e55df6 --- /dev/null +++ b/packages/libdatadog/test/transport.test.js @@ -0,0 +1,82 @@ +'use strict' + +const assert = require('node:assert/strict') +const http = require('node:http') +const { test } = require('node:test') + +const { createHostTransport } = require('../lib/agentless-transport') + +test('host transport rejects when a response is aborted', async () => { + const transport = createHostTransport() + let resolveResponseClosed + const responseClosed = new Promise((resolve) => { + resolveResponseClosed = resolve + }) + const server = http.createServer((request, response) => { + response.writeHead(200, { 'content-length': 100 }) + response.write('partial') + response.once('close', resolveResponseClosed) + setImmediate(() => response.destroy()) + }) + + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + try { + const { port } = server.address() + const result = transport.request({ + id: 1, + url: `http://127.0.0.1:${port}`, + method: 'POST', + headers: [], + body: Buffer.alloc(0), + }).then( + () => ({ status: 'resolved' }), + error => ({ error, status: 'rejected' }), + ) + + await responseClosed + const outcome = await result + + assert.strictEqual(outcome.status, 'rejected') + assert.match(outcome.error.message, /response aborted|aborted/) + } finally { + await new Promise(resolve => server.close(resolve)) + } +}) + +test('host transport cancels an active request', async () => { + const transport = createHostTransport() + let resolveRequest + const received = new Promise((resolve) => { + resolveRequest = resolve + }) + const server = http.createServer((request) => { + request.resume() + request.once('end', resolveRequest) + }) + + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + try { + const { port } = server.address() + const request = transport.request({ + id: 2, + url: `http://127.0.0.1:${port}`, + method: 'POST', + headers: [], + body: Buffer.alloc(0), + }) + await received + transport.cancelRequest(2) + await assert.rejects(request, /request was cancelled/) + } finally { + await new Promise(resolve => server.close(resolve)) + } +}) + +test('host transport cancels a pending timer', async () => { + const transport = createHostTransport() + const sleep = transport.sleep(3, 60_000) + + transport.cancelSleep(3) + + await assert.rejects(sleep, /timer was cancelled/) +}) diff --git a/packages/libdatadog/wasm.mjs b/packages/libdatadog/wasm.mjs index cbd61ba..7674c8e 100644 --- a/packages/libdatadog/wasm.mjs +++ b/packages/libdatadog/wasm.mjs @@ -2,6 +2,7 @@ import libdatadog from './wasm.js' export const { backend, + createAgentlessExporter, DDSketch, zstd_compress, } = libdatadog