From b1f76cfd5a9711f20a4164bb6eed2052afb39c01 Mon Sep 17 00:00:00 2001 From: Roch Devost Date: Thu, 27 Aug 2026 17:51:51 -0400 Subject: [PATCH 1/5] feat(libdatadog): add remote config --- .github/actions/build-test-wasm/action.yaml | 7 +- .github/workflows/build.yml | 1 - .github/workflows/release.yml | 1 - Cargo.lock | 97 ++--- README.md | 4 +- crates/libdatadog-remote-config/Cargo.toml | 28 ++ crates/libdatadog-remote-config/src/lib.rs | 244 ++++++++++++ crates/libdatadog/Cargo.toml | 9 + crates/libdatadog/src/data_pipeline/mod.rs | 207 +++++++++-- crates/libdatadog/src/lib.rs | 1 + crates/libdatadog/src/remote_config.rs | 140 +++++++ crates/remote_config/Cargo.toml | 27 -- crates/remote_config/src/lib.rs | 351 ------------------ packages/libdatadog/README.md | 10 +- packages/libdatadog/index.d.ts | 34 ++ packages/libdatadog/index.mjs | 1 + packages/libdatadog/lib/native.js | 2 + packages/libdatadog/lib/remote-config.js | 38 ++ packages/libdatadog/lib/wasm.js | 2 + .../libdatadog/scripts/report-wasm-size.js | 4 + .../libdatadog/test/package-contents.test.js | 4 + packages/libdatadog/test/package.test.js | 4 +- .../libdatadog/test/remote-config.test.js | 155 ++++++++ packages/libdatadog/test/size-report.test.js | 2 + packages/libdatadog/test/types.test.ts | 22 ++ packages/libdatadog/wasm.mjs | 1 + scripts/build-wasm.js | 1 - scripts/check-dependencies.js | 40 +- test/dependencies.js | 67 +++- test/remote-config.js | 284 -------------- 30 files changed, 1026 insertions(+), 762 deletions(-) create mode 100644 crates/libdatadog-remote-config/Cargo.toml create mode 100644 crates/libdatadog-remote-config/src/lib.rs create mode 100644 crates/libdatadog/src/remote_config.rs delete mode 100644 crates/remote_config/Cargo.toml delete mode 100644 crates/remote_config/src/lib.rs create mode 100644 packages/libdatadog/lib/remote-config.js create mode 100644 packages/libdatadog/test/remote-config.test.js delete mode 100644 test/remote-config.js diff --git a/.github/actions/build-test-wasm/action.yaml b/.github/actions/build-test-wasm/action.yaml index fe006065..9e04e5ff 100644 --- a/.github/actions/build-test-wasm/action.yaml +++ b/.github/actions/build-test-wasm/action.yaml @@ -22,13 +22,12 @@ runs: wasm-pack build --target nodejs ./crates/${{ inputs.crate }} --out-dir ../../prebuilds/${{ inputs.crate }} shell: bash - name: Test WASM - # pipeline and remote_config have top-level node:test suites; the other wasm crates use - # plain test/wasm// scripts. pipeline additionally needs --test-force-exit, since its - # wasm exporter keeps the event loop alive after a flush. + # pipeline has a top-level node:test suite that needs --test-force-exit, since its WASM + # exporter keeps the event loop alive after a flush. The other WASM crates use plain + # test/wasm// scripts. run: | case "${{ inputs.crate }}" in pipeline) node --test --test-force-exit test/pipeline.js ;; - remote_config) node --test test/remote-config.js ;; *) node test-wasm.js ${{ inputs.crate }} ;; esac shell: bash diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index a0fbe3b0..de2fc042 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -259,7 +259,6 @@ jobs: crate: - library_config - pipeline - - remote_config steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 - name: 'Use composite action' diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 2fa71a0c..8b1ad84b 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -44,7 +44,6 @@ jobs: crate: - library_config - pipeline - - remote_config steps: - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 - name: 'Use composite action' diff --git a/Cargo.lock b/Cargo.lock index 689d32b6..55401b3b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1272,14 +1272,20 @@ dependencies = [ "anyhow", "bytes", "futures", + "getrandom 0.2.17", "http 1.5.0", + "js-sys", "libdatadog-data-pipeline", + "libdatadog-remote-config", "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", "libdd-ddsketch 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", "napi 3.12.2", "napi-async-runtime", "napi-build", "napi-derive 3.6.3", + "tokio", + "wasm-bindgen", + "wasm-bindgen-futures", "zrip", "zstd", ] @@ -1309,6 +1315,17 @@ dependencies = [ "wasm-bindgen-test", ] +[[package]] +name = "libdatadog-remote-config" +version = "0.1.0" +dependencies = [ + "anyhow", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-remote-config", + "serde_json", +] + [[package]] name = "libdatadog-wasm-bindgen" version = "0.1.0" @@ -1367,6 +1384,20 @@ dependencies = [ "tokio", ] +[[package]] +name = "libdd-capabilities-impl" +version = "4.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" +dependencies = [ + "anyhow", + "bytes", + "http 1.5.0", + "http-body-util", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "tokio", +] + [[package]] name = "libdd-capabilities-impl" version = "4.0.0" @@ -1430,14 +1461,25 @@ dependencies = [ "futures-util", "hex", "http 1.5.0", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", "libc", "nix 0.29.0", "pin-project", "regex", "regex-lite", + "rustls", + "rustls-native-certs", + "rustls-platform-verifier", "serde", "static_assertions", "thiserror 1.0.69", + "tokio", + "tokio-rustls", + "tower-service", "windows-sys 0.52.0", ] @@ -1458,20 +1500,15 @@ dependencies = [ "http-body", "http-body-util", "hyper", - "hyper-rustls", "hyper-util", "libc", "nix 0.29.0", "pin-project", "regex", - "rustls", - "rustls-native-certs", - "rustls-platform-verifier", "serde", "static_assertions", "thiserror 1.0.69", "tokio", - "tokio-rustls", "tower-service", "windows-sys 0.52.0", ] @@ -1524,7 +1561,7 @@ dependencies = [ "http 1.5.0", "http-body-util", "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", - "libdd-capabilities-impl 4.0.0", + "libdd-capabilities-impl 4.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-dogstatsd-client", @@ -1630,8 +1667,8 @@ dependencies = [ [[package]] name = "libdd-remote-config" -version = "3.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75#966f921c226b63a1ba84c2e53e3d0f4625f90c75" +version = "4.0.0" +source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" dependencies = [ "anyhow", "base64", @@ -1643,10 +1680,10 @@ dependencies = [ "hashbrown 0.15.5", "http 1.5.0", "http-body-util", - "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 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", - "libdd-trace-protobuf 4.0.1", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-capabilities-impl 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-trace-protobuf 5.0.0", "libdd-tuf", "manual_future", "prost", @@ -1692,7 +1729,7 @@ dependencies = [ "futures", "futures-util", "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", - "libdd-capabilities-impl 4.0.0", + "libdd-capabilities-impl 4.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", "tokio-util", @@ -1850,7 +1887,7 @@ dependencies = [ "hashbrown 0.15.5", "http 1.5.0", "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", - "libdd-capabilities-impl 4.0.0", + "libdd-capabilities-impl 4.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-dogstatsd-client", @@ -1884,7 +1921,7 @@ dependencies = [ "indexmap 2.14.0", "itoa 1.0.18", "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", - "libdd-capabilities-impl 4.0.0", + "libdd-capabilities-impl 4.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-tinybytes 1.1.2 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", "libdd-trace-normalization 3.0.1", @@ -2774,23 +2811,6 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" -[[package]] -name = "remote-config" -version = "0.1.0" -dependencies = [ - "console_error_panic_hook", - "js-sys", - "libdatadog-nodejs-capabilities", - "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=966f921c226b63a1ba84c2e53e3d0f4625f90c75)", - "libdd-remote-config", - "serde", - "serde-wasm-bindgen", - "serde_json", - "wasm-bindgen", - "wasm-bindgen-futures", - "wasm-bindgen-test", -] - [[package]] name = "ring" version = "0.17.14" @@ -3059,17 +3079,6 @@ dependencies = [ "serde", ] -[[package]] -name = "serde-wasm-bindgen" -version = "0.4.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3b4c031cd0d9014307d82b8abf653c0290fbdaeb4c02d00c63cf52f728628bf" -dependencies = [ - "js-sys", - "serde", - "wasm-bindgen", -] - [[package]] name = "serde_bytes" version = "0.11.19" @@ -3336,7 +3345,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] diff --git a/README.md b/README.md index 7555a185..5ad0af41 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 the agentless data pipeline, Zstandard compression, and -DDSketch from a single native or WASM artifact. +backends expose remote configuration, the agentless data pipeline, Zstandard +compression, and DDSketch from a single native or WASM artifact. The package uses one napi-rs binding crate for both native and WASM artifacts. Native artifacts are published as optional dependencies using napi-rs-compatible diff --git a/crates/libdatadog-remote-config/Cargo.toml b/crates/libdatadog-remote-config/Cargo.toml new file mode 100644 index 00000000..4b263fd6 --- /dev/null +++ b/crates/libdatadog-remote-config/Cargo.toml @@ -0,0 +1,28 @@ +[package] +name = "libdatadog-remote-config" +version = "0.1.0" +edition = "2021" + +[features] +default = [] +agentless = ["libdd-remote-config/agentless"] +regex-lite = ["libdd-remote-config/regex-lite"] + +[dependencies] +anyhow = "1" +serde_json = "1" + +[dependencies.libdd-capabilities] +git = "https://github.com/DataDog/libdatadog.git" +rev = "7327f3049c281090f7cce7830b9430206fba11cf" + +[dependencies.libdd-common] +git = "https://github.com/DataDog/libdatadog.git" +rev = "7327f3049c281090f7cce7830b9430206fba11cf" +default-features = false + +[dependencies.libdd-remote-config] +git = "https://github.com/DataDog/libdatadog.git" +rev = "7327f3049c281090f7cce7830b9430206fba11cf" +default-features = false +features = ["client"] diff --git a/crates/libdatadog-remote-config/src/lib.rs b/crates/libdatadog-remote-config/src/lib.rs new file mode 100644 index 00000000..19c50ced --- /dev/null +++ b/crates/libdatadog-remote-config/src/lib.rs @@ -0,0 +1,244 @@ +use std::str::FromStr; +use std::sync::Arc; + +use libdd_capabilities::{HttpClientCapability, SleepCapability}; +use libdd_remote_config::fetch::{ + AgentlessConfig, ConfigApplyState, ConfigInvariants, ConfigOptions, SingleChangesFetcher, +}; +use libdd_remote_config::file_change_tracker::{Change, FilePath}; +use libdd_remote_config::file_storage::{RawFile, SimpleFileStorage}; +use libdd_remote_config::{ + RemoteConfigCapabilities, RemoteConfigPath, RemoteConfigProduct, Target, +}; + +type Fetcher = SingleChangesFetcher; + +pub struct RemoteConfigOptions { + pub client_id: String, + pub runtime_id: String, + pub service: String, + pub env: String, + pub app_version: String, + pub tags: Vec, + pub process_tags: Vec, + pub language: String, + pub tracer_version: String, + pub url: String, + pub timeout_ms: u64, + pub api_key: String, + pub hostname: String, +} + +pub struct ChangeRecord { + pub kind: &'static str, + pub path: String, + pub product: String, + pub config_id: String, + pub name: String, + pub version: f64, + pub contents: Option, +} + +#[derive(Default)] +pub struct PendingUpdates { + product_capabilities: Option<(Vec, Vec)>, + extra_services: Option>, + config_states: Vec<(RemoteConfigPath, ConfigApplyState)>, +} + +impl PendingUpdates { + pub fn set_config_state( + &mut self, + path: &str, + apply_state: u32, + apply_error: Option, + ) -> Result<(), String> { + let state = match apply_state { + 1 => ConfigApplyState::Unacknowledged, + 2 => ConfigApplyState::Acknowledged, + 3 => ConfigApplyState::Error(apply_error.unwrap_or_default()), + other => return Err(format!("Unknown apply state {other}")), + }; + let path = RemoteConfigPath::try_parse(path).map_err(|error| error.to_string())?; + self.config_states.push((path.into(), state)); + Ok(()) + } + + pub fn set_extra_services(&mut self, services: Vec) { + self.extra_services = Some(services); + } + + pub fn set_product_capabilities( + &mut self, + products: Vec, + capabilities: Vec, + ) -> Vec { + let mut unknown = vec![]; + let products = products + .into_iter() + .filter_map(|name| match RemoteConfigProduct::from_str(&name) { + Ok(product) => Some(product), + Err(_) => { + unknown.push(name); + None + } + }) + .collect(); + let capabilities = capabilities + .into_iter() + .filter_map(|name| match parse_capability(&name) { + Some(capability) => Some(capability), + None => { + unknown.push(name); + None + } + }) + .collect(); + + self.product_capabilities = Some((products, capabilities)); + unknown + } +} + +struct FetcherConfig { + target: Target, + runtime_id: String, + client_id: String, + invariants: ConfigInvariants, + capabilities: C, +} + +impl FetcherConfig +where + C: HttpClientCapability + SleepCapability, +{ + async fn build(&self) -> anyhow::Result> { + Ok(SingleChangesFetcher::new( + SimpleFileStorage::default(), + self.target.clone(), + self.runtime_id.clone(), + ConfigOptions { + invariants: self.invariants.clone(), + products: vec![], + capabilities: vec![], + }, + self.capabilities.clone(), + ) + .await? + .with_client_id(self.client_id.clone())) + } +} + +pub struct RemoteConfigClient +where + C: HttpClientCapability + SleepCapability, +{ + config: FetcherConfig, + fetcher: Option>, +} + +impl RemoteConfigClient +where + C: HttpClientCapability + SleepCapability, +{ + pub fn new(options: RemoteConfigOptions, capabilities: C) -> Result { + let url = libdd_common::parse_uri(&options.url).map_err(|error| error.to_string())?; + if url.scheme().is_none() || url.authority().is_none() { + return Err(format!( + "Remote config agent URL needs both a scheme and a host: {}", + options.url + )); + } + + let endpoint = libdd_common::Endpoint { + url, + timeout_ms: options.timeout_ms, + api_key: Some(options.api_key.into()), + ..Default::default() + }; + let agentless = + AgentlessConfig::new(options.hostname, &endpoint).map_err(|error| error.to_string())?; + + Ok(Self { + config: FetcherConfig { + target: Target::new( + options.service, + options.env, + options.app_version, + options.tags, + options.process_tags, + ), + runtime_id: options.runtime_id, + client_id: options.client_id, + invariants: ConfigInvariants { + language: options.language, + tracer_version: options.tracer_version, + endpoint, + agentless: Some(agentless), + }, + capabilities, + }, + fetcher: None, + }) + } + + pub async fn fetch_changes( + &mut self, + updates: PendingUpdates, + ) -> anyhow::Result> { + if self.fetcher.is_none() { + self.fetcher = Some(self.config.build().await?); + } + let fetcher = self.fetcher.as_mut().expect("fetcher was initialized"); + + if let Some((products, capabilities)) = updates.product_capabilities { + fetcher.set_product_capabilities(products, capabilities); + } + if let Some(services) = updates.extra_services { + fetcher.set_extra_services(services); + } + for (path, state) in updates.config_states { + fetcher.fetcher.set_config_state(&path, state); + } + + Ok(fetcher + .fetch_changes::>() + .await? + .into_iter() + .map(to_change_record) + .collect()) + } +} + +fn parse_capability(name: &str) -> Option { + serde_json::from_value(serde_json::Value::String(name.to_string())).ok() +} + +fn to_change_record(change: Change>>, Vec>) -> ChangeRecord { + match change { + Change::Add(file) => to_record("add", &file, contents(&file)), + Change::Update(file, _) => to_record("update", &file, contents(&file)), + Change::Remove(file) => to_record("remove", &file, None), + } +} + +fn to_record( + kind: &'static str, + file: &Arc>>, + contents: Option, +) -> ChangeRecord { + let path = file.path(); + ChangeRecord { + kind, + path: path.to_string(), + product: path.product().to_string(), + config_id: path.config_id().to_string(), + name: path.name().to_string(), + version: file.version() as f64, + contents, + } +} + +fn contents(file: &Arc>>) -> Option { + Some(String::from_utf8_lossy(file.contents().as_slice()).into_owned()) +} diff --git a/crates/libdatadog/Cargo.toml b/crates/libdatadog/Cargo.toml index 313f58a8..d735387d 100644 --- a/crates/libdatadog/Cargo.toml +++ b/crates/libdatadog/Cargo.toml @@ -18,11 +18,16 @@ libdd-ddsketch = "1.1.1" [target.'cfg(not(target_arch = "wasm32"))'.dependencies] napi = { version = "3", default-features = false, features = ["async"] } +tokio = { version = "1", features = ["time"] } zstd = { version = "0.13.3", default-features = false, features = ["thin"] } [target.'cfg(target_arch = "wasm32")'.dependencies] +getrandom = { version = "0.2", features = ["js"] } +js-sys = "0.3" napi = { version = "3", default-features = false, features = ["async-runtime"] } napi-async-runtime = "0.2" +wasm-bindgen = "0.2" +wasm-bindgen-futures = "0.4" zrip = { version = "=0.6.0", default-features = false, features = ["alloc"] } [dependencies.libdd-capabilities] @@ -38,5 +43,9 @@ features = ["compression"] path = "../libdatadog-data-pipeline" features = ["compression", "regex-lite"] +[dependencies.libdatadog-remote-config] +path = "../libdatadog-remote-config" +features = ["agentless", "regex-lite"] + [build-dependencies] napi-build = "2" diff --git a/crates/libdatadog/src/data_pipeline/mod.rs b/crates/libdatadog/src/data_pipeline/mod.rs index 1ab402f5..e66263ed 100644 --- a/crates/libdatadog/src/data_pipeline/mod.rs +++ b/crates/libdatadog/src/data_pipeline/mod.rs @@ -4,8 +4,21 @@ use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; +#[cfg(target_arch = "wasm32")] +use std::cell::RefCell; +#[cfg(target_arch = "wasm32")] +use std::future::Future; +#[cfg(target_arch = "wasm32")] +use std::pin::Pin; +#[cfg(target_arch = "wasm32")] +use std::rc::Rc; +#[cfg(target_arch = "wasm32")] +use std::task::{Context, Poll}; + use bytes::Bytes; use futures::future::{AbortHandle, Abortable}; +#[cfg(target_arch = "wasm32")] +use js_sys::{Function as JsFunction, Promise as JsPromise, Reflect}; use libdatadog_data_pipeline::{ send_agentless_v04, AgentlessTraceConfig, SendAgentlessV04Error, TracerMetadata, DEFAULT_AGENTLESS_TIMEOUT, @@ -15,6 +28,10 @@ use napi::bindgen_prelude::*; use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode}; use napi::Status; use napi_derive::napi; +#[cfg(target_arch = "wasm32")] +use wasm_bindgen::{JsCast, JsValue}; +#[cfg(target_arch = "wasm32")] +use wasm_bindgen_futures::JsFuture; type RequestFunction = ThreadsafeFunction< AgentlessRequest, @@ -66,12 +83,16 @@ pub struct AgentlessResponse { } #[derive(Clone)] -struct HostCapabilities { +pub(crate) struct HostCapabilities { + host: Option>, +} + +struct HostFunctions { request: Arc, cancel_request: Arc, sleep: Arc, cancel_sleep: Arc, - next_call_id: Arc, + next_call_id: AtomicU32, } impl fmt::Debug for HostCapabilities { @@ -81,14 +102,47 @@ impl fmt::Debug for HostCapabilities { } impl HostCapabilities { - fn next_call_id(&self) -> u32 { - self.next_call_id.fetch_add(1, Ordering::Relaxed) + pub(crate) fn new( + request: Function<'_, AgentlessRequest, Promise>, + cancel_request: Function<'_, u32, ()>, + sleep: Function<'_, FnArgs<(u32, u32)>, Promise<()>>, + cancel_sleep: Function<'_, u32, ()>, + ) -> Result { + Ok(Self { + host: Some(Arc::new(HostFunctions { + 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: AtomicU32::new(1), + })), + }) } } impl HttpClientCapability for HostCapabilities { fn new_client() -> Self { - panic!("host capabilities must be constructed with JavaScript functions") + Self { host: None } } fn new_without_connection_pooling() -> Self { @@ -99,7 +153,10 @@ impl HttpClientCapability for HostCapabilities { &self, request: http::Request, ) -> std::result::Result, HttpError> { - let id = self.next_call_id(); + let host = self.host.as_ref().ok_or_else(|| { + HttpError::Network(anyhow::anyhow!("host HTTP capability is unavailable")) + })?; + let id = host.next_call_id.fetch_add(1, Ordering::Relaxed); let (parts, body) = request.into_parts(); let headers = parts .headers @@ -121,8 +178,8 @@ impl HttpClientCapability for HostCapabilities { headers, body: body.to_vec().into(), }; - let mut guard = CancelGuard::new(id, self.cancel_request.clone()); - let promise = self + let mut guard = CancelGuard::new(id, host.cancel_request.clone()); + let promise = host .request .call_async(request) .await @@ -139,20 +196,118 @@ impl HttpClientCapability for HostCapabilities { impl SleepCapability for HostCapabilities { fn new() -> Self { - panic!("host capabilities must be constructed with JavaScript functions") + Self { host: None } } async fn sleep(&self, duration: Duration) { - let id = self.next_call_id(); + let Some(host) = &self.host else { + sleep_without_host(duration).await; + return; + }; + let id = host.next_call_id.fetch_add(1, Ordering::Relaxed); 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 mut guard = CancelGuard::new(id, host.cancel_sleep.clone()); + if let Ok(promise) = host.sleep.call_async((id, milliseconds).into()).await { let _ = promise.await; } guard.disarm(); } } +#[cfg(not(target_arch = "wasm32"))] +async fn sleep_without_host(duration: Duration) { + tokio::time::sleep(duration).await; +} + +#[cfg(target_arch = "wasm32")] +async fn sleep_without_host(duration: Duration) { + WasmSendFuture(Box::pin(async move { + let global = js_sys::global(); + let Ok(set_timeout) = Reflect::get(&global, &JsValue::from_str("setTimeout")) + .and_then(|value| value.dyn_into::()) + else { + return; + }; + let clear_timeout = Reflect::get(&global, &JsValue::from_str("clearTimeout")) + .and_then(|value| value.dyn_into::()) + .ok(); + let handle = Rc::new(RefCell::new(None)); + let promise_handle = handle.clone(); + let promise_global = global.clone(); + let milliseconds = duration_millis(duration); + let promise = JsPromise::new(&mut move |resolve, _| { + let result = set_timeout.call2( + &promise_global, + resolve.as_ref(), + &JsValue::from_f64(f64::from(milliseconds)), + ); + match result { + Ok(value) => { + if let Ok(unref) = Reflect::get(&value, &JsValue::from_str("unref")) + .and_then(|value| value.dyn_into::()) + { + let _ = unref.call0(&value); + } + *promise_handle.borrow_mut() = Some(value); + } + Err(_) => { + let _ = resolve.call0(&JsValue::UNDEFINED); + } + } + }); + let mut guard = GlobalTimerGuard { + clear_timeout, + global: global.into(), + handle, + }; + let _ = JsFuture::from(promise).await; + guard.disarm(); + })) + .await; +} + +#[cfg(target_arch = "wasm32")] +struct WasmSendFuture(Pin>>); + +// SAFETY: wasm32-unknown-unknown uses the single-thread NAPI runtime, so this +// future cannot move to another thread while it contains JavaScript values. +#[cfg(target_arch = "wasm32")] +unsafe impl Send for WasmSendFuture {} + +#[cfg(target_arch = "wasm32")] +impl Future for WasmSendFuture { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll { + self.0.as_mut().poll(context) + } +} + +#[cfg(target_arch = "wasm32")] +struct GlobalTimerGuard { + clear_timeout: Option, + global: JsValue, + handle: Rc>>, +} + +#[cfg(target_arch = "wasm32")] +impl GlobalTimerGuard { + fn disarm(&mut self) { + self.handle.borrow_mut().take(); + } +} + +#[cfg(target_arch = "wasm32")] +impl Drop for GlobalTimerGuard { + fn drop(&mut self) { + if let (Some(clear_timeout), Some(handle)) = + (&self.clear_timeout, self.handle.borrow_mut().take()) + { + let _ = clear_timeout.call1(&self.global, &handle); + } + } +} + struct CancelGuard { id: u32, cancel: Arc, @@ -223,33 +378,7 @@ impl AgentlessExporter { 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)), - }; + let capabilities = HostCapabilities::new(request, cancel_request, sleep, cancel_sleep)?; Ok(Self { metadata, diff --git a/crates/libdatadog/src/lib.rs b/crates/libdatadog/src/lib.rs index 604bc5bd..505b6c01 100644 --- a/crates/libdatadog/src/lib.rs +++ b/crates/libdatadog/src/lib.rs @@ -1,4 +1,5 @@ mod data_pipeline; +mod remote_config; mod sketches; mod zstd; diff --git a/crates/libdatadog/src/remote_config.rs b/crates/libdatadog/src/remote_config.rs new file mode 100644 index 00000000..7d627ad3 --- /dev/null +++ b/crates/libdatadog/src/remote_config.rs @@ -0,0 +1,140 @@ +use std::sync::{Arc, Mutex, MutexGuard}; + +use futures::lock::Mutex as AsyncMutex; +use libdatadog_remote_config::{ + ChangeRecord, PendingUpdates, RemoteConfigClient, RemoteConfigOptions, +}; +use napi::bindgen_prelude::*; +use napi_derive::napi; + +use crate::data_pipeline::{AgentlessRequest, AgentlessResponse, HostCapabilities}; + +#[napi(object)] +pub struct RemoteConfigFetcherOptions { + pub client_id: String, + pub runtime_id: String, + pub service: String, + pub env: String, + pub app_version: String, + pub tags: Vec, + pub process_tags: Vec, + pub language: String, + pub tracer_version: String, + pub url: String, + pub timeout_ms: u32, + pub api_key: String, + pub hostname: String, +} + +impl From for RemoteConfigOptions { + fn from(options: RemoteConfigFetcherOptions) -> Self { + Self { + client_id: options.client_id, + runtime_id: options.runtime_id, + service: options.service, + env: options.env, + app_version: options.app_version, + tags: options.tags, + process_tags: options.process_tags, + language: options.language, + tracer_version: options.tracer_version, + url: options.url, + timeout_ms: u64::from(options.timeout_ms), + api_key: options.api_key, + hostname: options.hostname, + } + } +} + +#[napi(object)] +pub struct RemoteConfigChange { + pub kind: String, + pub path: String, + pub product: String, + pub config_id: String, + pub name: String, + pub version: f64, + pub contents: Option, +} + +impl From for RemoteConfigChange { + fn from(change: ChangeRecord) -> Self { + Self { + kind: change.kind.to_string(), + path: change.path, + product: change.product, + config_id: change.config_id, + name: change.name, + version: change.version, + contents: change.contents, + } + } +} + +#[napi] +pub struct RemoteConfigFetcher { + client: Arc>>, + pending: Arc>, +} + +#[napi] +impl RemoteConfigFetcher { + #[napi(constructor)] + pub fn new( + options: RemoteConfigFetcherOptions, + request: Function<'_, AgentlessRequest, Promise>, + cancel_request: Function<'_, u32, ()>, + sleep: Function<'_, FnArgs<(u32, u32)>, Promise<()>>, + cancel_sleep: Function<'_, u32, ()>, + ) -> Result { + let capabilities = HostCapabilities::new(request, cancel_request, sleep, cancel_sleep)?; + let client = + RemoteConfigClient::new(options.into(), capabilities).map_err(Error::from_reason)?; + + Ok(Self { + client: Arc::new(AsyncMutex::new(client)), + pending: Arc::new(Mutex::new(PendingUpdates::default())), + }) + } + + #[napi] + pub async fn fetch_changes(&self) -> Result> { + let updates = std::mem::take(&mut *lock(&self.pending)); + let mut client = self.client.lock().await; + client + .fetch_changes(updates) + .await + .map(|changes| changes.into_iter().map(Into::into).collect()) + .map_err(|error| Error::from_reason(error.to_string())) + } + + #[napi] + pub fn set_config_state( + &self, + path: String, + apply_state: u32, + apply_error: Option, + ) -> Result<()> { + lock(&self.pending) + .set_config_state(&path, apply_state, apply_error) + .map_err(Error::from_reason) + } + + #[napi] + pub fn set_extra_services(&self, services: Vec) { + lock(&self.pending).set_extra_services(services); + } + + #[napi] + pub fn set_product_capabilities( + &self, + products: Vec, + capabilities: Vec, + ) -> Vec { + lock(&self.pending).set_product_capabilities(products, capabilities) + } +} + +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex.lock().unwrap_or_else(|error| error.into_inner()) +} diff --git a/crates/remote_config/Cargo.toml b/crates/remote_config/Cargo.toml deleted file mode 100644 index 4512e8aa..00000000 --- a/crates/remote_config/Cargo.toml +++ /dev/null @@ -1,27 +0,0 @@ -[package] -name = "remote-config" -version = "0.1.0" -edition = "2021" -description = "Wasm binding for libdatadog's remote config client" - -[lib] -crate-type = ["cdylib", "rlib"] - -[dependencies] -libdd-remote-config = { git = "https://github.com/DataDog/libdatadog.git", rev = "966f921c226b63a1ba84c2e53e3d0f4625f90c75", default-features = false, features = ["client", "agentless"] } -libdd-common = { git = "https://github.com/DataDog/libdatadog.git", rev = "966f921c226b63a1ba84c2e53e3d0f4625f90c75", default-features = false } -libdatadog-nodejs-capabilities = { path = "../capabilities" } - -wasm-bindgen = "0.2" -js-sys = "0.3" -wasm-bindgen-futures = "0.4" -serde = { version = "1.0", features = ["derive"] } -serde-wasm-bindgen = "0.4" -serde_json = "1" -console_error_panic_hook = "0.1" - -[package.metadata.wasm-pack.profile.release] -wasm-opt = ["-O", "--all-features"] - -[dev-dependencies] -wasm-bindgen-test = "0.3" diff --git a/crates/remote_config/src/lib.rs b/crates/remote_config/src/lib.rs deleted file mode 100644 index 6c69eef8..00000000 --- a/crates/remote_config/src/lib.rs +++ /dev/null @@ -1,351 +0,0 @@ -//! Wasm binding for libdatadog's remote config client. - -use std::cell::RefCell; -use std::rc::Rc; -use std::str::FromStr; -use std::sync::Arc; - -use libdatadog_nodejs_capabilities::{HttpClientCapability, WasmCapabilities}; -use libdd_remote_config::fetch::{ - AgentlessConfig, ConfigApplyState, ConfigInvariants, ConfigOptions, SingleChangesFetcher, -}; -use libdd_remote_config::file_change_tracker::{Change, FilePath}; -use libdd_remote_config::file_storage::{RawFile, SimpleFileStorage}; -use libdd_remote_config::{ - RemoteConfigCapabilities, RemoteConfigPath, RemoteConfigProduct, Target, -}; -use serde::{Deserialize, Serialize}; -use wasm_bindgen::prelude::*; - -#[wasm_bindgen(start)] -fn init() { - console_error_panic_hook::set_once(); -} - -const APPLY_STATE_UNACKNOWLEDGED: u32 = 1; -const APPLY_STATE_ACKNOWLEDGED: u32 = 2; -const APPLY_STATE_ERROR: u32 = 3; - -type Fetcher = SingleChangesFetcher; - -/// A real `Error`, not a bare string: the JS side logs rejections with a logger that identifies a -/// cause by its `stack`, and the pure-JS fallback client rejects with `Error`s too. -fn to_js_err(err: impl std::fmt::Display) -> JsValue { - js_sys::Error::new(&err.to_string()).into() -} - -/// Deserializing capabilities one name at a time keeps an unrecognized one from discarding -/// the whole set. -fn parse_capability(name: &str) -> Option { - serde_json::from_value(serde_json::Value::String(name.to_string())).ok() -} - -/// Options accepted by [`RemoteConfigFetcher::new`]. These populate the `Client`/`ClientTracer` of -/// the remote config request. -#[derive(Deserialize)] -#[serde(rename_all = "camelCase")] -struct FetcherOptions { - client_id: String, - runtime_id: String, - service: String, - env: String, - app_version: String, - /// Already-formatted `"key:value"` strings. - tags: Vec, - /// Already-formatted `"key:value"` strings. - process_tags: Vec, - language: String, - tracer_version: String, - /// In agent mode the agent base URL, either `http(s)://host:port` or `unix:///path/to/socket`; - /// in agentless mode the Datadog site, e.g. `https://api.datadoghq.com`. - url: String, - timeout_ms: u64, - /// Enables agentless if set. - #[serde(default)] - api_key: Option, - #[serde(default)] - hostname: Option, -} - -/// A single add/update/remove of one remote config file, as diffed against the previous poll. -#[derive(Serialize)] -#[serde(rename_all = "camelCase")] -struct ChangeRecord { - /// One of `"add"`, `"update"`, `"remove"`. - kind: &'static str, - /// Full unparsed remote config path, e.g. `datadog/2/APM_TRACING/config-id/name` or - /// `employee/APM_TRACING/config-id/name`. Pass this back to `setConfigState`. - path: String, - product: String, - config_id: String, - name: String, - version: f64, - /// The hash-verified file contents. Omitted for `"remove"`: the consumer already holds them from - /// the `"add"`/`"update"` that introduced the config. - #[serde(skip_serializing_if = "Option::is_none")] - contents: Option, -} - -fn to_record( - kind: &'static str, - file: &Arc>>, - contents: Option, -) -> ChangeRecord { - let path = file.path(); - ChangeRecord { - kind, - path: path.to_string(), - product: path.product().to_string(), - config_id: path.config_id().to_string(), - name: path.name().to_string(), - // Read after `contents`: both lock the same mutex, which is not reentrant. - version: file.version() as f64, - contents, - } -} - -fn to_contents(file: &Arc>>) -> Option { - // Lossy like `Buffer#toString('utf8')`: a config file that is not valid UTF-8 is not valid JSON - // either, so the consumer rejects it either way. - Some(String::from_utf8_lossy(file.contents().as_slice()).into_owned()) -} - -fn to_change_record(change: Change>>, Vec>) -> ChangeRecord { - match change { - Change::Add(file) => to_record("add", &file, to_contents(&file)), - Change::Update(file, _old_contents) => to_record("update", &file, to_contents(&file)), - Change::Remove(file) => to_record("remove", &file, None), - } -} - -/// Mutations recorded by the setters while a poll may be in flight, applied at the start of the next -/// one. `fetchChanges` holds the fetcher borrow across its whole round trip, so the setters must not -/// take it -- they would find it already borrowed and have nowhere to put the update. -#[derive(Default)] -struct PendingUpdates { - product_capabilities: Option<(Vec, Vec)>, - extra_services: Option>, - config_states: Vec<(RemoteConfigPath, ConfigApplyState)>, -} - -/// Everything needed to build the fetcher, kept around because building it is deferred. -struct FetcherConfig { - target: Target, - runtime_id: String, - client_id: String, - invariants: ConfigInvariants, -} - -impl FetcherConfig { - async fn build(&self) -> Result { - Ok(SingleChangesFetcher::new( - SimpleFileStorage::default(), - self.target.clone(), - self.runtime_id.clone(), - ConfigOptions { - invariants: self.invariants.clone(), - // Products and capabilities are only known once the subsystems that own them have - // registered their handlers, which always happens before the first poll. - products: vec![], - capabilities: vec![], - }, - WasmCapabilities::new_without_connection_pooling(), - ) - .await - .map_err(to_js_err)? - .with_client_id(self.client_id.clone())) - } -} - -/// The fetcher is built on the first poll rather than in the constructor to be able to await.. -struct FetcherState { - config: FetcherConfig, - fetcher: Option, -} - -#[wasm_bindgen] -pub struct RemoteConfigFetcher { - state: Rc>, - pending: Rc>, -} - -#[wasm_bindgen] -impl RemoteConfigFetcher { - #[wasm_bindgen(constructor)] - pub fn new(options: JsValue) -> Result { - let options: FetcherOptions = serde_wasm_bindgen::from_value(options).map_err(to_js_err)?; - - let url = libdd_common::parse_uri(&options.url).map_err(to_js_err)?; - - // Validate, otherwise libdatadog will panic here - if url.scheme().is_none() || url.authority().is_none() { - return Err(to_js_err(format!( - "Remote config agent URL needs both a scheme and a host: {}", - options.url - ))); - } - - let endpoint = libdd_common::Endpoint { - url, - timeout_ms: options.timeout_ms, - api_key: options.api_key.map(Into::into), - ..Default::default() - }; - - let agentless = match endpoint.api_key { - Some(_) => Some( - AgentlessConfig::new(options.hostname.unwrap_or_default(), &endpoint) - .map_err(to_js_err)?, - ), - None => None, - }; - - let config = FetcherConfig { - target: Target::new( - options.service, - options.env, - options.app_version, - options.tags, - options.process_tags, - ), - runtime_id: options.runtime_id, - client_id: options.client_id, - invariants: ConfigInvariants { - language: options.language, - tracer_version: options.tracer_version, - endpoint, - agentless, - }, - }; - - Ok(RemoteConfigFetcher { - state: Rc::new(RefCell::new(FetcherState { - config, - fetcher: None, - })), - pending: Rc::new(RefCell::new(PendingUpdates::default())), - }) - } - - /// Polls the agent once and resolves with the changes (add/update/remove) relative to the - /// previous successful poll. An empty array means nothing changed. - /// - /// Holding the borrow across the await is deliberate: on wasm there is one thread, and the - /// borrow is what serializes polls. - #[allow(clippy::await_holding_refcell_ref)] - #[wasm_bindgen(js_name = "fetchChanges")] - pub async fn fetch_changes(&self) -> Result { - let cell = self.state.clone(); - let pending = self.pending.clone(); - - // The scheduler above this only arms the next poll once the previous settled, so an - // overlapping call is a guard, not a path. - let mut state = cell - .try_borrow_mut() - .map_err(|_| to_js_err("A remote config poll is already in flight"))?; - let state = &mut *state; - - let fetcher = match &mut state.fetcher { - Some(fetcher) => fetcher, - slot => slot.insert(state.config.build().await?), - }; - - let updates = std::mem::take(&mut *pending.borrow_mut()); - if let Some((products, capabilities)) = updates.product_capabilities { - fetcher.set_product_capabilities(products, capabilities); - } - if let Some(services) = updates.extra_services { - fetcher.set_extra_services(services); - } - for (path, state) in updates.config_states { - fetcher.fetcher.set_config_state(&path, state); - } - - let changes = fetcher - .fetch_changes::>() - .await - .map_err(to_js_err)?; - - let records: Vec = changes.into_iter().map(to_change_record).collect(); - - serde_wasm_bindgen::to_value(&records).map_err(to_js_err) - } - - /// Reports the apply outcome of a previously received change, identified by the `path` handed - /// back by `fetchChanges`. `applyState` is one of `APPLY_STATE_*` constants. - #[wasm_bindgen(js_name = "setConfigState")] - pub fn set_config_state( - &self, - path: String, - apply_state: u32, - apply_error: Option, - ) -> Result<(), JsValue> { - let state = match apply_state { - APPLY_STATE_UNACKNOWLEDGED => ConfigApplyState::Unacknowledged, - APPLY_STATE_ACKNOWLEDGED => ConfigApplyState::Acknowledged, - APPLY_STATE_ERROR => ConfigApplyState::Error(apply_error.unwrap_or_default()), - other => return Err(to_js_err(format!("Unknown apply state {other}"))), - }; - let path = RemoteConfigPath::try_parse(&path).map_err(to_js_err)?; - - self.pending - .borrow_mut() - .config_states - .push((path.into(), state)); - - Ok(()) - } - - /// Replaces the set of extra services reported to the agent. - #[wasm_bindgen(js_name = "setExtraServices")] - pub fn set_extra_services(&self, services: Vec) { - self.pending.borrow_mut().extra_services = Some(services); - } - - /// Replaces the set of subscribed products and capabilities. - /// - /// Names this build does not know are skipped and returned, so that a tracer whose own list has - /// moved ahead of libdatadog's keeps working with the names that do resolve. Reporting them is - /// left to the caller, which owns the logger. - #[wasm_bindgen(js_name = "setProductCapabilities")] - pub fn set_product_capabilities( - &self, - products: Vec, - capabilities: Vec, - ) -> Vec { - let mut unknown = vec![]; - - let products = products - .into_iter() - .filter_map(|name| match RemoteConfigProduct::from_str(&name) { - Ok(product) => Some(product), - Err(_) => { - unknown.push(name); - None - } - }) - .collect(); - - let capabilities = capabilities - .into_iter() - .filter_map(|name| match parse_capability(&name) { - Some(capability) => Some(capability), - None => { - unknown.push(name); - None - } - }) - .collect(); - - self.pending.borrow_mut().product_capabilities = Some((products, capabilities)); - - unknown - } -} - -/// Installs the host's async-context hook, so the HTTP request this module issues is not -/// re-instrumented by the tracer's own http plugin. -#[wasm_bindgen(js_name = "setStorage")] -pub fn set_storage(new_storage: &JsValue) { - libdatadog_nodejs_capabilities::http::set_storage(new_storage); -} diff --git a/packages/libdatadog/README.md b/packages/libdatadog/README.md index b58c8449..4c46e431 100644 --- a/packages/libdatadog/README.md +++ b/packages/libdatadog/README.md @@ -10,8 +10,14 @@ from the single root `crates/libdatadog` napi-rs workspace crate. Optional libdatadog functionality is published separately as `@datadog/libdatadog-extras`. -Both backends expose the agentless data pipeline, Zstandard compression, and -DDSketch from a single native or WASM artifact. +Both backends expose remote configuration, the agentless data pipeline, +Zstandard compression, and DDSketch from a single native or WASM artifact. + +Remote configuration connects directly to the Datadog backend and verifies +responses with TUF. Both bindings use the smaller regex implementation and +Node's host-managed HTTP transport. Reusing Node's HTTPS implementation avoids +shipping another TLS stack in native packages and keeps networking behavior +consistent across the NAPI and WASM backends. The package accepts Datadog v0.4 MessagePack payloads and exports them to an agentless intake. diff --git a/packages/libdatadog/index.d.ts b/packages/libdatadog/index.d.ts index 6aec92a1..098fb03d 100644 --- a/packages/libdatadog/index.d.ts +++ b/packages/libdatadog/index.d.ts @@ -21,6 +21,40 @@ export interface AgentlessExporter { export function createAgentlessExporter(options: AgentlessExporterOptions): AgentlessExporter export function backend(): 'native' | 'wasm' +export interface RemoteConfigFetcherOptions { + clientId: string + runtimeId: string + service: string + env: string + appVersion: string + tags: string[] + processTags: string[] + language: string + tracerVersion: string + url: string + timeoutMs: number + apiKey: string + hostname: string +} + +export interface RemoteConfigChange { + kind: 'add' | 'update' | 'remove' + path: string + product: string + configId: string + name: string + version: number + contents?: string +} + +export class RemoteConfigFetcher { + constructor(options: RemoteConfigFetcherOptions) + fetchChanges(): Promise + setConfigState(path: string, applyState: number, applyError?: string): void + setExtraServices(services: string[]): void + setProductCapabilities(products: string[], capabilities: string[]): string[] +} + export function zstd_compress(data: Uint8Array, level: number): Uint8Array export class DDSketch { diff --git a/packages/libdatadog/index.mjs b/packages/libdatadog/index.mjs index 17e46561..4a7b4c31 100644 --- a/packages/libdatadog/index.mjs +++ b/packages/libdatadog/index.mjs @@ -4,6 +4,7 @@ export const { backend, createAgentlessExporter, DDSketch, + RemoteConfigFetcher, zstd_compress, } = libdatadog diff --git a/packages/libdatadog/lib/native.js b/packages/libdatadog/lib/native.js index 02e8a926..106251ce 100644 --- a/packages/libdatadog/lib/native.js +++ b/packages/libdatadog/lib/native.js @@ -4,6 +4,7 @@ const path = require('node:path') const os = require('node:os') const { createAgentlessExporter } = require('./agentless') +const { remoteConfigFetcher } = require('./remote-config') const target = getNativeTarget() const binding = loadBinding(target) @@ -47,6 +48,7 @@ function loadBinding (target) { module.exports = { backend: () => 'native', DDSketch: binding.DDSketch, + RemoteConfigFetcher: remoteConfigFetcher(binding), createAgentlessExporter: options => createAgentlessExporter(binding, options), zstd_compress: binding.zstd_compress, } diff --git a/packages/libdatadog/lib/remote-config.js b/packages/libdatadog/lib/remote-config.js new file mode 100644 index 00000000..f07d95b7 --- /dev/null +++ b/packages/libdatadog/lib/remote-config.js @@ -0,0 +1,38 @@ +'use strict' + +const { createHostTransport } = require('./agentless-transport') + +function remoteConfigFetcher (binding) { + return class RemoteConfigFetcher { + #binding + + constructor (options) { + const transport = createHostTransport() + this.#binding = new binding.RemoteConfigFetcher( + options, + transport.request, + transport.cancelRequest, + transport.sleep, + transport.cancelSleep, + ) + } + + fetchChanges () { + return this.#binding.fetchChanges() + } + + setConfigState (path, applyState, applyError) { + return this.#binding.setConfigState(path, applyState, applyError) + } + + setExtraServices (services) { + return this.#binding.setExtraServices(services) + } + + setProductCapabilities (products, capabilities) { + return this.#binding.setProductCapabilities(products, capabilities) + } + } +} + +module.exports = { remoteConfigFetcher } diff --git a/packages/libdatadog/lib/wasm.js b/packages/libdatadog/lib/wasm.js index d77de2c5..743a7f71 100644 --- a/packages/libdatadog/lib/wasm.js +++ b/packages/libdatadog/lib/wasm.js @@ -3,10 +3,12 @@ const binding = require('@datadog/libdatadog-wasm') const { createAgentlessExporter } = require('./agentless') +const { remoteConfigFetcher } = require('./remote-config') module.exports = { backend: () => 'wasm', DDSketch: binding.DDSketch, + RemoteConfigFetcher: remoteConfigFetcher(binding), createAgentlessExporter: options => createAgentlessExporter(binding, options), zstd_compress: binding.zstd_compress, } diff --git a/packages/libdatadog/scripts/report-wasm-size.js b/packages/libdatadog/scripts/report-wasm-size.js index 1b4a4a66..49f5c692 100644 --- a/packages/libdatadog/scripts/report-wasm-size.js +++ b/packages/libdatadog/scripts/report-wasm-size.js @@ -34,6 +34,10 @@ const forbiddenWasmCode = [ dependency: 'zstd-sys', owners: new Set(['zstd-sys', 'zstd-sys (C)']), }, + { + dependency: 'tokio', + owners: new Set(['tokio', 'tokio-util']), + }, ] function readUnsignedLeb128 (bytes, start) { diff --git a/packages/libdatadog/test/package-contents.test.js b/packages/libdatadog/test/package-contents.test.js index 503ef11e..202301f0 100644 --- a/packages/libdatadog/test/package-contents.test.js +++ b/packages/libdatadog/test/package-contents.test.js @@ -191,24 +191,28 @@ function assertEsmImports (installRoot, environment, expectedBackend) { backend, createAgentlessExporter, DDSketch, + RemoteConfigFetcher, zstd_compress, } from '@datadog/libdatadog' import wasm, { backend as wasmBackend, createAgentlessExporter as createWasmAgentlessExporter, DDSketch as WasmDDSketch, + RemoteConfigFetcher as WasmRemoteConfigFetcher, 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.strictEqual(libdatadog.RemoteConfigFetcher, RemoteConfigFetcher) 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.strictEqual(wasm.RemoteConfigFetcher, WasmRemoteConfigFetcher) assert(wasmCompress(new Uint8Array(16), 3) instanceof Uint8Array) assert.strictEqual(new WasmDDSketch().count(), 0) ` diff --git a/packages/libdatadog/test/package.test.js b/packages/libdatadog/test/package.test.js index 63bb4456..9b823f05 100644 --- a/packages/libdatadog/test/package.test.js +++ b/packages/libdatadog/test/package.test.js @@ -111,8 +111,8 @@ test('embeds a Brotli-compressed WASM fallback below the size budgets', () => { const runtime = brotliDecompressSync(Buffer.from(encodedRuntime, 'base64')).toString() const compressedWasm = Buffer.from(encodedWasm, 'base64') const wasm = brotliDecompressSync(compressedWasm) - assert.ok(Buffer.byteLength(glue) < 300 * 1024) - assert.ok(wasm.length < 500 * 1024) + assert.ok(Buffer.byteLength(glue) < 480 * 1024) + assert.ok(wasm.length < 1024 * 1024) assert.doesNotMatch(wasm.toString('latin1'), /wasi_snapshot_preview1/) assert.doesNotMatch(glue, /node:(?:wasi|worker_threads)/) assert.doesNotMatch(runtime, /node:(?:wasi|worker_threads)/) diff --git a/packages/libdatadog/test/remote-config.test.js b/packages/libdatadog/test/remote-config.test.js new file mode 100644 index 00000000..0e03694d --- /dev/null +++ b/packages/libdatadog/test/remote-config.test.js @@ -0,0 +1,155 @@ +'use strict' + +/* eslint-disable unicorn/prefer-event-target -- Node stream mocks use EventEmitter. */ + +const assert = require('node:assert/strict') +const { execFileSync } = require('node:child_process') +const { EventEmitter } = require('node:events') +const https = require('node:https') +const { test } = require('node:test') + +const selected = require('..') +const wasm = require('../wasm') +const backends = selected === wasm + ? [['WASM', wasm]] + : [['native', selected], ['WASM', wasm]] + +const CONFIG_PATH = 'datadog/2/ASM_FEATURES/asm-features-1/config' + +function fetcherOptions (overrides = {}) { + return { + clientId: 'client-id-1', + runtimeId: 'runtime-id-1', + service: 'my_svc', + env: 'my_env', + appVersion: '1.0.0', + tags: ['runtime-id:runtime-id-1'], + processTags: ['entrypoint.type:script'], + language: 'nodejs', + tracerVersion: '1.2.3', + url: 'https://datadoghq.com', + timeoutMs: 5000, + apiKey: 'test-api-key', + hostname: 'test-host', + ...overrides, + } +} + +function mockHttps (context) { + const requests = [] + context.mock.method(https, 'request', (url, options, onResponse) => { + const request = new EventEmitter() + request.destroy = error => request.emit('error', error) + request.end = (body) => { + requests.push({ body: Buffer.from(body), options, url: url.toString() }) + queueMicrotask(() => { + const response = new EventEmitter() + response.statusCode = 200 + onResponse(response) + response.emit('end') + }) + } + return request + }) + return requests +} + +test('WASM agentless remote config does not require WebCrypto', () => { + const entryPoint = require.resolve('../wasm') + const script = ` + delete globalThis.crypto + if (globalThis.crypto !== undefined) throw new Error('could not hide WebCrypto') + const { EventEmitter } = require('node:events') + const https = require('node:https') + let requested = false + https.request = (_url, _options, onResponse) => { + const request = new EventEmitter() + request.destroy = error => request.emit('error', error) + request.end = () => { + requested = true + queueMicrotask(() => { + const response = new EventEmitter() + response.statusCode = 200 + onResponse(response) + response.emit('end') + }) + } + return request + } + const { RemoteConfigFetcher } = require(${JSON.stringify(entryPoint)}) + new RemoteConfigFetcher(${JSON.stringify(fetcherOptions())}) + .fetchChanges() + .then(() => { throw new Error('invalid TUF response was accepted') }) + .catch(() => { + if (!requested) throw new Error('agentless request was not sent') + process.stdout.write('ok') + }) + ` + + assert.strictEqual( + execFileSync(process.execPath, ['--eval', script], { encoding: 'utf8' }), + 'ok', + ) +}) + +for (const [name, { RemoteConfigFetcher }] of backends) { + test(`${name} remote config sends directly to the backend`, async (context) => { + const requests = mockHttps(context) + const fetcher = new RemoteConfigFetcher(fetcherOptions()) + assert.deepStrictEqual( + fetcher.setProductCapabilities( + ['ASM_FEATURES', 'ASM_DD'], + ['ASM_ACTIVATION', 'ASM_DD_RULES'], + ), + [], + ) + fetcher.setExtraServices(['other_svc']) + + await assert.rejects(fetcher.fetchChanges(), /missing config meta/) + assert.strictEqual(requests.length, 2) + const request = requests.find(({ url }) => url.endsWith('/api/v0.1/configurations')) + assert.ok(request) + assert.strictEqual( + request.url, + 'https://config.datadoghq.com/api/v0.1/configurations', + ) + assert.strictEqual(request.options.headers['dd-api-key'], 'test-api-key') + assert.strictEqual(request.options.method, 'POST') + assert.ok(request.body.length > 0) + }) + + test(`${name} remote config validates input`, () => { + const fetcher = new RemoteConfigFetcher(fetcherOptions()) + assert.deepStrictEqual( + fetcher.setProductCapabilities( + ['ASM_FEATURES', 'NOT_A_PRODUCT'], + ['ASM_ACTIVATION', 'NOT_A_CAPABILITY'], + ), + ['NOT_A_PRODUCT', 'NOT_A_CAPABILITY'], + ) + assert.throws( + () => fetcher.setConfigState(CONFIG_PATH, 42), + /Unknown apply state 42/, + ) + assert.throws( + () => new RemoteConfigFetcher(fetcherOptions({ url: 'http://datadoghq.com' })), + /agentless endpoint is invalid/, + ) + assert.throws( + () => new RemoteConfigFetcher(fetcherOptions({ hostname: '' })), + /hostname is empty/, + ) + }) + + test(`${name} remote config enforces its request timeout`, async (context) => { + context.mock.method(https, 'request', () => { + const request = new EventEmitter() + request.destroy = error => request.emit('error', error) + request.end = () => {} + return request + }) + + const fetcher = new RemoteConfigFetcher(fetcherOptions({ timeoutMs: 1 })) + await assert.rejects(fetcher.fetchChanges(), /timed out after 1ms/) + }) +} diff --git a/packages/libdatadog/test/size-report.test.js b/packages/libdatadog/test/size-report.test.js index 86ba4821..f9b973fc 100644 --- a/packages/libdatadog/test/size-report.test.js +++ b/packages/libdatadog/test/size-report.test.js @@ -94,8 +94,10 @@ test('rejects forbidden code linked into WASM', () => { { bytes: 20, name: 'regex-automata' }, { bytes: 30, name: 'zstd-sys (C)' }, { bytes: 40, name: 'zrip-encode' }, + { bytes: 50, name: 'tokio' }, ]), [ { bytes: 20, dependency: 'regex', name: 'regex-automata' }, { bytes: 30, dependency: 'zstd-sys', name: 'zstd-sys (C)' }, + { bytes: 50, dependency: 'tokio', name: 'tokio' }, ]) }) diff --git a/packages/libdatadog/test/types.test.ts b/packages/libdatadog/test/types.test.ts index f921b6f6..98ddd1f2 100644 --- a/packages/libdatadog/test/types.test.ts +++ b/packages/libdatadog/test/types.test.ts @@ -1,6 +1,7 @@ import { backend, DDSketch, + RemoteConfigFetcher, zstd_compress, } from '@datadog/libdatadog' import * as wasm from '@datadog/libdatadog/wasm' @@ -8,20 +9,41 @@ import * as wasm from '@datadog/libdatadog/wasm' const selectedBackend: 'native' | 'wasm' = backend() const compressed: Uint8Array = zstd_compress(new Uint8Array(16), 3) const sketch = new DDSketch() +const remoteConfig = new RemoteConfigFetcher({ + clientId: 'client-id', + runtimeId: 'runtime-id', + service: 'service', + env: 'test', + appVersion: '1.0.0', + tags: [], + processTags: [], + language: 'nodejs', + tracerVersion: '1.2.3', + url: 'https://datadoghq.com', + timeoutMs: 5_000, + apiKey: 'test-api-key', + hostname: 'test-host', +}) sketch.add(1) sketch.addWithCount(2, 3) const count: number = sketch.count() const encoded: Uint8Array = sketch.encode() +const changes = remoteConfig.fetchChanges() +remoteConfig.setExtraServices(['other-service']) +remoteConfig.setProductCapabilities(['ASM_FEATURES'], ['ASM_ACTIVATION']) const wasmBackend: typeof backend = wasm.backend const wasmSketch: typeof DDSketch = wasm.DDSketch const wasmCompress: typeof zstd_compress = wasm.zstd_compress +const wasmRemoteConfig: typeof RemoteConfigFetcher = wasm.RemoteConfigFetcher void selectedBackend void compressed void count void encoded +void changes void wasmBackend void wasmSketch void wasmCompress +void wasmRemoteConfig diff --git a/packages/libdatadog/wasm.mjs b/packages/libdatadog/wasm.mjs index 7674c8e2..fc2e59c7 100644 --- a/packages/libdatadog/wasm.mjs +++ b/packages/libdatadog/wasm.mjs @@ -4,6 +4,7 @@ export const { backend, createAgentlessExporter, DDSketch, + RemoteConfigFetcher, zstd_compress, } = libdatadog diff --git a/scripts/build-wasm.js b/scripts/build-wasm.js index a6717f93..4c290926 100644 --- a/scripts/build-wasm.js +++ b/scripts/build-wasm.js @@ -19,7 +19,6 @@ const isMacOS = os.platform() === 'darwin' const libraries = [ 'library_config', 'pipeline', - 'remote_config', ] const env = { diff --git a/scripts/check-dependencies.js b/scripts/check-dependencies.js index 1fa4ca02..daaa3bc9 100644 --- a/scripts/check-dependencies.js +++ b/scripts/check-dependencies.js @@ -9,6 +9,28 @@ const trees = [ ...packageJson.napi.targets.map(target => ({ package: 'libdatadog', target })), { package: 'libdatadog', target: 'wasm32-unknown-unknown' }, ] +// TODO: Remove these exceptions after porting the Datadog TUF changes onto +// modern upstream TUF and aligning the remaining remote config dependencies. +const remoteConfigDuplicatePackages = new Set([ + 'getrandom', + 'hashbrown', + 'http', + 'itoa', + 'syn', + 'thiserror', + 'thiserror-impl', + 'untrusted', +]) +// libdd-remote-config exposes a single-client API but still compiles its Tokio +// scheduler modules. The symbolized WASM report separately prevents Tokio code +// from reaching the shipped fallback. +const remoteConfigTokioPackages = new Set(['tokio', 'tokio-macros', 'tokio-util']) +// The existing agentless feature enables libdd-common/https across the native +// feature graph. LTO removes its unused implementation from the shipped addon. +const remoteConfigNativeFeatureUnion = new Set([ + ...remoteConfigTokioPackages, + 'tokio-rustls', +]) function parseCargoTree (output) { const paths = [] @@ -39,13 +61,18 @@ function findDuplicateVersions (dependencies) { } for (const [name, versions] of versionsByPackage) { - if (versions.size > 1) { + if (versions.size > 1 && !isRemoteConfigDuplicate(dependencies, name)) { failures.push({ name, versions: [...versions] }) } } return failures } +function isRemoteConfigDuplicate (dependencies, name) { + return remoteConfigDuplicatePackages.has(name) + && dependencies.some(({ path }) => path.includes('libdd-remote-config')) +} + function findForbiddenDependencies (dependencies, tree) { const failures = [] @@ -58,7 +85,14 @@ function findForbiddenDependencies (dependencies, tree) { const allowedNapiBridge = tree.package === 'libdatadog' && isTokio && parent === 'napi' - if (!allowedNapiBridge) failures.push(dependency) + const allowedRemoteConfigRuntime = remoteConfigTokioPackages.has(dependency.name) + && dependency.path.includes('libdd-remote-config') + const allowedNativeFeatureUnion = tree.target !== 'wasm32-unknown-unknown' + && remoteConfigNativeFeatureUnion.has(dependency.name) + && dependencies.some(({ name }) => name === 'libdd-remote-config') + if (!allowedNapiBridge && !allowedRemoteConfigRuntime && !allowedNativeFeatureUnion) { + failures.push(dependency) + } } return failures @@ -112,7 +146,7 @@ function checkTrees () { ) } } else { - console.log('Tokio is limited to the NAPI async bridge in every artifact.') + console.log('Tokio is limited to the NAPI bridge and remote config runtime.') } if (duplicateFailures.length > 0 || forbiddenFailures.length > 0) { diff --git a/test/dependencies.js b/test/dependencies.js index 5f5e3c35..aca135ee 100644 --- a/test/dependencies.js +++ b/test/dependencies.js @@ -42,11 +42,57 @@ test('dependency validation allows the NAPI Tokio bridge in WASM', () => { ].join('\n')) assert.deepStrictEqual( - findForbiddenDependencies(dependencies, { package: 'libdatadog' }), + findForbiddenDependencies( + dependencies, + { package: 'libdatadog', target: 'wasm32-unknown-unknown' }, + ), + [], + ) +}) + +test('dependency validation allows the remote config Tokio runtime', () => { + const dependencies = parseCargoTree([ + '0libdatadog v0.1.0', + '1libdatadog-remote-config v0.1.0', + '2libdd-remote-config v4.0.0', + '3tokio v1.53.1', + '4tokio-macros v2.7.2', + '3tokio-util v0.7.19', + ].join('\n')) + + assert.deepStrictEqual( + findForbiddenDependencies( + dependencies, + { package: 'libdatadog', target: 'wasm32-unknown-unknown' }, + ), [], ) }) +test('dependency validation allows the native HTTPS feature union', () => { + const dependencies = parseCargoTree([ + '0libdatadog v0.1.0', + '1libdd-remote-config v4.0.0', + '1libdd-common v5.2.0', + '2tokio-rustls v0.26.4', + ].join('\n')) + + assert.deepStrictEqual( + findForbiddenDependencies(dependencies, { + package: 'libdatadog', + target: 'x86_64-unknown-linux-gnu', + }), + [], + ) + assert.strictEqual( + findForbiddenDependencies(dependencies, { + package: 'libdatadog', + target: 'wasm32-unknown-unknown', + }).length, + 1, + ) +}) + test('dependency validation still finds multiple versions in one artifact tree', () => { const dependencies = parseCargoTree([ '0libdatadog v0.1.0', @@ -60,3 +106,22 @@ test('dependency validation still finds multiple versions in one artifact tree', versions: ['1.10.0', '1.11.0'], }]) }) + +test('dependency validation allows known remote config duplicate crates', () => { + const dependencies = parseCargoTree([ + '0libdatadog v0.1.0', + '1syn v2.0.119', + '1libdd-remote-config v4.0.0', + '2syn v3.0.4', + ].join('\n')) + + assert.deepStrictEqual(findDuplicateVersions(dependencies), []) + + dependencies.push({ + depth: 2, + name: 'syn', + path: ['libdatadog', 'libdd-remote-config', 'syn'], + version: '3.0.5', + }) + assert.deepStrictEqual(findDuplicateVersions(dependencies), []) +}) diff --git a/test/remote-config.js b/test/remote-config.js deleted file mode 100644 index a5e94524..00000000 --- a/test/remote-config.js +++ /dev/null @@ -1,284 +0,0 @@ -'use strict' - -const assert = require('node:assert') -const { createHash } = require('node:crypto') -const { execFileSync } = require('node:child_process') -const { createServer } = require('node:http') -const { test } = require('node:test') - -const libdatadog = require('..') -const { RemoteConfigFetcher } = libdatadog.load('remote_config') -assert(RemoteConfigFetcher !== undefined) - -const APPLY_STATE_ACKNOWLEDGED = 2 -const APPLY_STATE_ERROR = 3 - -const CONFIG_PATH = 'datadog/2/ASM_FEATURES/asm-features-1/config' - -/** - * Builds a `/v0.7/config` response body: a base64 encoded TUF-like `targets` document plus the - * base64 encoded files themselves, which is what the agent sends. - */ -function agentResponse (configs, targetsVersion) { - const targets = {} - const targetFiles = [] - - for (const { path, file, version } of configs) { - const raw = Buffer.from(JSON.stringify(file), 'utf8') - targets[path] = { - custom: { v: version }, - hashes: { sha256: createHash('sha256').update(raw).digest('hex') }, - length: raw.length, - } - targetFiles.push({ path, raw: raw.toString('base64') }) - } - - return JSON.stringify({ - client_configs: configs.map(({ path }) => path), - targets: Buffer.from(JSON.stringify({ - signatures: [], - signed: { - _type: 'targets', - custom: { agent_refresh_interval: 5, opaque_backend_state: `backend-state-${targetsVersion}` }, - expires: '2100-01-01T00:00:00.000000000Z', - spec_version: '1.0.0', - targets, - version: targetsVersion, - }, - }), 'utf8').toString('base64'), - target_files: targetFiles, - }) -} - -function fetcherOptions (overrides) { - return { - clientId: 'client-id-1', - runtimeId: 'runtime-id-1', - service: 'my_svc', - env: 'my_env', - appVersion: '1.0.0', - tags: ['runtime-id:runtime-id-1'], - processTags: ['entrypoint.type:script'], - language: 'nodejs', - tracerVersion: '1.2.3', - url: 'http://127.0.0.1:8126', - timeoutMs: 5000, - ...overrides, - } -} - -async function withAgent (run) { - const requests = [] - const responses = [] - - const server = createServer((req, res) => { - const chunks = [] - req - .on('data', chunk => chunks.push(chunk)) - .on('end', () => { - requests.push(JSON.parse(Buffer.concat(chunks).toString('utf8'))) - res.writeHead(200, { 'content-type': 'application/json' }) - res.end(responses.shift() ?? '{}') - }) - }) - - await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) - - const fetcher = new RemoteConfigFetcher(fetcherOptions({ - url: `http://127.0.0.1:${server.address().port}`, - })) - - try { - await run({ fetcher, requests, responses }) - } finally { - await new Promise(resolve => server.close(resolve)) - } -} - -test('reports the client identity, products and capabilities', async () => { - await withAgent(async ({ fetcher, requests }) => { - fetcher.setProductCapabilities(['ASM_FEATURES', 'ASM_DD'], ['ASM_ACTIVATION', 'ASM_DD_RULES']) - fetcher.setExtraServices(['other_svc']) - - assert.deepStrictEqual(await fetcher.fetchChanges(), []) - - const { client } = requests[0] - - assert.strictEqual(client.id, 'client-id-1') - assert.strictEqual(client.is_tracer, true) - assert.deepStrictEqual(client.products, ['ASM_FEATURES', 'ASM_DD']) - // Bits 1 (ASM_ACTIVATION) and 3 (ASM_DD_RULES) of a big-endian octet string. - assert.deepStrictEqual(client.capabilities, [0b1010]) - assert.strictEqual(client.client_tracer.language, 'nodejs') - assert.strictEqual(client.client_tracer.service, 'my_svc') - assert.deepStrictEqual(client.client_tracer.extra_services, ['other_svc']) - assert.deepStrictEqual(client.client_tracer.process_tags, ['entrypoint.type:script']) - }) -}) - -test('diffs successive polls into add, update and remove changes', async () => { - await withAgent(async ({ fetcher, responses }) => { - fetcher.setProductCapabilities(['ASM_FEATURES'], []) - - responses.push(agentResponse([{ path: CONFIG_PATH, file: { asm: { enabled: true } }, version: 1 }], 1)) - - const added = await fetcher.fetchChanges() - assert.strictEqual(added.length, 1) - assert.strictEqual(added[0].kind, 'add') - assert.strictEqual(added[0].path, CONFIG_PATH) - assert.strictEqual(added[0].product, 'ASM_FEATURES') - assert.strictEqual(added[0].configId, 'asm-features-1') - assert.strictEqual(added[0].name, 'config') - assert.strictEqual(added[0].version, 1) - assert.deepStrictEqual(JSON.parse(added[0].contents), { asm: { enabled: true } }) - - // An unchanged config is not reported again. - responses.push(agentResponse([{ path: CONFIG_PATH, file: { asm: { enabled: true } }, version: 1 }], 2)) - assert.deepStrictEqual(await fetcher.fetchChanges(), []) - - responses.push(agentResponse([{ path: CONFIG_PATH, file: { asm: { enabled: false } }, version: 2 }], 3)) - - const updated = await fetcher.fetchChanges() - assert.strictEqual(updated.length, 1) - assert.strictEqual(updated[0].kind, 'update') - assert.strictEqual(updated[0].version, 2) - assert.deepStrictEqual(JSON.parse(updated[0].contents), { asm: { enabled: false } }) - - responses.push(agentResponse([], 4)) - - const removed = await fetcher.fetchChanges() - assert.strictEqual(removed.length, 1) - assert.strictEqual(removed[0].kind, 'remove') - assert.strictEqual(removed[0].path, CONFIG_PATH) - assert.strictEqual(removed[0].contents, undefined) - }) -}) - -test('sends the apply state set for a config on the next poll', async () => { - await withAgent(async ({ fetcher, requests, responses }) => { - fetcher.setProductCapabilities(['ASM_FEATURES'], []) - - responses.push(agentResponse([{ path: CONFIG_PATH, file: { asm: { enabled: true } }, version: 1 }], 1)) - await fetcher.fetchChanges() - - fetcher.setConfigState(CONFIG_PATH, APPLY_STATE_ERROR, 'Error: could not apply') - - responses.push(agentResponse([{ path: CONFIG_PATH, file: { asm: { enabled: true } }, version: 1 }], 2)) - await fetcher.fetchChanges() - - assert.deepStrictEqual(requests[1].client.state.config_states, [{ - id: 'asm-features-1', - version: 1, - product: 'ASM_FEATURES', - apply_state: APPLY_STATE_ERROR, - apply_error: 'Error: could not apply', - }]) - assert.deepStrictEqual(requests[1].cached_target_files, [{ - path: CONFIG_PATH, - length: 24, - hashes: [{ - algorithm: 'sha256', - hash: createHash('sha256').update(JSON.stringify({ asm: { enabled: true } })).digest('hex'), - }], - }]) - - fetcher.setConfigState(CONFIG_PATH, APPLY_STATE_ACKNOWLEDGED, '') - - responses.push(agentResponse([{ path: CONFIG_PATH, file: { asm: { enabled: true } }, version: 1 }], 3)) - await fetcher.fetchChanges() - - assert.strictEqual(requests[2].client.state.config_states[0].apply_state, APPLY_STATE_ACKNOWLEDGED) - assert.strictEqual(requests[2].client.state.config_states[0].apply_error, '') - }) -}) - -test('skips unknown products and capabilities, and rejects bad apply states', async () => { - await withAgent(async ({ fetcher }) => { - // Names this build does not know are skipped and returned, so that a tracer whose own lists - // have moved ahead of libdatadog's keeps working with the names that do resolve. - assert.deepStrictEqual( - fetcher.setProductCapabilities(['ASM_FEATURES', 'NOT_A_PRODUCT'], ['ASM_ACTIVATION', 'NOT_A_CAPABILITY']), - ['NOT_A_PRODUCT', 'NOT_A_CAPABILITY'], - ) - - assert.throws(() => fetcher.setConfigState(CONFIG_PATH, 42, ''), { message: /Unknown apply state 42/ }) - assert.throws(() => fetcher.setConfigState('nonsense', APPLY_STATE_ACKNOWLEDGED, '')) - }) -}) - -test('polls on a runtime without a WebCrypto global', () => { - // libdatadog identifies the client with a UUID, and `uuid`'s wasm RNG is bound to - // `globalThis.crypto` with no fallback -- a global Node only exposes inside a module from v20, so - // it panicked there and trapped the whole module. libdatadog draws those bytes through - // `getrandom` instead, which reaches Node's `crypto` module when the global is missing. - // - // Polling rather than constructing: the id is generated when the fetcher is built, which is - // deferred to the first poll, so a constructor alone proves nothing. A child process, because - // the global has to be absent before the module loads. - const script = ` - delete globalThis.crypto - if (globalThis.crypto !== undefined) throw new Error('could not hide the global') - const { createServer } = require('node:http') - const { RemoteConfigFetcher } = require(${JSON.stringify(require.resolve('..'))}).load('remote_config') - const server = createServer((req, res) => { res.writeHead(200); res.end('{}') }) - server.listen(0, '127.0.0.1', async () => { - const options = ${JSON.stringify(fetcherOptions())} - options.url = 'http://127.0.0.1:' + server.address().port - const changes = await new RemoteConfigFetcher(options).fetchChanges() - process.stdout.write('polled ' + JSON.stringify(changes)) - server.close() - }) - ` - - const out = execFileSync(process.execPath, ['-e', script], { encoding: 'utf8' }) - - assert.strictEqual(out, 'polled []') -}) - -test('rejects an agent url without a scheme and host', () => { - // libdatadog appends the remote config path to this URI and unwraps the result, so a scheme - // without an authority would panic -- and abort the process, since release builds do not unwind. - // `agent-host:8126` is the reachable shape: JavaScript's `new URL` accepts it too. - assert.throws( - () => new RemoteConfigFetcher(fetcherOptions({ url: 'agent-host:8126' })), - { message: /needs both a scheme and a host/ }, - ) -}) - -test('rejects a failed poll', async () => { - // Only the URL parsing is exercised here: no socket is listening, so the poll fails. - const fetcher = new RemoteConfigFetcher(fetcherOptions({ - url: 'unix:///tmp/definitely-not-a-datadog-apm-socket', - })) - - await assert.rejects(fetcher.fetchChanges()) -}) - -test('polls the backend directly when an api key is set', async () => { - const fetcher = new RemoteConfigFetcher(fetcherOptions({ - url: 'https://api.example.invalid', - apiKey: 'an-api-key', - hostname: 'my-host', - })) - - // The backend cannot be faked: its responses are verified against TUF roots embedded in - // libdatadog. What the failure does show is that the poll went to the configs endpoint derived - // from the site URL rather than to the URL itself, which only happens in agentless mode. - await assert.rejects(fetcher.fetchChanges(), { message: /config\.api\.example\.invalid/ }) -}) - -test('rejects an api key without an https site url', () => { - // Talking to the backend without the agent in between means there is no localhost hop to be - // plaintext on, so libdatadog refuses anything but https. - assert.throws( - () => new RemoteConfigFetcher(fetcherOptions({ apiKey: 'an-api-key', hostname: 'my-host' })), - { message: /agentless endpoint is invalid/ }, - ) -}) - -test('rejects an api key without a hostname', () => { - assert.throws( - () => new RemoteConfigFetcher(fetcherOptions({ url: 'https://api.example.invalid', apiKey: 'an-api-key' })), - { message: /hostname is empty/ }, - ) -}) From 0d4abbc0c008d1c9e3f696e8c7593079ba705488 Mon Sep 17 00:00:00 2001 From: Roch Devost Date: Thu, 27 Aug 2026 18:44:01 -0400 Subject: [PATCH 2/5] refactor(libdatadog): reuse remote config host timer --- Cargo.lock | 59 ++++--- crates/libdatadog-data-pipeline/Cargo.toml | 6 +- crates/libdatadog-remote-config/Cargo.toml | 6 +- crates/libdatadog/Cargo.toml | 6 +- crates/libdatadog/src/data_pipeline/mod.rs | 196 ++++----------------- scripts/check-dependencies.js | 3 - 6 files changed, 75 insertions(+), 201 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 55401b3b..d9d2ea0e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -809,6 +809,11 @@ name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +dependencies = [ + "allocator-api2", + "equivalent", + "foldhash 0.2.0", +] [[package]] name = "heck" @@ -1274,18 +1279,14 @@ dependencies = [ "futures", "getrandom 0.2.17", "http 1.5.0", - "js-sys", "libdatadog-data-pipeline", "libdatadog-remote-config", - "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", "libdd-ddsketch 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)", "napi 3.12.2", "napi-async-runtime", "napi-build", "napi-derive 3.6.3", - "tokio", - "wasm-bindgen", - "wasm-bindgen-futures", "zrip", "zstd", ] @@ -1294,7 +1295,7 @@ dependencies = [ name = "libdatadog-data-pipeline" version = "0.1.0" dependencies = [ - "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", "libdd-data-pipeline-core", "libdd-trace-utils 11.0.0", "thiserror 1.0.69", @@ -1320,8 +1321,8 @@ name = "libdatadog-remote-config" version = "0.1.0" dependencies = [ "anyhow", - "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", - "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", "libdd-remote-config", "serde_json", ] @@ -1348,7 +1349,7 @@ dependencies = [ [[package]] name = "libdd-capabilities" version = "3.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" +source = "git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df#3fa1d1395022d3f3fcff6215cc8aa22d5df616df" dependencies = [ "anyhow", "bytes", @@ -1387,14 +1388,14 @@ dependencies = [ [[package]] name = "libdd-capabilities-impl" version = "4.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" +source = "git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df#3fa1d1395022d3f3fcff6215cc8aa22d5df616df" dependencies = [ "anyhow", "bytes", "http 1.5.0", "http-body-util", - "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", - "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", "tokio", ] @@ -1450,7 +1451,7 @@ dependencies = [ [[package]] name = "libdd-common" version = "5.2.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" +source = "git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df#3fa1d1395022d3f3fcff6215cc8aa22d5df616df" dependencies = [ "anyhow", "bytes", @@ -1586,11 +1587,11 @@ dependencies = [ [[package]] name = "libdd-data-pipeline-core" version = "1.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" +source = "git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df#3fa1d1395022d3f3fcff6215cc8aa22d5df616df" dependencies = [ "http 1.5.0", - "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", - "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", "libdd-trace-utils 11.0.0", "serde_json", "thiserror 1.0.69", @@ -1668,7 +1669,7 @@ dependencies = [ [[package]] name = "libdd-remote-config" version = "4.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" +source = "git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df#3fa1d1395022d3f3fcff6215cc8aa22d5df616df" dependencies = [ "anyhow", "base64", @@ -1677,12 +1678,12 @@ dependencies = [ "futures", "futures-util", "getrandom 0.2.17", - "hashbrown 0.15.5", + "hashbrown 0.17.1", "http 1.5.0", "http-body-util", - "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", - "libdd-capabilities-impl 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", - "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", + "libdd-capabilities-impl 4.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", "libdd-trace-protobuf 5.0.0", "libdd-tuf", "manual_future", @@ -1695,7 +1696,7 @@ dependencies = [ "sha2", "strum", "strum_macros", - "thiserror 2.0.20", + "thiserror 1.0.69", "time", "tokio", "tokio-util", @@ -1798,7 +1799,7 @@ dependencies = [ [[package]] name = "libdd-tinybytes" version = "1.1.2" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" +source = "git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df#3fa1d1395022d3f3fcff6215cc8aa22d5df616df" dependencies = [ "serde", ] @@ -1823,7 +1824,7 @@ dependencies = [ [[package]] name = "libdd-trace-normalization" version = "4.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" +source = "git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df#3fa1d1395022d3f3fcff6215cc8aa22d5df616df" dependencies = [ "anyhow", "libdd-trace-protobuf 5.0.0", @@ -1868,7 +1869,7 @@ dependencies = [ [[package]] name = "libdd-trace-protobuf" version = "5.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" +source = "git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df#3fa1d1395022d3f3fcff6215cc8aa22d5df616df" dependencies = [ "prost", "serde", @@ -1943,7 +1944,7 @@ dependencies = [ [[package]] name = "libdd-trace-utils" version = "11.0.0" -source = "git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf#7327f3049c281090f7cce7830b9430206fba11cf" +source = "git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df#3fa1d1395022d3f3fcff6215cc8aa22d5df616df" dependencies = [ "anyhow", "base64", @@ -1956,9 +1957,9 @@ dependencies = [ "http-body-util", "indexmap 2.14.0", "itoa 1.0.18", - "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", - "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", - "libdd-tinybytes 1.1.2 (git+https://github.com/DataDog/libdatadog.git?rev=7327f3049c281090f7cce7830b9430206fba11cf)", + "libdd-capabilities 3.0.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", + "libdd-common 5.2.0 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", + "libdd-tinybytes 1.1.2 (git+https://github.com/DataDog/libdatadog.git?rev=3fa1d1395022d3f3fcff6215cc8aa22d5df616df)", "libdd-trace-normalization 4.0.0", "libdd-trace-protobuf 5.0.0", "prost", diff --git a/crates/libdatadog-data-pipeline/Cargo.toml b/crates/libdatadog-data-pipeline/Cargo.toml index 1769b9df..245fb54c 100644 --- a/crates/libdatadog-data-pipeline/Cargo.toml +++ b/crates/libdatadog-data-pipeline/Cargo.toml @@ -9,17 +9,17 @@ thiserror = "1" [dependencies.libdd-capabilities] git = "https://github.com/DataDog/libdatadog.git" -rev = "7327f3049c281090f7cce7830b9430206fba11cf" +rev = "3fa1d1395022d3f3fcff6215cc8aa22d5df616df" default-features = false [dependencies.libdd-data-pipeline-core] git = "https://github.com/DataDog/libdatadog.git" -rev = "7327f3049c281090f7cce7830b9430206fba11cf" +rev = "3fa1d1395022d3f3fcff6215cc8aa22d5df616df" default-features = false [dependencies.libdd-trace-utils] git = "https://github.com/DataDog/libdatadog.git" -rev = "7327f3049c281090f7cce7830b9430206fba11cf" +rev = "3fa1d1395022d3f3fcff6215cc8aa22d5df616df" default-features = false [features] diff --git a/crates/libdatadog-remote-config/Cargo.toml b/crates/libdatadog-remote-config/Cargo.toml index 4b263fd6..f3250e5b 100644 --- a/crates/libdatadog-remote-config/Cargo.toml +++ b/crates/libdatadog-remote-config/Cargo.toml @@ -14,15 +14,15 @@ serde_json = "1" [dependencies.libdd-capabilities] git = "https://github.com/DataDog/libdatadog.git" -rev = "7327f3049c281090f7cce7830b9430206fba11cf" +rev = "3fa1d1395022d3f3fcff6215cc8aa22d5df616df" [dependencies.libdd-common] git = "https://github.com/DataDog/libdatadog.git" -rev = "7327f3049c281090f7cce7830b9430206fba11cf" +rev = "3fa1d1395022d3f3fcff6215cc8aa22d5df616df" default-features = false [dependencies.libdd-remote-config] git = "https://github.com/DataDog/libdatadog.git" -rev = "7327f3049c281090f7cce7830b9430206fba11cf" +rev = "3fa1d1395022d3f3fcff6215cc8aa22d5df616df" default-features = false features = ["client"] diff --git a/crates/libdatadog/Cargo.toml b/crates/libdatadog/Cargo.toml index d735387d..c55ab907 100644 --- a/crates/libdatadog/Cargo.toml +++ b/crates/libdatadog/Cargo.toml @@ -18,21 +18,17 @@ libdd-ddsketch = "1.1.1" [target.'cfg(not(target_arch = "wasm32"))'.dependencies] napi = { version = "3", default-features = false, features = ["async"] } -tokio = { version = "1", features = ["time"] } zstd = { version = "0.13.3", default-features = false, features = ["thin"] } [target.'cfg(target_arch = "wasm32")'.dependencies] getrandom = { version = "0.2", features = ["js"] } -js-sys = "0.3" napi = { version = "3", default-features = false, features = ["async-runtime"] } napi-async-runtime = "0.2" -wasm-bindgen = "0.2" -wasm-bindgen-futures = "0.4" zrip = { version = "=0.6.0", default-features = false, features = ["alloc"] } [dependencies.libdd-capabilities] git = "https://github.com/DataDog/libdatadog.git" -rev = "7327f3049c281090f7cce7830b9430206fba11cf" +rev = "3fa1d1395022d3f3fcff6215cc8aa22d5df616df" default-features = false [target.'cfg(not(target_arch = "wasm32"))'.dependencies.libdatadog-data-pipeline] diff --git a/crates/libdatadog/src/data_pipeline/mod.rs b/crates/libdatadog/src/data_pipeline/mod.rs index e66263ed..91bcced1 100644 --- a/crates/libdatadog/src/data_pipeline/mod.rs +++ b/crates/libdatadog/src/data_pipeline/mod.rs @@ -4,21 +4,8 @@ use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; -#[cfg(target_arch = "wasm32")] -use std::cell::RefCell; -#[cfg(target_arch = "wasm32")] -use std::future::Future; -#[cfg(target_arch = "wasm32")] -use std::pin::Pin; -#[cfg(target_arch = "wasm32")] -use std::rc::Rc; -#[cfg(target_arch = "wasm32")] -use std::task::{Context, Poll}; - use bytes::Bytes; use futures::future::{AbortHandle, Abortable}; -#[cfg(target_arch = "wasm32")] -use js_sys::{Function as JsFunction, Promise as JsPromise, Reflect}; use libdatadog_data_pipeline::{ send_agentless_v04, AgentlessTraceConfig, SendAgentlessV04Error, TracerMetadata, DEFAULT_AGENTLESS_TIMEOUT, @@ -28,10 +15,6 @@ use napi::bindgen_prelude::*; use napi::threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode}; use napi::Status; use napi_derive::napi; -#[cfg(target_arch = "wasm32")] -use wasm_bindgen::{JsCast, JsValue}; -#[cfg(target_arch = "wasm32")] -use wasm_bindgen_futures::JsFuture; type RequestFunction = ThreadsafeFunction< AgentlessRequest, @@ -84,15 +67,11 @@ pub struct AgentlessResponse { #[derive(Clone)] pub(crate) struct HostCapabilities { - host: Option>, -} - -struct HostFunctions { request: Arc, cancel_request: Arc, sleep: Arc, cancel_sleep: Arc, - next_call_id: AtomicU32, + next_call_id: Arc, } impl fmt::Debug for HostCapabilities { @@ -109,40 +88,42 @@ impl HostCapabilities { cancel_sleep: Function<'_, u32, ()>, ) -> Result { Ok(Self { - host: Some(Arc::new(HostFunctions { - 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: AtomicU32::new(1), - })), + 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)), }) } + + fn next_call_id(&self) -> u32 { + self.next_call_id.fetch_add(1, Ordering::Relaxed) + } } impl HttpClientCapability for HostCapabilities { fn new_client() -> Self { - Self { host: None } + panic!("host capabilities must be constructed with JavaScript functions") } fn new_without_connection_pooling() -> Self { @@ -153,10 +134,7 @@ impl HttpClientCapability for HostCapabilities { &self, request: http::Request, ) -> std::result::Result, HttpError> { - let host = self.host.as_ref().ok_or_else(|| { - HttpError::Network(anyhow::anyhow!("host HTTP capability is unavailable")) - })?; - let id = host.next_call_id.fetch_add(1, Ordering::Relaxed); + let id = self.next_call_id(); let (parts, body) = request.into_parts(); let headers = parts .headers @@ -178,8 +156,8 @@ impl HttpClientCapability for HostCapabilities { headers, body: body.to_vec().into(), }; - let mut guard = CancelGuard::new(id, host.cancel_request.clone()); - let promise = host + let mut guard = CancelGuard::new(id, self.cancel_request.clone()); + let promise = self .request .call_async(request) .await @@ -196,118 +174,20 @@ impl HttpClientCapability for HostCapabilities { impl SleepCapability for HostCapabilities { fn new() -> Self { - Self { host: None } + panic!("host capabilities must be constructed with JavaScript functions") } async fn sleep(&self, duration: Duration) { - let Some(host) = &self.host else { - sleep_without_host(duration).await; - return; - }; - let id = host.next_call_id.fetch_add(1, Ordering::Relaxed); + let id = self.next_call_id(); let milliseconds = duration_millis(duration); - let mut guard = CancelGuard::new(id, host.cancel_sleep.clone()); - if let Ok(promise) = host.sleep.call_async((id, milliseconds).into()).await { + 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(); } } -#[cfg(not(target_arch = "wasm32"))] -async fn sleep_without_host(duration: Duration) { - tokio::time::sleep(duration).await; -} - -#[cfg(target_arch = "wasm32")] -async fn sleep_without_host(duration: Duration) { - WasmSendFuture(Box::pin(async move { - let global = js_sys::global(); - let Ok(set_timeout) = Reflect::get(&global, &JsValue::from_str("setTimeout")) - .and_then(|value| value.dyn_into::()) - else { - return; - }; - let clear_timeout = Reflect::get(&global, &JsValue::from_str("clearTimeout")) - .and_then(|value| value.dyn_into::()) - .ok(); - let handle = Rc::new(RefCell::new(None)); - let promise_handle = handle.clone(); - let promise_global = global.clone(); - let milliseconds = duration_millis(duration); - let promise = JsPromise::new(&mut move |resolve, _| { - let result = set_timeout.call2( - &promise_global, - resolve.as_ref(), - &JsValue::from_f64(f64::from(milliseconds)), - ); - match result { - Ok(value) => { - if let Ok(unref) = Reflect::get(&value, &JsValue::from_str("unref")) - .and_then(|value| value.dyn_into::()) - { - let _ = unref.call0(&value); - } - *promise_handle.borrow_mut() = Some(value); - } - Err(_) => { - let _ = resolve.call0(&JsValue::UNDEFINED); - } - } - }); - let mut guard = GlobalTimerGuard { - clear_timeout, - global: global.into(), - handle, - }; - let _ = JsFuture::from(promise).await; - guard.disarm(); - })) - .await; -} - -#[cfg(target_arch = "wasm32")] -struct WasmSendFuture(Pin>>); - -// SAFETY: wasm32-unknown-unknown uses the single-thread NAPI runtime, so this -// future cannot move to another thread while it contains JavaScript values. -#[cfg(target_arch = "wasm32")] -unsafe impl Send for WasmSendFuture {} - -#[cfg(target_arch = "wasm32")] -impl Future for WasmSendFuture { - type Output = (); - - fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll { - self.0.as_mut().poll(context) - } -} - -#[cfg(target_arch = "wasm32")] -struct GlobalTimerGuard { - clear_timeout: Option, - global: JsValue, - handle: Rc>>, -} - -#[cfg(target_arch = "wasm32")] -impl GlobalTimerGuard { - fn disarm(&mut self) { - self.handle.borrow_mut().take(); - } -} - -#[cfg(target_arch = "wasm32")] -impl Drop for GlobalTimerGuard { - fn drop(&mut self) { - if let (Some(clear_timeout), Some(handle)) = - (&self.clear_timeout, self.handle.borrow_mut().take()) - { - let _ = clear_timeout.call1(&self.global, &handle); - } - } -} - struct CancelGuard { id: u32, cancel: Arc, diff --git a/scripts/check-dependencies.js b/scripts/check-dependencies.js index daaa3bc9..09b5b23f 100644 --- a/scripts/check-dependencies.js +++ b/scripts/check-dependencies.js @@ -13,12 +13,9 @@ const trees = [ // modern upstream TUF and aligning the remaining remote config dependencies. const remoteConfigDuplicatePackages = new Set([ 'getrandom', - 'hashbrown', 'http', 'itoa', 'syn', - 'thiserror', - 'thiserror-impl', 'untrusted', ]) // libdd-remote-config exposes a single-client API but still compiles its Tokio From ece33d284c8adf2d5db5933d9d0174c25a7fc386 Mon Sep 17 00:00:00 2001 From: Roch Devost Date: Thu, 27 Aug 2026 19:38:28 -0400 Subject: [PATCH 3/5] fix(libdatadog): preserve async context --- crates/libdatadog/src/data_pipeline/mod.rs | 36 ++++++++++-- crates/libdatadog/src/remote_config.rs | 11 ++-- .../libdatadog/lib/agentless-transport.js | 57 +++++++++++++++++-- packages/libdatadog/lib/agentless.js | 16 ++++-- packages/libdatadog/lib/remote-config.js | 16 ++++-- packages/libdatadog/test/exporter.test.js | 46 +++++++++++++++ .../libdatadog/test/remote-config.test.js | 19 ++++++- 7 files changed, 173 insertions(+), 28 deletions(-) diff --git a/crates/libdatadog/src/data_pipeline/mod.rs b/crates/libdatadog/src/data_pipeline/mod.rs index 91bcced1..baa15f66 100644 --- a/crates/libdatadog/src/data_pipeline/mod.rs +++ b/crates/libdatadog/src/data_pipeline/mod.rs @@ -24,7 +24,7 @@ type RequestFunction = ThreadsafeFunction< false, true, >; -type SleepArgs = FnArgs<(u32, u32)>; +type SleepArgs = FnArgs<(u32, u32, u32)>; type SleepFunction = ThreadsafeFunction, SleepArgs, Status, false, true>; type CancelFunction = ThreadsafeFunction; @@ -53,6 +53,7 @@ pub struct AgentlessRequestHeader { #[napi(object)] pub struct AgentlessRequest { pub id: u32, + pub context_id: u32, pub url: String, pub method: String, pub headers: Vec, @@ -72,6 +73,7 @@ pub(crate) struct HostCapabilities { sleep: Arc, cancel_sleep: Arc, next_call_id: Arc, + context_id: Arc, } impl fmt::Debug for HostCapabilities { @@ -84,7 +86,7 @@ impl HostCapabilities { pub(crate) fn new( request: Function<'_, AgentlessRequest, Promise>, cancel_request: Function<'_, u32, ()>, - sleep: Function<'_, FnArgs<(u32, u32)>, Promise<()>>, + sleep: Function<'_, FnArgs<(u32, u32, u32)>, Promise<()>>, cancel_sleep: Function<'_, u32, ()>, ) -> Result { Ok(Self { @@ -113,12 +115,27 @@ impl HostCapabilities { .build()?, ), next_call_id: Arc::new(AtomicU32::new(1)), + context_id: Arc::new(AtomicU32::new(0)), }) } fn next_call_id(&self) -> u32 { self.next_call_id.fetch_add(1, Ordering::Relaxed) } + + pub(crate) fn with_context(&self, context_id: u32) -> Self { + let mut capabilities = self.clone(); + capabilities.context_id = Arc::new(AtomicU32::new(context_id)); + capabilities + } + + pub(crate) fn set_context(&self, context_id: u32) { + self.context_id.store(context_id, Ordering::Relaxed); + } + + fn context_id(&self) -> u32 { + self.context_id.load(Ordering::Relaxed) + } } impl HttpClientCapability for HostCapabilities { @@ -151,6 +168,7 @@ impl HttpClientCapability for HostCapabilities { .collect::, _>>()?; let request = AgentlessRequest { id, + context_id: self.context_id(), url: parts.uri.to_string(), method: parts.method.to_string(), headers, @@ -181,7 +199,8 @@ impl SleepCapability for HostCapabilities { 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 args = (id, milliseconds, self.context_id()).into(); + if let Ok(promise) = self.sleep.call_async(args).await { let _ = promise.await; } guard.disarm(); @@ -233,7 +252,7 @@ impl AgentlessExporter { options: AgentlessExporterOptions, request: Function<'_, AgentlessRequest, Promise>, cancel_request: Function<'_, u32, ()>, - sleep: Function<'_, FnArgs<(u32, u32)>, Promise<()>>, + sleep: Function<'_, FnArgs<(u32, u32, u32)>, Promise<()>>, cancel_sleep: Function<'_, u32, ()>, ) -> Result { let timeout = options @@ -270,11 +289,16 @@ impl AgentlessExporter { } #[napi] - pub fn send_v04<'env>(&self, env: &'env Env, payload: Buffer) -> Result> { + pub fn send_v04<'env>( + &self, + env: &'env Env, + payload: Buffer, + context_id: u32, + ) -> 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 capabilities = self.capabilities.with_context(context_id); let metadata = self.metadata.clone(); let config = self.config.clone(); let in_flight = self.in_flight.clone(); diff --git a/crates/libdatadog/src/remote_config.rs b/crates/libdatadog/src/remote_config.rs index 7d627ad3..ba6a3669 100644 --- a/crates/libdatadog/src/remote_config.rs +++ b/crates/libdatadog/src/remote_config.rs @@ -74,6 +74,7 @@ impl From for RemoteConfigChange { #[napi] pub struct RemoteConfigFetcher { client: Arc>>, + capabilities: HostCapabilities, pending: Arc>, } @@ -84,23 +85,25 @@ impl RemoteConfigFetcher { options: RemoteConfigFetcherOptions, request: Function<'_, AgentlessRequest, Promise>, cancel_request: Function<'_, u32, ()>, - sleep: Function<'_, FnArgs<(u32, u32)>, Promise<()>>, + sleep: Function<'_, FnArgs<(u32, u32, u32)>, Promise<()>>, cancel_sleep: Function<'_, u32, ()>, ) -> Result { let capabilities = HostCapabilities::new(request, cancel_request, sleep, cancel_sleep)?; - let client = - RemoteConfigClient::new(options.into(), capabilities).map_err(Error::from_reason)?; + let client = RemoteConfigClient::new(options.into(), capabilities.clone()) + .map_err(Error::from_reason)?; Ok(Self { client: Arc::new(AsyncMutex::new(client)), + capabilities, pending: Arc::new(Mutex::new(PendingUpdates::default())), }) } #[napi] - pub async fn fetch_changes(&self) -> Result> { + pub async fn fetch_changes(&self, context_id: u32) -> Result> { let updates = std::mem::take(&mut *lock(&self.pending)); let mut client = self.client.lock().await; + self.capabilities.set_context(context_id); client .fetch_changes(updates) .await diff --git a/packages/libdatadog/lib/agentless-transport.js b/packages/libdatadog/lib/agentless-transport.js index daef09b8..676e2cb6 100644 --- a/packages/libdatadog/lib/agentless-transport.js +++ b/packages/libdatadog/lib/agentless-transport.js @@ -1,10 +1,49 @@ 'use strict' +const { AsyncResource } = require('node:async_hooks') + function createHostTransport () { + const asyncResources = new Map() const requests = new Map() const timers = new Map() + let nextContextId = 1 + + function runWithAsyncResource (type, callback) { + const contextId = nextContextId + nextContextId = nextContextId === 4_294_967_295 ? 1 : nextContextId + 1 + const resource = new AsyncResource(type, { requireManualDestroy: true }) + asyncResources.set(contextId, resource) + + let operation + try { + operation = resource.runInAsyncScope(callback, undefined, contextId) + } catch (error) { + destroyAsyncResource(contextId, resource) + throw error + } + + return Promise.resolve(operation).finally(() => { + destroyAsyncResource(contextId, resource) + }) + } + + function runInAsyncScope (contextId, callback, ...args) { + const resource = asyncResources.get(contextId) + return resource + ? resource.runInAsyncScope(callback, undefined, ...args) + : callback(...args) + } + + function destroyAsyncResource (contextId, resource) { + asyncResources.delete(contextId) + resource.emitDestroy() + } - async function request ({ id, url, method, headers: headerList, body }) { + function request (args) { + return runInAsyncScope(args.contextId, startRequest, args) + } + + function startRequest ({ id, contextId, 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])) @@ -33,6 +72,7 @@ function createHostTransport () { }) requests.set(id, { + contextId, cancel: () => { const error = new Error('agentless request was cancelled') finish(reject, error) @@ -45,13 +85,18 @@ function createHostTransport () { } function cancelRequest (id) { - requests.get(id)?.cancel() + const request = requests.get(id) + if (request) runInAsyncScope(request.contextId, request.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) { + function sleep (id, milliseconds, contextId) { + return runInAsyncScope(contextId, startSleep, id, milliseconds, contextId) + } + + function startSleep (id, milliseconds, contextId) { return new Promise((resolve, reject) => { const timeout = setTimeout(() => { timers.delete(id) @@ -59,6 +104,7 @@ function createHostTransport () { }, milliseconds) timeout.unref?.() timers.set(id, { + contextId, cancel: () => { clearTimeout(timeout) timers.delete(id) @@ -69,10 +115,11 @@ function createHostTransport () { } function cancelSleep (id) { - timers.get(id)?.cancel() + const timer = timers.get(id) + if (timer) runInAsyncScope(timer.contextId, timer.cancel) } - return { request, cancelRequest, sleep, cancelSleep } + return { request, cancelRequest, sleep, cancelSleep, runWithAsyncResource } } module.exports = { createHostTransport } diff --git a/packages/libdatadog/lib/agentless.js b/packages/libdatadog/lib/agentless.js index a7912b8c..4365a13c 100644 --- a/packages/libdatadog/lib/agentless.js +++ b/packages/libdatadog/lib/agentless.js @@ -8,6 +8,7 @@ class AgentlessExporter { #binding #closed = false #inFlight = new Set() + #transport constructor (binding, options) { validateOptions(options) @@ -16,13 +17,13 @@ class AgentlessExporter { Object.entries({ ...options, runtimeId }) .filter(([, value]) => value !== null), ) - const transport = createHostTransport() + this.#transport = createHostTransport() this.#binding = new binding.AgentlessExporter( normalized, - transport.request, - transport.cancelRequest, - transport.sleep, - transport.cancelSleep, + this.#transport.request, + this.#transport.cancelRequest, + this.#transport.sleep, + this.#transport.cancelSleep, ) } @@ -33,7 +34,10 @@ class AgentlessExporter { let operation try { - operation = this.#binding.sendV04(payload) + operation = this.#transport.runWithAsyncResource( + 'libdatadog:AgentlessExporter.sendV04', + contextId => this.#binding.sendV04(payload, contextId), + ) } catch (error) { operation = Promise.reject(error) } diff --git a/packages/libdatadog/lib/remote-config.js b/packages/libdatadog/lib/remote-config.js index f07d95b7..ff779e4f 100644 --- a/packages/libdatadog/lib/remote-config.js +++ b/packages/libdatadog/lib/remote-config.js @@ -5,20 +5,24 @@ const { createHostTransport } = require('./agentless-transport') function remoteConfigFetcher (binding) { return class RemoteConfigFetcher { #binding + #transport constructor (options) { - const transport = createHostTransport() + this.#transport = createHostTransport() this.#binding = new binding.RemoteConfigFetcher( options, - transport.request, - transport.cancelRequest, - transport.sleep, - transport.cancelSleep, + this.#transport.request, + this.#transport.cancelRequest, + this.#transport.sleep, + this.#transport.cancelSleep, ) } fetchChanges () { - return this.#binding.fetchChanges() + return this.#transport.runWithAsyncResource( + 'libdatadog:RemoteConfigFetcher.fetchChanges', + contextId => this.#binding.fetchChanges(contextId), + ) } setConfigState (path, applyState, applyError) { diff --git a/packages/libdatadog/test/exporter.test.js b/packages/libdatadog/test/exporter.test.js index c13fcc89..84369bbd 100644 --- a/packages/libdatadog/test/exporter.test.js +++ b/packages/libdatadog/test/exporter.test.js @@ -1,7 +1,11 @@ 'use strict' +/* eslint-disable unicorn/prefer-event-target -- Node stream mocks use EventEmitter. */ + const assert = require('node:assert/strict') +const { AsyncLocalStorage } = require('node:async_hooks') const { spawnSync } = require('node:child_process') +const { EventEmitter } = require('node:events') const fs = require('node:fs') const http = require('node:http') const path = require('node:path') @@ -84,6 +88,48 @@ const backends = [ ] for (const backend of backends) { + test(`${backend.name} preserves async context in host callbacks`, { + skip: backend.skip, + }, async (context) => { + const storage = new AsyncLocalStorage() + const stores = [] + context.mock.method(http, 'request', (_url, _options, onResponse) => { + stores.push(storage.getStore()) + const request = new EventEmitter() + request.destroy = error => request.emit('error', error) + request.end = () => queueMicrotask(() => { + const response = new EventEmitter() + response.statusCode = 200 + onResponse(response) + response.emit('end') + }) + return request + }) + const pipeline = backend.load() + const exporter = pipeline.createAgentlessExporter({ + endpoint: 'http://example.test/api/v2/spans', + apiKey: 'test-api-key', + tracerVersion: '0.1.0', + languageVersion: process.version, + languageInterpreter: 'v8', + }) + const first = { operation: 'first' } + const second = { operation: 'second' } + + try { + await Promise.all([ + storage.run(first, () => exporter.sendV04(tracePayload())), + storage.run(second, () => exporter.sendV04(tracePayload())), + ]) + } finally { + await exporter.close() + } + + assert.strictEqual(stores.length, 2) + assert.ok(stores.includes(first)) + assert.ok(stores.includes(second)) + }) + test(`${backend.name} retries in Rust until the third attempt succeeds`, { skip: backend.skip, }, async () => { diff --git a/packages/libdatadog/test/remote-config.test.js b/packages/libdatadog/test/remote-config.test.js index 0e03694d..a1164c32 100644 --- a/packages/libdatadog/test/remote-config.test.js +++ b/packages/libdatadog/test/remote-config.test.js @@ -3,6 +3,7 @@ /* eslint-disable unicorn/prefer-event-target -- Node stream mocks use EventEmitter. */ const assert = require('node:assert/strict') +const { AsyncLocalStorage } = require('node:async_hooks') const { execFileSync } = require('node:child_process') const { EventEmitter } = require('node:events') const https = require('node:https') @@ -35,9 +36,10 @@ function fetcherOptions (overrides = {}) { } } -function mockHttps (context) { +function mockHttps (context, onRequest = () => {}) { const requests = [] context.mock.method(https, 'request', (url, options, onResponse) => { + onRequest() const request = new EventEmitter() request.destroy = error => request.emit('error', error) request.end = (body) => { @@ -93,6 +95,21 @@ test('WASM agentless remote config does not require WebCrypto', () => { }) for (const [name, { RemoteConfigFetcher }] of backends) { + test(`${name} remote config preserves async context in host callbacks`, async (context) => { + const storage = new AsyncLocalStorage() + const stores = [] + mockHttps(context, () => stores.push(storage.getStore())) + const fetcher = new RemoteConfigFetcher(fetcherOptions()) + const expected = { operation: 'remote-config' } + + await assert.rejects( + storage.run(expected, () => fetcher.fetchChanges()), + /missing config meta/, + ) + assert.ok(stores.length > 0) + assert.ok(stores.every(store => store === expected)) + }) + test(`${name} remote config sends directly to the backend`, async (context) => { const requests = mockHttps(context) const fetcher = new RemoteConfigFetcher(fetcherOptions()) From 47af6f2ddb18fd3553939959cf247767e824a6e2 Mon Sep 17 00:00:00 2001 From: Ruben Bridgewater Date: Fri, 28 Aug 2026 16:37:39 +0200 Subject: [PATCH 4/5] test(remote-config): cover change conversion Agentless responses verify against embedded TUF roots, so a local success fixture would require a test-only root override. --- .github/workflows/build.yml | 4 ++ crates/libdatadog-remote-config/src/lib.rs | 57 ++++++++++++++++++++++ 2 files changed, 61 insertions(+) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index de2fc042..ac3f1e7e 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -31,6 +31,10 @@ jobs: run: rustup target add wasm32-unknown-unknown - name: Validate artifact dependencies run: npm run check:dependencies + - name: Test remote config adapter + run: >- + cargo test --locked --package libdatadog-remote-config + --features agentless,regex-lite - run: npm ci --prefix packages/libdatadog - run: cargo install cargo-bloat --version 0.12.1 --locked - run: npm run build --prefix packages/libdatadog diff --git a/crates/libdatadog-remote-config/src/lib.rs b/crates/libdatadog-remote-config/src/lib.rs index 19c50ced..8188446f 100644 --- a/crates/libdatadog-remote-config/src/lib.rs +++ b/crates/libdatadog-remote-config/src/lib.rs @@ -242,3 +242,60 @@ fn to_record( fn contents(file: &Arc>>) -> Option { Some(String::from_utf8_lossy(file.contents().as_slice()).into_owned()) } + +#[cfg(test)] +mod tests { + use libdd_remote_config::fetch::FileStorage; + + use super::*; + + const CONFIG_PATH: &str = "datadog/2/ASM_FEATURES/asm-features-1/config"; + + fn stored_file(version: u64, contents: &[u8]) -> Arc>> { + let path = Arc::new(RemoteConfigPath::parse(CONFIG_PATH).unwrap()); + SimpleFileStorage::default() + .store(version, path, contents.to_vec()) + .unwrap() + } + + #[test] + fn converts_add_update_and_remove_changes() { + let added_file = stored_file(1, br#"{"asm":{"enabled":true}}"#); + let added = to_change_record(Change::Add(added_file)); + + assert_eq!(added.kind, "add"); + assert_eq!(added.path, CONFIG_PATH); + assert_eq!(added.product, "ASM_FEATURES"); + assert_eq!(added.config_id, "asm-features-1"); + assert_eq!(added.name, "config"); + assert_eq!(added.version, 1.0); + assert_eq!( + added.contents.as_deref(), + Some(r#"{"asm":{"enabled":true}}"#) + ); + + let updated_file = stored_file(2, br#"{"asm":{"enabled":false}}"#); + let updated = to_change_record(Change::Update(updated_file.clone(), Vec::new())); + + assert_eq!(updated.kind, "update"); + assert_eq!(updated.path, CONFIG_PATH); + assert_eq!(updated.product, "ASM_FEATURES"); + assert_eq!(updated.config_id, "asm-features-1"); + assert_eq!(updated.name, "config"); + assert_eq!(updated.version, 2.0); + assert_eq!( + updated.contents.as_deref(), + Some(r#"{"asm":{"enabled":false}}"#) + ); + + let removed = to_change_record(Change::Remove(updated_file)); + + assert_eq!(removed.kind, "remove"); + assert_eq!(removed.path, CONFIG_PATH); + assert_eq!(removed.product, "ASM_FEATURES"); + assert_eq!(removed.config_id, "asm-features-1"); + assert_eq!(removed.name, "config"); + assert_eq!(removed.version, 2.0); + assert_eq!(removed.contents, None); + } +} From 78f6cc452d43f7527effe0217beeca704ba1b3f9 Mon Sep 17 00:00:00 2001 From: Ruben Bridgewater Date: Fri, 28 Aug 2026 16:38:41 +0200 Subject: [PATCH 5/5] fix(dependencies): scope remote config duplicate exceptions Any remote-config dependency previously exempted allowlisted crate names, which hid unrelated duplicate versions from artifact validation. --- scripts/check-dependencies.js | 4 +++- test/dependencies.js | 26 ++++++++++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/scripts/check-dependencies.js b/scripts/check-dependencies.js index 09b5b23f..8f313eb1 100644 --- a/scripts/check-dependencies.js +++ b/scripts/check-dependencies.js @@ -67,7 +67,9 @@ function findDuplicateVersions (dependencies) { function isRemoteConfigDuplicate (dependencies, name) { return remoteConfigDuplicatePackages.has(name) - && dependencies.some(({ path }) => path.includes('libdd-remote-config')) + && dependencies.some(({ name: dependencyName, path }) => ( + dependencyName === name && path.includes('libdd-remote-config') + )) } function findForbiddenDependencies (dependencies, tree) { diff --git a/test/dependencies.js b/test/dependencies.js index aca135ee..715f3b1c 100644 --- a/test/dependencies.js +++ b/test/dependencies.js @@ -125,3 +125,29 @@ test('dependency validation allows known remote config duplicate crates', () => }) assert.deepStrictEqual(findDuplicateVersions(dependencies), []) }) + +test('dependency validation rejects known crates duplicated outside remote config', () => { + const dependencies = parseCargoTree([ + '0libdatadog v0.1.0', + '1first-dependency v1.0.0', + '2syn v2.0.119', + '1libdd-remote-config v4.0.0', + '2bytes v1.0.0', + '1second-dependency v1.0.0', + '2syn v3.0.4', + ].join('\n')) + + const expected = [{ + name: 'syn', + versions: ['2.0.119', '3.0.4'], + }] + assert.deepStrictEqual(findDuplicateVersions(dependencies), expected) + + dependencies.push({ + depth: 2, + name: 'syn', + path: ['libdatadog', 'libdd-remote-config', 'syn'], + version: '3.0.5', + }) + assert.deepStrictEqual(findDuplicateVersions(dependencies), []) +})