diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 6a267e9997..f5a5da7f6d 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -138,6 +138,16 @@ jobs: # shellcheck disable=SC2086 cargo build $PACKAGES --verbose fi + - name: "[${{ steps.rust-version.outputs.version}}] check minimal native features" + if: runner.os == 'Linux' + shell: bash + run: | + if [[ -z "$PACKAGES" ]] || echo "$PACKAGES" | \ + grep -Eq "libdd-(common|dogstatsd-client|http-client)"; then + cargo check -p libdd-dogstatsd-client --no-default-features + cargo check -p libdd-http-client --no-default-features \ + --features hyper-backend + fi - name: "[${{ steps.rust-version.outputs.version}}] cargo test (doc) and cargo nextest run" shell: bash # Run doc tests with cargo test and run tests with nextest and generate junit.xml diff --git a/Cargo.lock b/Cargo.lock index 93007635ea..be27cd26eb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2971,6 +2971,7 @@ dependencies = [ "libdd-capabilities", "libdd-capabilities-impl", "libdd-common", + "libdd-data-pipeline-core", "libdd-ddsketch", "libdd-dogstatsd-client", "libdd-log", @@ -2999,6 +3000,20 @@ dependencies = [ "zstd", ] +[[package]] +name = "libdd-data-pipeline-core" +version = "1.0.0" +dependencies = [ + "bytes", + "http 1.1.0", + "libdd-common", + "libdd-tinybytes", + "libdd-trace-utils", + "serde_json", + "thiserror 1.0.68", + "zstd", +] + [[package]] name = "libdd-data-pipeline-ffi" version = "41.0.0" diff --git a/Cargo.toml b/Cargo.toml index d1abbb4082..5ddc26dc02 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -53,6 +53,7 @@ members = [ "libdd-shared-runtime", "libdd-shared-runtime-ffi", "libdd-data-pipeline", + "libdd-data-pipeline-core", "libdd-data-pipeline-ffi", "libdd-ddsketch", "libdd-ddsketch-ffi", diff --git a/libdd-capabilities-impl/Cargo.toml b/libdd-capabilities-impl/Cargo.toml index 52b2024543..c0ea357e6a 100644 --- a/libdd-capabilities-impl/Cargo.toml +++ b/libdd-capabilities-impl/Cargo.toml @@ -20,7 +20,7 @@ anyhow.workspace = true bytes = "1" http = "1" libdd-capabilities = { path = "../libdd-capabilities", version = "3.0.0" } -libdd-common = { path = "../libdd-common", version = "5.2.0", default-features = false } +libdd-common = { path = "../libdd-common", version = "5.2.0", default-features = false, features = ["http-client"] } tokio = { workspace = true, features = ["fs", "time"] } [target.'cfg(not(target_arch = "wasm32"))'.dependencies] diff --git a/libdd-common/Cargo.toml b/libdd-common/Cargo.toml index b036c810d4..3d71cd9fd9 100644 --- a/libdd-common/Cargo.toml +++ b/libdd-common/Cargo.toml @@ -34,7 +34,7 @@ regex-lite = { version = "0.1", optional = true } # The default resolver can hold locks or other global state that can cause deadlocks # or corruption when the process forks (e.g., in PHP-FPM or other forking environments). # Use rustls-no-provider instead of rustls to avoid reqwest forcing aws-lc-rs as the crypto -# backend. We install the ring provider explicitly in connector/mod.rs instead. +# backend. We install the ring provider explicitly in connector/http_client.rs instead. reqwest = { version = "0.13.2", features = ["rustls-no-provider", "hickory-dns"], default-features = false, optional = true } criterion = { workspace = true, optional = true } rustls-native-certs = { version = ">=0.8.1, <0.8.3", optional = true } @@ -55,15 +55,15 @@ hyper-rustls = { version = "0.27.7", default-features = false, features = [ rustls-webpki = { version = ">=0.103.13", optional = true } [target.'cfg(not(target_arch = "wasm32"))'.dependencies] -hyper = { workspace = true, features = ["http1", "client"] } -hyper-util = { workspace = true, features = ["http1", "client", "client-legacy"] } -http-body = "1.0" -http-body-util = "0.1" -tower-service = "0.3" +hyper = { workspace = true, features = ["http1", "client"], optional = true } +hyper-util = { workspace = true, features = ["http1", "client", "client-legacy"], optional = true } +http-body = { version = "1.0", optional = true } +http-body-util = { version = "0.1", optional = true } +tower-service = { version = "0.3", optional = true } cc = "1.1.31" pin-project = "1" libc.workspace = true -tokio = { workspace = true, features = ["rt", "rt-multi-thread", "macros", "net", "io-util", "fs", "time"] } +tokio = { workspace = true, features = ["rt", "rt-multi-thread", "macros", "net", "io-util", "fs", "time"], optional = true } [target.'cfg(windows)'.dependencies.windows-sys] version = "0.52" @@ -92,8 +92,16 @@ tokio = { version = "1.23", features = ["rt", "macros", "time"] } [features] default = ["https"] +http-client = [ + "dep:hyper", + "dep:hyper-util", + "dep:http-body", + "dep:http-body-util", + "dep:tower-service", + "dep:tokio", +] # TLS plumbing without a crypto provider. Use `https` or `fips` to select one. -tls-core = ["tokio-rustls", "rustls", "hyper-rustls","rustls-native-certs", "rustls-platform-verifier"] +tls-core = ["http-client", "tokio-rustls", "rustls", "hyper-rustls","rustls-native-certs", "rustls-platform-verifier"] # Default HTTPS: ring as crypto provider https = ["tls-core", "rustls/ring", "hyper-rustls/ring"] use_webpki_roots = ["hyper-rustls/webpki-roots"] @@ -108,7 +116,7 @@ require-regex-full = [] # FIPS mode uses the FIPS-compliant cryptographic provider (Unix only) fips = ["tls-core", "hyper-rustls/fips"] # Enable reqwest client builder support with file dump debugging -reqwest = ["dep:reqwest", "test-utils"] +reqwest = ["http-client", "dep:reqwest", "test-utils"] # Enable test utilities for use in other crates test-utils = ["dep:httparse", "dep:rand", "dep:mime", "dep:multer"] # Enable benchmark utilities (ReportingAllocator, Criterion allocation measurement) @@ -120,8 +128,8 @@ bench-utils = ["dep:criterion"] # provider default needs to be set by the caller in fips mode. For now, we want # to make sure that the coverage tests use the non-fips version of the crypto # provider initialization logic, so we added a coverage cfg check on the -# function in src/connector/mod.rs. The coverage config is actually not used in -# normal environments, so we need to let the rust linter know that it is in +# function in src/connector/http_client.rs. The coverage config is actually not +# used in normal environments, so we need to let the rust linter know that it is in # fact a real thing, though one that shows up only in some situations. unexpected_cfgs = { level = "warn", check-cfg = ['cfg(coverage)'] } diff --git a/libdd-common/src/connector/http_client.rs b/libdd-common/src/connector/http_client.rs new file mode 100644 index 0000000000..282bd9605e --- /dev/null +++ b/libdd-common/src/connector/http_client.rs @@ -0,0 +1,251 @@ +// Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use super::conn_stream::{ConnStream, ConnStreamError}; +use super::errors; +use core::future::Future; +use core::pin::Pin; +use core::task::{Context, Poll}; +use futures::future::BoxFuture; +use futures::{future, FutureExt}; +use hyper_util::client::legacy::connect; +use std::sync::LazyLock; + +#[derive(Clone)] +pub enum Connector { + Http(connect::HttpConnector), + #[cfg(feature = "tls-core")] + Https(hyper_rustls::HttpsConnector), +} + +static DEFAULT_CONNECTOR: LazyLock = LazyLock::new(Connector::new); + +impl Default for Connector { + fn default() -> Self { + DEFAULT_CONNECTOR.clone() + } +} + +impl Connector { + /// Make sure this function is not called frequently. Fetching the root certificates is an + /// expensive operation. Access the globally cached connector via Connector::default(). + fn new() -> Self { + #[cfg(feature = "tls-core")] + { + #[cfg(feature = "use_webpki_roots")] + let https_connector_fn = https::build_https_connector_with_webpki_roots; + #[cfg(not(feature = "use_webpki_roots"))] + let https_connector_fn = https::build_https_connector; + + match https_connector_fn() { + Ok(connector) => Connector::Https(connector), + Err(_) => Connector::Http(connect::HttpConnector::new()), + } + } + #[cfg(not(feature = "tls-core"))] + { + Connector::Http(connect::HttpConnector::new()) + } + } + + fn build_conn_stream( + &mut self, + uri: hyper::Uri, + require_tls: bool, + ) -> BoxFuture<'static, Result> { + match self { + Self::Http(c) => { + if require_tls { + future::err::( + errors::Error::CannotEstablishTlsConnection.into(), + ) + .boxed() + } else { + ConnStream::from_http_connector_with_uri(c, uri).boxed() + } + } + #[cfg(feature = "tls-core")] + Self::Https(c) => { + ConnStream::from_https_connector_with_uri(c, uri, require_tls).boxed() + } + } + } +} + +#[cfg(feature = "tls-core")] +mod https { + #[cfg(feature = "use_webpki_roots")] + use hyper_rustls::ConfigBuilderExt; + + use rustls::ClientConfig; + + /// Ensures the rustls default CryptoProvider is installed (ring for non-FIPS). + /// In FIPS mode, the caller must install the FIPS provider before any TLS use. + #[cfg(feature = "https")] + fn ensure_crypto_provider_initialized() { + use std::sync::Once; + + static INIT_CRYPTO_PROVIDER: Once = Once::new(); + + INIT_CRYPTO_PROVIDER.call_once(|| { + let _ = rustls::crypto::ring::default_provider().install_default(); + }); + } + + /// In FIPS mode, the caller must install the FIPS-compliant crypto provider + /// (e.g., aws-lc-rs FIPS) before any TLS connections are established. + #[cfg(not(feature = "https"))] + fn ensure_crypto_provider_initialized() {} + + #[cfg(feature = "use_webpki_roots")] + pub(super) fn build_https_connector_with_webpki_roots() -> anyhow::Result< + hyper_rustls::HttpsConnector, + > { + ensure_crypto_provider_initialized(); + + let client_config = ClientConfig::builder() + .with_webpki_roots() + .with_no_client_auth(); + Ok(hyper_rustls::HttpsConnectorBuilder::new() + .with_tls_config(client_config) + .https_or_http() + .enable_http1() + .build()) + } + + #[cfg(not(feature = "use_webpki_roots"))] + /// Returns a default connector that uses the system trust roots. + /// `SSL_CERT_FILE` and `SSL_CERT_DIR` variable are only supported on linux, see + /// `rustls_platform_verifier` doc for details. + pub(super) fn build_https_connector() -> anyhow::Result< + hyper_rustls::HttpsConnector, + > { + use rustls_platform_verifier::BuilderVerifierExt; + + ensure_crypto_provider_initialized(); + + let client_config = ClientConfig::builder() + .with_platform_verifier()? + .with_no_client_auth(); + + Ok(hyper_rustls::HttpsConnectorBuilder::new() + .with_tls_config(client_config) + .https_or_http() + .enable_http1() + .build()) + } +} + +impl tower_service::Service for Connector { + type Response = ConnStream; + type Error = ConnStreamError; + + // This lint gets lifted in this place in a newer version, see: + // https://github.com/rust-lang/rust-clippy/pull/8030 + #[allow(clippy::type_complexity)] + type Future = Pin> + Send>>; + + fn call(&mut self, uri: hyper::Uri) -> Self::Future { + match uri.scheme_str() { + Some("unix") => ConnStream::from_uds_uri(uri).boxed(), + Some("windows") => ConnStream::from_named_pipe_uri(uri).boxed(), + Some("https") => self.build_conn_stream(uri, true), + _ => self.build_conn_stream(uri, false), + } + } + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + match self { + Connector::Http(c) => c.poll_ready(cx).map_err(|e| e.into()), + #[cfg(feature = "tls-core")] + Connector::Https(c) => c.poll_ready(cx), + } + } +} + +#[cfg(test)] +mod tests { + use crate::http_common; + #[cfg(any(feature = "use_webpki_roots", target_os = "linux"))] + use {super::*, std::env}; + #[cfg(feature = "tls-core")] + use {crate::http_common::Body, hyper::Request}; + + #[test] + #[cfg_attr(miri, ignore)] + #[cfg(not(feature = "use_webpki_roots"))] + /// Verify that the Connector type implements the correct bound Connect + Clone + /// to be able to use the hyper::Client + fn test_hyper_client_from_connector() { + let _ = http_common::new_default_client(); + } + + #[test] + #[cfg_attr(miri, ignore)] + #[cfg(feature = "use_webpki_roots")] + fn test_hyper_client_from_connector_with_webpki_roots() { + let _ = http_common::new_default_client(); + } + + #[test] + #[cfg_attr(miri, ignore)] + #[cfg(not(feature = "use_webpki_roots"))] + // Only Linux eagerly loads roots at connector construction; macOS/Windows verify lazily + // during the TLS handshake, so SSL_CERT_FILE/SSL_CERT_DIR cannot be exercised there. + #[cfg(target_os = "linux")] + /// Verify that Connector falls back to Http when native root certificates + /// are not available and webpki roots are not enabled. + fn test_missing_root_certificates_only_allow_http_connections() { + const ENV_SSL_CERT_FILE: &str = "SSL_CERT_FILE"; + const ENV_SSL_CERT_DIR: &str = "SSL_CERT_DIR"; + let old_value = env::var(ENV_SSL_CERT_FILE).unwrap_or_default(); + let old_dir_value = env::var(ENV_SSL_CERT_DIR).unwrap_or_default(); + + env::set_var(ENV_SSL_CERT_FILE, "this/folder/does/not/exist"); + env::set_var(ENV_SSL_CERT_DIR, "this/folder/does/not/exist"); + let connector = Connector::new(); + + assert!(matches!(connector, Connector::Http(_))); + + env::set_var(ENV_SSL_CERT_FILE, old_value); + env::set_var(ENV_SSL_CERT_DIR, old_dir_value); + } + + #[test] + #[cfg_attr(miri, ignore)] + #[cfg(feature = "use_webpki_roots")] + #[cfg(feature = "tls-core")] + /// Verify that Connector builds an Https connector using webpki certificates + /// even when native root certificates are not available. + fn test_missing_root_certificates_use_webpki_certificates() { + const ENV_SSL_CERT_FILE: &str = "SSL_CERT_FILE"; + let old_value = env::var(ENV_SSL_CERT_FILE).unwrap_or_default(); + + env::set_var(ENV_SSL_CERT_FILE, "this/folder/does/not/exist"); + let connector = Connector::new(); + assert!(matches!(connector, Connector::Https(_))); + + env::set_var(ENV_SSL_CERT_FILE, old_value); + } + + #[tokio::test] + #[cfg_attr(miri, ignore)] + #[cfg(feature = "tls-core")] + /// Verify that a HTTPS GET request succeeds using + /// the default Connector (native platform TLS verifier or webpki roots). + async fn test_https_request_succeeds() { + let client = http_common::new_default_client(); + let request = Request::get("https://www.datadoghq.com") + .body(Body::empty()) + .expect("failed to build request"); + let response = client + .request(request) + .await + .expect("HTTPS request to datadoghq.com failed"); + let status = response.status(); + assert!( + status.is_success() || status.is_redirection(), + "unexpected status code: {status}" + ); + } +} diff --git a/libdd-common/src/connector/mod.rs b/libdd-common/src/connector/mod.rs index 8647f8e71d..bdb64a5b1f 100644 --- a/libdd-common/src/connector/mod.rs +++ b/libdd-common/src/connector/mod.rs @@ -1,261 +1,14 @@ // Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -use futures::future::BoxFuture; -use futures::{future, FutureExt}; -use hyper_util::client::legacy::connect; - -use core::future::Future; -use core::pin::Pin; -use core::task::{Context, Poll}; -use std::sync::LazyLock; - +pub mod errors; +pub mod named_pipe; #[cfg(unix)] pub mod uds; -pub mod named_pipe; - -pub mod errors; - +#[cfg(feature = "http-client")] mod conn_stream; -use conn_stream::{ConnStream, ConnStreamError}; - -#[derive(Clone)] -pub enum Connector { - Http(connect::HttpConnector), - #[cfg(feature = "tls-core")] - Https(hyper_rustls::HttpsConnector), -} - -static DEFAULT_CONNECTOR: LazyLock = LazyLock::new(Connector::new); - -impl Default for Connector { - fn default() -> Self { - DEFAULT_CONNECTOR.clone() - } -} - -impl Connector { - /// Make sure this function is not called frequently. Fetching the root certificates is an - /// expensive operation. Access the globally cached connector via Connector::default(). - fn new() -> Self { - #[cfg(feature = "tls-core")] - { - #[cfg(feature = "use_webpki_roots")] - let https_connector_fn = https::build_https_connector_with_webpki_roots; - #[cfg(not(feature = "use_webpki_roots"))] - let https_connector_fn = https::build_https_connector; - - match https_connector_fn() { - Ok(connector) => Connector::Https(connector), - Err(_) => Connector::Http(connect::HttpConnector::new()), - } - } - #[cfg(not(feature = "tls-core"))] - { - Connector::Http(connect::HttpConnector::new()) - } - } - - fn build_conn_stream( - &mut self, - uri: hyper::Uri, - require_tls: bool, - ) -> BoxFuture<'static, Result> { - match self { - Self::Http(c) => { - if require_tls { - future::err::( - errors::Error::CannotEstablishTlsConnection.into(), - ) - .boxed() - } else { - ConnStream::from_http_connector_with_uri(c, uri).boxed() - } - } - #[cfg(feature = "tls-core")] - Self::Https(c) => { - ConnStream::from_https_connector_with_uri(c, uri, require_tls).boxed() - } - } - } -} - -#[cfg(feature = "tls-core")] -mod https { - #[cfg(feature = "use_webpki_roots")] - use hyper_rustls::ConfigBuilderExt; - - use rustls::ClientConfig; - - /// Ensures the rustls default CryptoProvider is installed (ring for non-FIPS). - /// In FIPS mode, the caller must install the FIPS provider before any TLS use. - #[cfg(feature = "https")] - fn ensure_crypto_provider_initialized() { - use std::sync::Once; - - static INIT_CRYPTO_PROVIDER: Once = Once::new(); - - INIT_CRYPTO_PROVIDER.call_once(|| { - let _ = rustls::crypto::ring::default_provider().install_default(); - }); - } - - /// In FIPS mode, the caller must install the FIPS-compliant crypto provider - /// (e.g., aws-lc-rs FIPS) before any TLS connections are established. - #[cfg(not(feature = "https"))] - fn ensure_crypto_provider_initialized() {} - - #[cfg(feature = "use_webpki_roots")] - pub(super) fn build_https_connector_with_webpki_roots() -> anyhow::Result< - hyper_rustls::HttpsConnector, - > { - ensure_crypto_provider_initialized(); // One-time initialization of a crypto provider if needed - - let client_config = ClientConfig::builder() - .with_webpki_roots() - .with_no_client_auth(); - Ok(hyper_rustls::HttpsConnectorBuilder::new() - .with_tls_config(client_config) - .https_or_http() - .enable_http1() - .build()) - } - - #[cfg(not(feature = "use_webpki_roots"))] - /// Returns a default connector that uses the system trust roots. - /// `SSL_CERT_FILE` and `SSL_CERT_DIR` variable are only supported on linux, see - /// `rustls_platform_verifier` doc for details. - pub(super) fn build_https_connector() -> anyhow::Result< - hyper_rustls::HttpsConnector, - > { - use rustls_platform_verifier::BuilderVerifierExt; - - ensure_crypto_provider_initialized(); // One-time initialization of a crypto provider if needed - - let client_config = ClientConfig::builder() - .with_platform_verifier()? - .with_no_client_auth(); - - Ok(hyper_rustls::HttpsConnectorBuilder::new() - .with_tls_config(client_config) - .https_or_http() - .enable_http1() - .build()) - } -} - -impl tower_service::Service for Connector { - type Response = ConnStream; - type Error = ConnStreamError; - - // This lint gets lifted in this place in a newer version, see: - // https://github.com/rust-lang/rust-clippy/pull/8030 - #[allow(clippy::type_complexity)] - type Future = Pin> + Send>>; - - fn call(&mut self, uri: hyper::Uri) -> Self::Future { - match uri.scheme_str() { - Some("unix") => conn_stream::ConnStream::from_uds_uri(uri).boxed(), - Some("windows") => conn_stream::ConnStream::from_named_pipe_uri(uri).boxed(), - Some("https") => self.build_conn_stream(uri, true), - _ => self.build_conn_stream(uri, false), - } - } - - fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { - match self { - Connector::Http(c) => c.poll_ready(cx).map_err(|e| e.into()), - #[cfg(feature = "tls-core")] - Connector::Https(c) => c.poll_ready(cx), - } - } -} - -#[cfg(test)] -mod tests { - use crate::http_common; - #[cfg(any(feature = "use_webpki_roots", target_os = "linux"))] - use {super::*, std::env}; - #[cfg(feature = "tls-core")] - use {crate::http_common::Body, hyper::Request}; - - #[test] - #[cfg_attr(miri, ignore)] - #[cfg(not(feature = "use_webpki_roots"))] - /// Verify that the Connector type implements the correct bound Connect + Clone - /// to be able to use the hyper::Client - fn test_hyper_client_from_connector() { - let _ = http_common::new_default_client(); - } - - #[test] - #[cfg_attr(miri, ignore)] - #[cfg(feature = "use_webpki_roots")] - fn test_hyper_client_from_connector_with_webpki_roots() { - let _ = http_common::new_default_client(); - } - - #[test] - #[cfg_attr(miri, ignore)] - #[cfg(not(feature = "use_webpki_roots"))] - // Only Linux eagerly loads roots at connector construction; macOS/Windows verify lazily - // during the TLS handshake, so SSL_CERT_FILE/SSL_CERT_DIR cannot be exercised there. - #[cfg(target_os = "linux")] - /// Verify that Connector falls back to Http when native root certificates - /// are not available and webpki roots are not enabled. - fn test_missing_root_certificates_only_allow_http_connections() { - const ENV_SSL_CERT_FILE: &str = "SSL_CERT_FILE"; - const ENV_SSL_CERT_DIR: &str = "SSL_CERT_DIR"; - let old_value = env::var(ENV_SSL_CERT_FILE).unwrap_or_default(); - let old_dir_value = env::var(ENV_SSL_CERT_DIR).unwrap_or_default(); - - env::set_var(ENV_SSL_CERT_FILE, "this/folder/does/not/exist"); - env::set_var(ENV_SSL_CERT_DIR, "this/folder/does/not/exist"); - let connector = Connector::new(); - - assert!(matches!(connector, Connector::Http(_))); - - env::set_var(ENV_SSL_CERT_FILE, old_value); - env::set_var(ENV_SSL_CERT_DIR, old_dir_value); - } - - #[test] - #[cfg_attr(miri, ignore)] - #[cfg(feature = "use_webpki_roots")] - #[cfg(feature = "tls-core")] - /// Verify that Connector builds an Https connector using webpki certificates - /// even when native root certificates are not available. - fn test_missing_root_certificates_use_webpki_certificates() { - const ENV_SSL_CERT_FILE: &str = "SSL_CERT_FILE"; - let old_value = env::var(ENV_SSL_CERT_FILE).unwrap_or_default(); - - env::set_var(ENV_SSL_CERT_FILE, "this/folder/does/not/exist"); - let connector = Connector::new(); - assert!(matches!(connector, Connector::Https(_))); - - env::set_var(ENV_SSL_CERT_FILE, old_value); - } - - #[tokio::test] - #[cfg_attr(miri, ignore)] - #[cfg(feature = "tls-core")] - /// Verify that a HTTPS GET request succeeds using - /// the default Connector (native platform TLS verifier or webpki roots). - async fn test_https_request_succeeds() { - let client = http_common::new_default_client(); - let request = Request::get("https://www.datadoghq.com") - .body(Body::empty()) - .expect("failed to build request"); - let response = client - .request(request) - .await - .expect("HTTPS request to datadoghq.com failed"); - let status = response.status(); - // Accept any successful (2xx) or redirect (3xx) response. - assert!( - status.is_success() || status.is_redirection(), - "unexpected status code: {status}" - ); - } -} +#[cfg(feature = "http-client")] +mod http_client; +#[cfg(feature = "http-client")] +pub use http_client::Connector; diff --git a/libdd-common/src/connector/named_pipe.rs b/libdd-common/src/connector/named_pipe.rs index 6ee29190e6..c5c61a0024 100644 --- a/libdd-common/src/connector/named_pipe.rs +++ b/libdd-common/src/connector/named_pipe.rs @@ -27,17 +27,17 @@ pub const ANONYMOUS_IMPERSONATION_QOS: u32 = 0; // SECURITY_ANONYMOUS /// /// Build a URI from a Path representing a named pipe /// `path` - named pipe path. ex: \\.\pipe\pipename -pub fn named_pipe_path_to_uri(path: &Path) -> Result { +pub fn named_pipe_path_to_uri(path: &Path) -> Result { #[allow(clippy::unwrap_used)] let path = hex::encode(path.as_os_str().to_str().unwrap()); - hyper::Uri::builder() + http::Uri::builder() .scheme("windows") .authority(path) .path_and_query("/") .build() } -pub fn named_pipe_path_from_uri(uri: &hyper::Uri) -> anyhow::Result { +pub fn named_pipe_path_from_uri(uri: &http::Uri) -> anyhow::Result { if uri.scheme_str() != Some("windows") { return Err(super::errors::Error::InvalidUrl.into()); } diff --git a/libdd-common/src/connector/uds.rs b/libdd-common/src/connector/uds.rs index 42d5a4fbb1..52ca1b4453 100644 --- a/libdd-common/src/connector/uds.rs +++ b/libdd-common/src/connector/uds.rs @@ -7,16 +7,16 @@ use std::path::{Path, PathBuf}; /// Creates a new Uri, with the `unix` scheme, and the path to the socket /// encoded as a hex string, to prevent special characters in the url authority -pub fn socket_path_to_uri(path: &Path) -> Result { +pub fn socket_path_to_uri(path: &Path) -> Result { let path = hex::encode(path.as_os_str().as_bytes()); - hyper::Uri::builder() + http::Uri::builder() .scheme("unix") .authority(path) .path_and_query("/") .build() } -pub fn socket_path_from_uri(uri: &hyper::Uri) -> anyhow::Result { +pub fn socket_path_from_uri(uri: &http::Uri) -> anyhow::Result { if uri.scheme_str() != Some("unix") { return Err(super::errors::Error::InvalidUrl.into()); } diff --git a/libdd-common/src/lib.rs b/libdd-common/src/lib.rs index 2db7ced8ab..de15227ad2 100644 --- a/libdd-common/src/lib.rs +++ b/libdd-common/src/lib.rs @@ -23,6 +23,7 @@ pub mod cc_utils; #[cfg(not(target_arch = "wasm32"))] pub mod connector; #[cfg(feature = "reqwest")] +#[cfg(feature = "http-client")] pub mod dump_server; pub mod entity_id; pub mod machine_id; @@ -33,6 +34,7 @@ pub mod cstr; pub mod bench_utils; pub mod config; pub mod error; +#[cfg(feature = "http-client")] pub mod http_common; pub mod multipart; #[cfg(not(target_arch = "wasm32"))] @@ -199,17 +201,17 @@ pub mod header { HeaderName::from_static("x-datadog-test-session-token"); } -#[cfg(not(target_arch = "wasm32"))] +#[cfg(all(not(target_arch = "wasm32"), feature = "http-client"))] pub type HttpClient = http_common::GenericHttpClient; -#[cfg(not(target_arch = "wasm32"))] +#[cfg(all(not(target_arch = "wasm32"), feature = "http-client"))] pub type HttpResponse = http_common::HttpResponse; pub type HttpRequestBuilder = http::request::Builder; -#[cfg(not(target_arch = "wasm32"))] +#[cfg(all(not(target_arch = "wasm32"), feature = "http-client"))] pub trait Connect: hyper_util::client::legacy::connect::Connect + Clone + Send + Sync + 'static { } -#[cfg(not(target_arch = "wasm32"))] +#[cfg(all(not(target_arch = "wasm32"), feature = "http-client"))] impl Connect for C { diff --git a/libdd-data-pipeline-core/Cargo.toml b/libdd-data-pipeline-core/Cargo.toml new file mode 100644 index 0000000000..908394548f --- /dev/null +++ b/libdd-data-pipeline-core/Cargo.toml @@ -0,0 +1,26 @@ +[package] +name = "libdd-data-pipeline-core" +version = "1.0.0" +description = "Runtime-independent request preparation for Datadog APM data pipelines." +homepage = "https://github.com/DataDog/libdatadog/tree/main/libdd-data-pipeline-core" +repository = "https://github.com/DataDog/libdatadog/tree/main/libdd-data-pipeline-core" +rust-version.workspace = true +edition.workspace = true +license.workspace = true + +[dependencies] +bytes = "1.11.1" +http = "1" +serde_json.workspace = true +thiserror = "1.0" +libdd-common = { version = "5.2.0", path = "../libdd-common", default-features = false } +libdd-trace-utils = { version = "10.1.0", path = "../libdd-trace-utils", default-features = false } + +[dev-dependencies] +libdd-tinybytes = { version = "1.1.2", path = "../libdd-tinybytes", features = ["bytes_string"] } +zstd = { version = "0.13", default-features = false } + +[features] +default = [] +compression = ["libdd-trace-utils/compression"] +regex-lite = ["libdd-common/regex-lite"] diff --git a/libdd-data-pipeline-core/src/agentless.rs b/libdd-data-pipeline-core/src/agentless.rs new file mode 100644 index 0000000000..9e379ebdfc --- /dev/null +++ b/libdd-data-pipeline-core/src/agentless.rs @@ -0,0 +1,312 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +use std::{fmt, time::Duration}; + +use http::{HeaderMap, StatusCode}; +use libdd_common::Endpoint; +use libdd_trace_utils::send_with_retry::{ + CompressionStrategy, PreparedRequest, RetryBackoffType, RetryStrategy, +}; +use libdd_trace_utils::span::{trace_utils::compute_top_level_span, TraceData}; +use libdd_trace_utils::tracer_metadata::TracerMetadata; +use thiserror::Error; + +const AGENTLESS_MAX_RETRIES: u32 = 2; +const AGENTLESS_RETRY_DELAY_MS: u64 = 1000; + +/// Default timeout for one agentless request attempt. +pub const DEFAULT_AGENTLESS_TIMEOUT: Duration = Duration::from_secs(15); + +/// Agentless trace request configuration. +#[derive(Clone)] +pub struct AgentlessTraceConfig { + /// Full URL to POST traces to. + pub endpoint_url: String, + /// Datadog API key used for the `dd-api-key` header. + pub api_key: String, + /// Timeout for one request attempt. + pub timeout: Duration, +} + +impl fmt::Debug for AgentlessTraceConfig { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("AgentlessTraceConfig") + .field("endpoint_url", &self.endpoint_url) + .field("api_key", &"") + .field("timeout", &self.timeout) + .finish() + } +} + +/// An error encountered while preparing an agentless request. +#[derive(Debug, Error)] +pub enum PrepareAgentlessError { + /// The v0.4 MessagePack input could not be decoded. + #[error("failed to decode v0.4 traces: {0}")] + Deserialization(libdd_trace_utils::msgpack_decoder::decode::error::DecodeError), + /// The decoded traces could not be serialized as agentless JSON. + #[error("failed to encode agentless JSON: {0}")] + Serialization(#[source] serde_json::Error), + /// The configured endpoint was not a valid URI. + #[error("invalid agentless endpoint URL: {0}")] + InvalidEndpoint(String), + /// The API key cannot be represented as an HTTP header. + #[error("invalid Datadog API key header value")] + InvalidApiKey, +} + +/// A fully encoded agentless request for transport by the host runtime. +#[derive(Clone, Debug)] +pub struct PreparedAgentlessRequest { + request: PreparedRequest, + retry_strategy: RetryStrategy, +} + +impl PreparedAgentlessRequest { + /// Returns the reusable request plan. + pub fn request_plan(&self) -> &PreparedRequest { + &self.request + } + + /// Returns the retry strategy associated with the request. + pub fn retry_strategy(&self) -> &RetryStrategy { + &self.retry_strategy + } + + /// Builds one HTTP request attempt. + pub fn request(&self) -> Result, http::Error> { + self.request.request() + } + + /// Returns the timeout for one HTTP request attempt. + pub fn timeout(&self) -> Duration { + self.request.timeout() + } + + /// Returns the maximum number of retries after the initial attempt. + pub fn max_retries(&self) -> u32 { + self.retry_strategy.max_retries() + } + + /// Returns whether an HTTP response is retryable. + pub fn is_retryable_status(status: StatusCode) -> bool { + RetryStrategy::is_retryable_status(status) + } + + /// Returns the delay before retrying after the one-indexed `attempt`. + pub fn retry_delay(&self, attempt: u32) -> Option { + self.retry_strategy.retry_delay(attempt) + } +} + +/// Converts v0.4 MessagePack traces into a host-transported request. +pub fn prepare_agentless_v04_request( + data: &[u8], + metadata: &TracerMetadata, + config: &AgentlessTraceConfig, +) -> Result { + let (traces, _) = libdd_trace_utils::msgpack_decoder::v04::from_slice(data) + .map_err(PrepareAgentlessError::Deserialization)?; + prepare_agentless_traces_request(traces, metadata, config) +} + +/// Prepares already-decoded v0.4 traces for host transport. +pub fn prepare_agentless_traces_request( + mut traces: Vec>>, + metadata: &TracerMetadata, + config: &AgentlessTraceConfig, +) -> Result { + if !metadata.client_computed_top_level { + for chunk in &mut traces { + compute_top_level_span(chunk); + } + } + let trace_count = traces.len(); + let json_body = libdd_trace_utils::agentless_encoder::encode_payload(&traces, metadata) + .map_err(PrepareAgentlessError::Serialization)?; + let headers = build_agentless_headers(metadata, trace_count); + prepare_agentless_json_request(config, headers, json_body) +} + +/// Prepares an encoded agentless JSON request for host transport. +/// +/// The configured API key replaces any `dd-api-key` value in `headers`. +pub fn prepare_agentless_json_request( + config: &AgentlessTraceConfig, + mut headers: HeaderMap, + json_body: Vec, +) -> Result { + let api_key = http::HeaderValue::from_str(&config.api_key) + .map_err(|_| PrepareAgentlessError::InvalidApiKey)?; + headers.insert(http::HeaderName::from_static("dd-api-key"), api_key); + + let url = libdd_common::parse_uri(&config.endpoint_url) + .map_err(|error| PrepareAgentlessError::InvalidEndpoint(error.to_string()))?; + let target = Endpoint { + url, + timeout_ms: u64::try_from(config.timeout.as_millis()).unwrap_or(u64::MAX), + ..Endpoint::default() + }; + let retry_strategy = RetryStrategy::new( + AGENTLESS_MAX_RETRIES, + AGENTLESS_RETRY_DELAY_MS, + RetryBackoffType::Exponential, + None, + ); + + #[cfg(feature = "compression")] + let compression_strategy = CompressionStrategy::Zstd { level: 1 }; + #[cfg(not(feature = "compression"))] + let compression_strategy = CompressionStrategy::None; + + Ok(PreparedAgentlessRequest { + request: PreparedRequest::new(target, json_body, headers, compression_strategy), + retry_strategy, + }) +} + +fn build_agentless_headers(metadata: &TracerMetadata, trace_count: usize) -> HeaderMap { + let mut headers: HeaderMap = metadata.into(); + headers.insert( + http::header::CONTENT_TYPE, + libdd_common::header::APPLICATION_JSON, + ); + headers.insert( + http::HeaderName::from_static("x-datadog-trace-count"), + http::HeaderValue::from(trace_count), + ); + for (name, value) in libdd_common::entity_id::get_entity_headers() { + if let (Ok(name), Ok(value)) = ( + http::HeaderName::from_bytes(name.as_bytes()), + http::HeaderValue::from_str(value), + ) { + headers.insert(name, value); + } + } + headers +} + +#[cfg(test)] +mod tests { + use super::*; + use libdd_tinybytes::BytesString; + use libdd_trace_utils::msgpack_encoder; + use libdd_trace_utils::span::v04::SpanBytes; + + fn metadata() -> TracerMetadata { + TracerMetadata { + hostname: "host-1".to_string(), + service: "service-1".to_string(), + tracer_version: "1.2.3".to_string(), + language: "nodejs".to_string(), + language_version: "v24".to_string(), + language_interpreter: "v8".to_string(), + ..Default::default() + } + } + + fn config() -> AgentlessTraceConfig { + AgentlessTraceConfig { + endpoint_url: "https://example.test/v1/input".to_string(), + api_key: "test-api-key".to_string(), + timeout: Duration::from_millis(1_234), + } + } + + fn v04_traces() -> Vec> { + vec![vec![SpanBytes { + name: BytesString::from_static("operation"), + service: BytesString::from_static("service-1"), + resource: BytesString::from_static("resource-1"), + trace_id: 1, + span_id: 2, + start: 1, + duration: 2, + ..Default::default() + }]] + } + + fn v04_payload() -> Vec { + msgpack_encoder::v04::to_vec_from_v04(&v04_traces()) + } + + fn request_body(prepared: &PreparedAgentlessRequest) -> Vec { + let request = prepared.request().unwrap(); + #[cfg(feature = "compression")] + let body = zstd::decode_all(request.body().as_ref()).unwrap(); + #[cfg(not(feature = "compression"))] + let body = request.body().to_vec(); + body + } + + #[test] + fn prepares_v04_request_without_runtime() { + let prepared = + prepare_agentless_v04_request(&v04_payload(), &metadata(), &config()).unwrap(); + let request = prepared.request().unwrap(); + + assert_eq!(request.uri(), "https://example.test/v1/input"); + assert_eq!(request.headers()["dd-api-key"], "test-api-key"); + assert_eq!(request.headers()["x-datadog-trace-count"], "1"); + assert_eq!(prepared.timeout(), Duration::from_millis(1_234)); + assert_eq!(prepared.retry_delay(1), Some(Duration::from_secs(1))); + assert_eq!(prepared.retry_delay(2), Some(Duration::from_secs(2))); + assert_eq!(prepared.retry_delay(3), None); + + assert!(String::from_utf8(request_body(&prepared)) + .unwrap() + .contains("\"_top_level\":1")); + } + + #[test] + fn prepares_decoded_traces_with_top_level_tags() { + let mut metadata = metadata(); + metadata.client_computed_top_level = false; + let prepared = + prepare_agentless_traces_request(v04_traces(), &metadata, &config()).unwrap(); + + assert!(String::from_utf8(request_body(&prepared)) + .unwrap() + .contains("\"_top_level\":1")); + } + + #[test] + fn json_request_uses_configured_api_key() { + let mut headers = HeaderMap::new(); + headers.insert( + http::HeaderName::from_static("dd-api-key"), + http::HeaderValue::from_static("caller-supplied-key"), + ); + + let prepared = prepare_agentless_json_request(&config(), headers, b"[]".to_vec()).unwrap(); + assert_eq!( + prepared.request().unwrap().headers()["dd-api-key"], + "test-api-key" + ); + } + + #[test] + fn json_request_rejects_invalid_api_key() { + let mut invalid_config = config(); + invalid_config.api_key = "invalid\nkey".to_string(); + + let error = + prepare_agentless_json_request(&invalid_config, HeaderMap::new(), b"[]".to_vec()) + .unwrap_err(); + assert!(matches!(error, PrepareAgentlessError::InvalidApiKey)); + } + + #[test] + fn debug_output_redacts_agentless_secrets() { + let config = config(); + assert!(!format!("{config:?}").contains("test-api-key")); + + let prepared = prepare_agentless_v04_request(&v04_payload(), &metadata(), &config).unwrap(); + let debug = format!("{prepared:?}"); + assert!(!debug.contains("test-api-key")); + assert!(!debug.contains("operation")); + assert!(!debug.contains("resource-1")); + } +} diff --git a/libdd-data-pipeline-core/src/lib.rs b/libdd-data-pipeline-core/src/lib.rs new file mode 100644 index 0000000000..aefdfb5f15 --- /dev/null +++ b/libdd-data-pipeline-core/src/lib.rs @@ -0,0 +1,17 @@ +// Copyright 2026-Present Datadog, Inc. https://www.datadoghq.com/ +// SPDX-License-Identifier: Apache-2.0 + +#![deny(clippy::panic)] +#![deny(clippy::unwrap_used)] +#![deny(clippy::expect_used)] + +//! Runtime-independent preparation of APM data-pipeline requests. + +mod agentless; + +pub use agentless::{ + prepare_agentless_json_request, prepare_agentless_traces_request, + prepare_agentless_v04_request, AgentlessTraceConfig, PrepareAgentlessError, + PreparedAgentlessRequest, DEFAULT_AGENTLESS_TIMEOUT, +}; +pub use libdd_trace_utils::tracer_metadata::TracerMetadata; diff --git a/libdd-data-pipeline/Cargo.toml b/libdd-data-pipeline/Cargo.toml index a7510fe038..19a5a5fadd 100644 --- a/libdd-data-pipeline/Cargo.toml +++ b/libdd-data-pipeline/Cargo.toml @@ -33,6 +33,7 @@ tokio = { workspace = true, features = [ uuid = { workspace = true, features = ["v4", "std"] } tokio-util = "0.7.11" libdd-capabilities = { path = "../libdd-capabilities", version = "3.0.0" } +libdd-data-pipeline-core = { path = "../libdd-data-pipeline-core", version = "1.0.0" } libdd-common = { version = "5.2.0", path = "../libdd-common", default-features = false } libdd-shared-runtime = { version = "3.0.0", path = "../libdd-shared-runtime", default-features = false } libdd-telemetry = { version = "7.0.0", path = "../libdd-telemetry", default-features = false, optional = true} @@ -120,4 +121,4 @@ fips = [ test-utils = [] regex-lite = ["libdd-common/regex-lite"] # Enable zstd compression for the agentless trace intake sender. -compression = ["libdd-trace-utils/compression"] +compression = ["libdd-trace-utils/compression", "libdd-data-pipeline-core/compression"] diff --git a/libdd-data-pipeline/src/agentless/config.rs b/libdd-data-pipeline/src/agentless/config.rs deleted file mode 100644 index 33bc4294b6..0000000000 --- a/libdd-data-pipeline/src/agentless/config.rs +++ /dev/null @@ -1,30 +0,0 @@ -// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ -// SPDX-License-Identifier: Apache-2.0 - -//! Agentless APM trace export configuration. - -use std::{fmt::Debug, time::Duration}; - -pub const DEFAULT_AGENTLESS_TIMEOUT: Duration = Duration::from_secs(15); - -///Agentless trace exporter configuration. -#[derive(Clone)] -pub struct AgentlessTraceConfig { - /// Full URL to POST traces to (e.g. - /// `https://public-trace-http-intake.logs.datadoghq.com/v1/input`). - pub endpoint_url: String, - /// Datadog API key used for the `dd-api-key` header. - pub api_key: String, - /// Request timeout. - pub timeout: Duration, -} - -impl Debug for AgentlessTraceConfig { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("AgentlessTraceConfig") - .field("endpoint_url", &self.endpoint_url) - .field("api_key", &"") - .field("timeout", &self.timeout) - .finish() - } -} diff --git a/libdd-data-pipeline/src/agentless/exporter.rs b/libdd-data-pipeline/src/agentless/exporter.rs index 5ac9d05209..b32d3818b3 100644 --- a/libdd-data-pipeline/src/agentless/exporter.rs +++ b/libdd-data-pipeline/src/agentless/exporter.rs @@ -1,104 +1,96 @@ // Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/ // SPDX-License-Identifier: Apache-2.0 -//! Agentless HTTP/JSON trace exporter. +//! Executor-owned adapter for agentless trace export. -use super::config::AgentlessTraceConfig; use crate::trace_exporter::error::{InternalErrorKind, RequestError, TraceExporterError}; use http::HeaderMap; use libdd_capabilities::{HttpClientCapability, SleepCapability}; -use libdd_common::Endpoint; -use libdd_trace_utils::send_with_retry::{ - send_with_retry, CompressionStrategy, RetryBackoffType, RetryStrategy, SendWithRetryError, +use libdd_data_pipeline_core::{ + prepare_agentless_json_request, prepare_agentless_traces_request as prepare_traces, + AgentlessTraceConfig, PrepareAgentlessError, PreparedAgentlessRequest, }; +use libdd_trace_utils::send_with_retry::{send_prepared_with_retry, SendWithRetryError}; +use libdd_trace_utils::span::TraceData; +use libdd_trace_utils::tracer_metadata::TracerMetadata; use tracing::error; -const AGENTLESS_MAX_RETRIES: u32 = 2; -const AGENTLESS_RETRY_DELAY_MS: u64 = 1000; +pub(crate) fn prepare_agentless_traces_request( + traces: Vec>>, + metadata: &TracerMetadata, + config: &AgentlessTraceConfig, +) -> Result { + prepare_traces(traces, metadata, config).map_err(map_prepare_error) +} -/// Send an agentless trace payload (JSON bytes) to the configured intake with retries. -/// -/// `headers` should already contain all required headers (api key, content-type, meta-*, -/// entity, trace-count, etc.). `test_token` is forwarded as `X-Datadog-Test-Session-Token` -/// when set, enabling snapshot tests against a local mock. +/// Sends an encoded agentless JSON request using executor-provided capabilities. pub async fn send_agentless_traces_http( capabilities: &C, config: &AgentlessTraceConfig, headers: HeaderMap, json_body: Vec, ) -> Result<(), TraceExporterError> { - let url = libdd_common::parse_uri(&config.endpoint_url).map_err(|e| { - TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(format!( - "Invalid agentless endpoint URL: {e}" - ))) - })?; - - let target = Endpoint { - url, - timeout_ms: config.timeout.as_millis() as u64, - ..Endpoint::default() - }; - - let retry_strategy = RetryStrategy::new( - AGENTLESS_MAX_RETRIES, - AGENTLESS_RETRY_DELAY_MS, - RetryBackoffType::Exponential, - None, - ); - - #[cfg(feature = "compression")] - let compression_strategy = CompressionStrategy::Zstd { level: 1 }; - #[cfg(not(feature = "compression"))] - let compression_strategy = CompressionStrategy::None; + let prepared = + prepare_agentless_json_request(config, headers, json_body).map_err(map_prepare_error)?; + send_prepared_agentless_request(capabilities, &prepared).await +} - match send_with_retry( +pub(crate) async fn send_prepared_agentless_request( + capabilities: &C, + prepared: &PreparedAgentlessRequest, +) -> Result<(), TraceExporterError> { + send_prepared_with_retry( capabilities, - &target, - json_body, - &headers, - &retry_strategy, - compression_strategy, + prepared.request_plan(), + prepared.retry_strategy(), ) .await - { - Ok(_) => Ok(()), - Err(e) => Err(map_send_error(e)), + .map(|_| ()) + .map_err(map_send_error) +} + +fn map_prepare_error(error: PrepareAgentlessError) -> TraceExporterError { + match error { + PrepareAgentlessError::Deserialization(error) => TraceExporterError::Deserialization(error), + error => { + TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(error.to_string())) + } } } -fn map_send_error(err: SendWithRetryError) -> TraceExporterError { - match err { +fn map_send_error(error: SendWithRetryError) -> TraceExporterError { + match error { SendWithRetryError::Http(response, _) => { let status = response.status(); - let body_str = String::from_utf8_lossy(response.body()); + let body = String::from_utf8_lossy(response.body()); match status.as_u16() { 401 | 403 => error!( status = status.as_u16(), - body = %body_str, + body = %body, "Agentless authentication failed. Verify DD_API_KEY is valid." ), 404 => error!( status = status.as_u16(), - body = %body_str, + body = %body, "Agentless endpoint not found. Verify DD_SITE is correctly configured." ), 429 => error!( status = status.as_u16(), - body = %body_str, + body = %body, "Agentless intake rate-limited the request. Traces were dropped." ), 500..=599 => error!( status = status.as_u16(), - body = %body_str, + body = %body, "Agentless intake returned a server error. Traces were dropped." ), _ => error!( status = status.as_u16(), - body = %body_str, + body = %body, "Agentless intake returned an unexpected status." ), } - TraceExporterError::Request(RequestError::new(status, &body_str)) + TraceExporterError::Request(RequestError::new(status, &body)) } SendWithRetryError::Timeout(_) => { TraceExporterError::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)) diff --git a/libdd-data-pipeline/src/agentless/mod.rs b/libdd-data-pipeline/src/agentless/mod.rs index 50263ef59c..4f99ad43ca 100644 --- a/libdd-data-pipeline/src/agentless/mod.rs +++ b/libdd-data-pipeline/src/agentless/mod.rs @@ -23,8 +23,10 @@ //! any of them together causes `build`/`build_async` to return //! `BuilderErrorKind::InvalidConfiguration`. -pub(crate) mod config; pub(crate) mod exporter; -pub use config::AgentlessTraceConfig; pub use exporter::send_agentless_traces_http; +pub use libdd_data_pipeline_core::{ + prepare_agentless_v04_request, AgentlessTraceConfig, PrepareAgentlessError, + PreparedAgentlessRequest, TracerMetadata, DEFAULT_AGENTLESS_TIMEOUT, +}; diff --git a/libdd-data-pipeline/src/lib.rs b/libdd-data-pipeline/src/lib.rs index fb374b6020..2c46d4f29a 100644 --- a/libdd-data-pipeline/src/lib.rs +++ b/libdd-data-pipeline/src/lib.rs @@ -12,7 +12,7 @@ //! in different languages. pub mod agent_info; -pub(crate) mod agentless; +pub mod agentless; mod health_metrics; pub(crate) mod otlp; // `OtlpProtocol` (documented on the enum itself) is the only public symbol from the otherwise diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index e190308461..e6e5f482e7 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/libdd-data-pipeline/src/trace_exporter/builder.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::agent_info::AgentInfoFetcher; -use crate::agentless::config::{AgentlessTraceConfig, DEFAULT_AGENTLESS_TIMEOUT}; +use crate::agentless::{AgentlessTraceConfig, DEFAULT_AGENTLESS_TIMEOUT}; use crate::otlp::config::{OtlpProtocol, DEFAULT_OTLP_TIMEOUT}; use crate::otlp::{OtlpMetricsConfig, OtlpResourceInfo, OtlpTraceConfig}; #[cfg(feature = "telemetry")] @@ -12,13 +12,16 @@ use crate::trace_exporter::error::BuilderErrorKind; use crate::trace_exporter::log_writer::DEFAULT_LOG_MAX_LINE_SIZE; #[cfg(feature = "telemetry")] use crate::trace_exporter::TelemetryConfig; +#[cfg(feature = "telemetry")] +use crate::trace_exporter::TelemetryInstrumentationSessions; use crate::trace_exporter::TraceExporterWorkers; use crate::trace_exporter::{ - add_path, StatsComputationStatus, TelemetryInstrumentationSessions, TraceExporter, - TraceExporterError, TraceExporterInputFormat, TraceExporterOutputFormat, TraceSerializer, - TracerMetadata, INFO_ENDPOINT, + add_path, StatsComputationStatus, TraceExporter, TraceExporterError, TraceExporterInputFormat, + TraceExporterOutputFormat, TraceSerializer, TracerMetadata, INFO_ENDPOINT, }; -use arc_swap::{ArcSwap, ArcSwapOption}; +use arc_swap::ArcSwap; +#[cfg(feature = "telemetry")] +use arc_swap::ArcSwapOption; use libdd_capabilities::{HttpClientCapability, LogWriterCapability, MaybeSend, SleepCapability}; use libdd_common::{parse_uri, tag, Endpoint}; use libdd_dogstatsd_client::DogStatsDClient; diff --git a/libdd-data-pipeline/src/trace_exporter/error.rs b/libdd-data-pipeline/src/trace_exporter/error.rs index 1e83445da5..0f4d2f5471 100644 --- a/libdd-data-pipeline/src/trace_exporter/error.rs +++ b/libdd-data-pipeline/src/trace_exporter/error.rs @@ -5,6 +5,7 @@ use crate::telemetry::error::TelemetryError; use crate::trace_exporter::msgpack_decoder::decode::error::DecodeError; use http::StatusCode; +#[cfg(not(target_arch = "wasm32"))] use libdd_common::http_common; use rmp_serde::encode::Error as EncodeError; use std::error::Error; @@ -104,6 +105,7 @@ impl Error for NetworkError { } impl NetworkError { + #[cfg(not(target_arch = "wasm32"))] fn new>(kind: NetworkErrorKind, source: E) -> Self { Self { kind, @@ -229,6 +231,7 @@ impl From for TraceExporterError { } } +#[cfg(not(target_arch = "wasm32"))] impl From for TraceExporterError { fn from(err: http_common::Error) -> Self { match err { @@ -242,6 +245,7 @@ impl From for TraceExporterError { } } +#[cfg(not(target_arch = "wasm32"))] impl From for TraceExporterError { fn from(err: http_common::ClientError) -> Self { use http_common::ErrorKind; diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index d7c41c2b63..0e0832b79a 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -18,7 +18,10 @@ use self::metrics::MetricsEmitter; use self::stats::StatsComputationStatus; use self::trace_serializer::TraceSerializer; use crate::agent_info::ResponseObserver; -use crate::agentless::{send_agentless_traces_http, AgentlessTraceConfig}; +use crate::agentless::exporter::{ + prepare_agentless_traces_request, send_prepared_agentless_request, +}; +use crate::agentless::AgentlessTraceConfig; use crate::otlp::{map_traces_to_otlp, send_otlp_traces_http, OtlpResourceInfo, OtlpTraceConfig}; #[cfg(feature = "telemetry")] use crate::telemetry::{SendPayloadTelemetry, TelemetryClient}; @@ -68,49 +71,6 @@ const V04_TRACES_ENDPOINT: &str = "/v0.4/traces"; const V05_TRACES_ENDPOINT: &str = "/v0.5/traces"; const V1_TRACES_ENDPOINT: &str = "/v1.0/traces"; -/// Build the HTTP headers required by the agentless intake. -/// -/// Includes the API key, content-type, trace count, `Datadog-Meta-*` tracer headers, -/// and entity headers (container-id / entity-id / external-env) when available. -fn build_agentless_headers( - metadata: &TracerMetadata, - api_key: &str, - trace_count: usize, -) -> Result { - let mut headers: HeaderMap = { - let tags: TracerHeaderTags = metadata.into(); - tags.into() - }; - - let api_key_val = http::HeaderValue::from_str(api_key).map_err(|_| { - TraceExporterError::Internal(error::InternalErrorKind::InvalidWorkerState( - "Invalid Datadog API key value for dd-api-key header".to_string(), - )) - })?; - headers.insert(http::HeaderName::from_static("dd-api-key"), api_key_val); - - headers.insert( - http::header::CONTENT_TYPE, - libdd_common::header::APPLICATION_JSON, - ); - - headers.insert( - http::HeaderName::from_static("x-datadog-trace-count"), - http::HeaderValue::from(trace_count), - ); - - for (name, value) in libdd_common::entity_id::get_entity_headers() { - if let (Ok(name), Ok(value)) = ( - http::HeaderName::from_bytes(name.as_bytes()), - http::HeaderValue::from_str(value), - ) { - headers.insert(name, value); - } - } - - Ok(headers) -} - /// Values for optional telemetry HTTP session headers (`dd-session-id`, root/parent). #[derive(Debug, Default, Clone)] pub struct TelemetryInstrumentationSessions { @@ -658,19 +618,8 @@ impl< traces: Vec>>, config: &AgentlessTraceConfig, ) -> Result { - let trace_count = traces.len(); - let json_body = libdd_trace_utils::agentless_encoder::encode_payload( - &traces, - &self.metadata, - ) - .map_err(|e| { - error!("Agentless JSON serialization error: {e}"); - TraceExporterError::Internal(InternalErrorKind::InvalidWorkerState(e.to_string())) - })?; - - let headers = build_agentless_headers(&self.metadata, &config.api_key, trace_count)?; - - send_agentless_traces_http(&self.capabilities, config, headers, json_body).await?; + let prepared = prepare_agentless_traces_request(traces, &self.metadata, config)?; + send_prepared_agentless_request(&self.capabilities, &prepared).await?; Ok(AgentResponse::Unchanged) } @@ -804,14 +753,6 @@ impl< let mut header_tags: TracerHeaderTags = self.metadata.borrow().into(); if let Some(ref config) = self.agentless_config { - // For agentless we want to tag top level spans, but not perform - // stats aggregation or span drops - if !self.client_computed_top_level { - for chunk in traces.iter_mut() { - libdd_trace_utils::span::trace_utils::compute_top_level_span(chunk); - } - } - return self.send_agentless_traces_inner(traces, config).await; } diff --git a/libdd-http-client/Cargo.toml b/libdd-http-client/Cargo.toml index 85003c695f..c23fc0c1c8 100644 --- a/libdd-http-client/Cargo.toml +++ b/libdd-http-client/Cargo.toml @@ -18,7 +18,13 @@ default = ["https", "reqwest-backend"] # installed explicitly by libdd-common's connector init. https = ["dep:reqwest", "reqwest?/rustls-no-provider", "libdd-common?/https"] reqwest-backend = ["dep:reqwest", "reqwest?/hickory-dns", "reqwest?/multipart"] -hyper-backend = ["dep:libdd-common", "dep:hyper", "dep:hyper-util", "dep:http-body-util"] +hyper-backend = [ + "dep:libdd-common", + "libdd-common/http-client", + "dep:hyper", + "dep:hyper-util", + "dep:http-body-util", +] fips = ["dep:reqwest", "reqwest?/rustls-no-provider", "dep:rustls", "rustls?/aws-lc-rs", "libdd-common?/fips"] [dependencies] diff --git a/libdd-trace-utils/Cargo.toml b/libdd-trace-utils/Cargo.toml index fcc9dfd105..4cc62b3ec1 100644 --- a/libdd-trace-utils/Cargo.toml +++ b/libdd-trace-utils/Cargo.toml @@ -39,7 +39,6 @@ tracing.workspace = true serde_json = { workspace = true, features = ["std"] } serde-transcode = "1.1" futures.workspace = true -tokio = { workspace = true, features = ["macros"] } rand = "0.8.5" bytes = "1.11.1" rmpv = { version = "1.3.0", default-features = false } @@ -68,8 +67,8 @@ httpmock = { version = "0.8.0-alpha.1", optional = true } urlencoding = { version = "2.1.3", optional = true } [target.'cfg(not(target_arch = "wasm32"))'.dependencies] -tokio = { version = "1", features = ["time", "rt-multi-thread"] } -libdd-capabilities-impl = { version = "4.0.0", path = "../libdd-capabilities-impl", default-features = false } +libdd-capabilities-impl = { version = "4.0.0", path = "../libdd-capabilities-impl", default-features = false, optional = true } +tokio = { version = "1", features = ["time"], optional = true } zstd = { version = "0.13.3", default-features = false, optional = true } [target.'cfg(target_arch = "wasm32")'.dependencies] @@ -91,9 +90,10 @@ tempfile.workspace = true [features] default = ["https"] -https = ["libdd-common/https", "libdd-capabilities-impl/https"] +https = ["libdd-common/https", "dep:libdd-capabilities-impl", "libdd-capabilities-impl/https"] mini_agent = ["compression", "libdd-common/use_webpki_roots", "dep:flate2"] test-utils = [ + "dep:tokio", "hyper/server", "httpmock", "cargo_metadata", @@ -108,4 +108,4 @@ compression = ["zstd"] # target: zstd on native platforms and zrip on WASM. zstd = ["dep:zstd", "dep:zrip"] # FIPS mode uses the FIPS-compliant cryptographic provider (Unix only) -fips = ["libdd-common/fips", "libdd-capabilities-impl/fips"] +fips = ["libdd-common/fips", "dep:libdd-capabilities-impl", "libdd-capabilities-impl/fips"] diff --git a/libdd-trace-utils/src/send_with_retry/mod.rs b/libdd-trace-utils/src/send_with_retry/mod.rs index 41129f4d9d..e1eba8a9f4 100644 --- a/libdd-trace-utils/src/send_with_retry/mod.rs +++ b/libdd-trace-utils/src/send_with_retry/mod.rs @@ -11,16 +11,88 @@ pub(crate) mod compression; pub use compression::CompressionStrategy; use bytes::Bytes; +use futures::future::{select, Either}; use http::HeaderMap; use libdd_capabilities::{HttpClientCapability, HttpError, SleepCapability}; use libdd_common::Endpoint; -use std::time::Duration; +use std::{fmt, time::Duration}; use tracing::{debug, error}; pub type Attempts = u32; pub type SendWithRetryResult = Result<(http::Response, Attempts), SendWithRetryError>; +/// An HTTP request prepared for repeated transport attempts. +/// +/// Payload compression and invariant header construction happen once. Each +/// call to [`Self::request`] creates a fresh request with a cheaply cloned +/// [`Bytes`] body, allowing a host runtime to implement retries without using +/// libdatadog's HTTP client or async executor. +#[derive(Clone)] +pub struct PreparedRequest { + target: Endpoint, + payload: Bytes, + headers: HeaderMap, + compression_strategy: CompressionStrategy, +} + +impl fmt::Debug for PreparedRequest { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("PreparedRequest") + .field("url", &self.target.url) + .field("timeout", &self.timeout()) + .field("payload_len", &self.payload.len()) + .field("compression_strategy", &self.compression_strategy) + .finish() + } +} + +impl PreparedRequest { + /// Prepares a payload and its invariant request metadata. + pub fn new( + target: Endpoint, + payload: Vec, + headers: HeaderMap, + compression_strategy: CompressionStrategy, + ) -> Self { + let (payload, compression_strategy) = compression::compress(payload, compression_strategy); + Self { + target, + payload: Bytes::from(payload), + headers, + compression_strategy, + } + } + + /// Builds one transport attempt. + pub fn request(&self) -> Result, http::Error> { + let mut builder = http::Request::builder() + .method(http::Method::POST) + .uri(self.target.url.clone()); + builder = self + .target + .set_standard_headers(builder, concat!("Tracer/", env!("CARGO_PKG_VERSION"))); + for (key, value) in &self.headers { + builder = builder.header(key, value); + } + if let Some(headers) = builder.headers_mut() { + compression::add_headers(headers, self.compression_strategy); + } + builder.body(self.payload.clone()) + } + + /// Returns the timeout for one transport attempt. + pub fn timeout(&self) -> Duration { + Duration::from_millis(self.target.timeout_ms) + } + + /// Returns the prepared payload after any configured compression. + pub fn payload(&self) -> &Bytes { + &self.payload + } +} + /// All errors contain the number of attempts after which the final error was returned #[derive(Debug)] pub enum SendWithRetryError { @@ -105,66 +177,72 @@ pub async fn send_with_retry( headers: &HeaderMap, retry_strategy: &RetryStrategy, compression_strategy: CompressionStrategy, +) -> SendWithRetryResult { + let prepared = PreparedRequest::new( + target.clone(), + payload, + headers.clone(), + compression_strategy, + ); + send_prepared_with_retry(capabilities, &prepared, retry_strategy).await +} + +/// Sends a pre-built request plan with retries. +/// +/// This is the executor-driven counterpart to the host-driven API exposed by +/// [`PreparedRequest`] and [`RetryStrategy`]. +#[allow(clippy::result_large_err)] +pub async fn send_prepared_with_retry( + capabilities: &C, + prepared: &PreparedRequest, + retry_strategy: &RetryStrategy, ) -> SendWithRetryResult { let mut request_attempt = 0; - let timeout = Duration::from_millis(target.timeout_ms); + let timeout = prepared.timeout(); debug!( - url = %target.url, - payload_size = payload.len(), + url = %prepared.target.url, + payload_size = prepared.payload.len(), max_retries = retry_strategy.max_retries(), "Sending with retry" ); - let (compressed, compression_strategy) = compression::compress(payload, compression_strategy); - let payload = Bytes::from(compressed); - loop { request_attempt += 1; debug!( - url = %target.url, + url = %prepared.target.url, attempt = request_attempt, max_retries = retry_strategy.max_retries(), "Attempting request" ); - let mut builder = http::Request::builder() - .method(http::Method::POST) - .uri(target.url.clone()); - builder = - target.set_standard_headers(builder, concat!("Tracer/", env!("CARGO_PKG_VERSION"))); - for (key, value) in headers { - builder = builder.header(key, value); - } - // headers_mut is only None if the builder is in an error state - if let Some(h) = builder.headers_mut() { - compression::add_headers(h, compression_strategy); - } - let req = match builder.body(payload.clone()) { + let req = match prepared.request() { Ok(r) => r, Err(_) => { return Err(SendWithRetryError::Build(request_attempt)); } }; - let result = tokio::select! { - biased; - r = capabilities.request(req) => Ok(r), - _ = capabilities.sleep(timeout) => Err(()), + let request = capabilities.request(req); + let timeout = capabilities.sleep(timeout); + futures::pin_mut!(request, timeout); + let result = match select(request, timeout).await { + Either::Left((response, _)) => Ok(response), + Either::Right(((), _)) => Err(()), }; match result { Ok(Ok(response)) => { let status = response.status(); debug!( - url = %target.url, + url = %prepared.target.url, status = status.as_u16(), attempt = request_attempt, "Received response" ); - if status.is_client_error() || status.is_server_error() { + if RetryStrategy::is_retryable_status(status) { debug!( status = status.as_u16(), attempt = request_attempt, @@ -199,7 +277,7 @@ pub async fn send_with_retry( } Ok(Err(e)) => { debug!( - url = %target.url, + url = %prepared.target.url, error = ?e, attempt = request_attempt, max_retries = retry_strategy.max_retries(), @@ -233,7 +311,7 @@ pub async fn send_with_retry( } Err(_) => { debug!( - url = %target.url, + url = %prepared.target.url, attempt = request_attempt, max_retries = retry_strategy.max_retries(), "Request timed out" @@ -263,10 +341,69 @@ pub async fn send_with_retry( mod tests { use super::*; use crate::test_utils::poll_for_mock_hit; + use http::{HeaderName, HeaderValue, Method}; use httpmock::MockServer; use libdd_capabilities::HttpClientCapability; use libdd_capabilities_impl::NativeCapabilities; + #[test] + fn prepared_request_builds_repeatable_attempts() { + let target = Endpoint { + url: "https://example.test/v1/input".parse().unwrap(), + timeout_ms: 1_234, + test_token: Some("test-token".into()), + ..Default::default() + }; + let mut headers = HeaderMap::new(); + headers.insert( + HeaderName::from_static("x-datadog-trace-count"), + HeaderValue::from_static("2"), + ); + let prepared = + PreparedRequest::new(target, vec![1, 2, 3], headers, CompressionStrategy::None); + + let first = prepared.request().unwrap(); + let second = prepared.request().unwrap(); + + assert_eq!(first.method(), Method::POST); + assert_eq!(first.uri(), "https://example.test/v1/input"); + assert_eq!(first.headers()["x-datadog-trace-count"], "2"); + assert_eq!( + first.headers()["x-datadog-test-session-token"], + "test-token" + ); + assert_eq!(first.body().as_ref(), &[1, 2, 3]); + assert_eq!(first.body(), second.body()); + assert_eq!(prepared.payload().as_ref(), &[1, 2, 3]); + assert_eq!(prepared.timeout(), Duration::from_millis(1_234)); + } + + #[test] + fn prepared_request_debug_redacts_sensitive_data() { + let target = Endpoint { + url: "https://example.test/v1/input".parse().unwrap(), + api_key: Some("endpoint-secret".into()), + ..Default::default() + }; + let mut headers = HeaderMap::new(); + headers.insert( + HeaderName::from_static("dd-api-key"), + HeaderValue::from_static("header-secret"), + ); + let prepared = PreparedRequest::new( + target, + b"payload-secret".to_vec(), + headers, + CompressionStrategy::None, + ); + + let debug = format!("{prepared:?}"); + assert!(!debug.contains("endpoint-secret")); + assert!(!debug.contains("header-secret")); + assert!(!debug.contains("payload-secret")); + assert!(debug.contains("payload_len: 14")); + } + #[cfg_attr(miri, ignore)] #[tokio::test] async fn test_zero_retries_on_error() { diff --git a/libdd-trace-utils/src/send_with_retry/retry_strategy.rs b/libdd-trace-utils/src/send_with_retry/retry_strategy.rs index 88d54e3f76..c8e8720962 100644 --- a/libdd-trace-utils/src/send_with_retry/retry_strategy.rs +++ b/libdd-trace-utils/src/send_with_retry/retry_strategy.rs @@ -93,26 +93,51 @@ impl RetryStrategy { /// * `attempt`: The number of the current attempt (1-indexed). /// * `capabilities`: Provides the sleep capability for the delay. pub(crate) async fn delay(&self, attempt: u32, capabilities: &C) { - let delay = match self.backoff_type { - RetryBackoffType::Exponential => self.delay_ms * 2u32.pow(attempt - 1), - RetryBackoffType::Constant => self.delay_ms, - RetryBackoffType::Linear => self.delay_ms + (self.delay_ms * (attempt - 1)), + capabilities.sleep(self.delay_for_attempt(attempt)).await; + } + + fn delay_for_attempt(&self, attempt: u32) -> Duration { + let retry_index = attempt.saturating_sub(1); + let multiplier = match self.backoff_type { + RetryBackoffType::Exponential => 2u32.checked_pow(retry_index).unwrap_or(u32::MAX), + RetryBackoffType::Constant => 1, + RetryBackoffType::Linear => retry_index.saturating_add(1), }; + let delay = self + .delay_ms + .checked_mul(multiplier) + .unwrap_or(Duration::MAX); if let Some(jitter) = self.jitter { - let jitter = rand::random::() % jitter.as_millis() as u64; - capabilities - .sleep(delay + Duration::from_millis(jitter)) - .await; + let jitter_ms = u64::try_from(jitter.as_millis()).unwrap_or(u64::MAX); + if jitter_ms == 0 { + delay + } else { + let randomized_ms = rand::random::() % jitter_ms; + delay.saturating_add(Duration::from_millis(randomized_ms)) + } } else { - capabilities.sleep(delay).await; + delay } } /// Returns the maximum number of retries. - pub(crate) fn max_retries(&self) -> u32 { + pub fn max_retries(&self) -> u32 { self.max_retries } + + /// Returns whether an HTTP response should be retried. + pub fn is_retryable_status(status: http::StatusCode) -> bool { + status.is_client_error() || status.is_server_error() + } + + /// Returns the delay before retrying after `attempt`. + /// + /// `attempt` is one-indexed. This returns `None` for attempt zero or when + /// the retry limit has been reached. + pub fn retry_delay(&self, attempt: u32) -> Option { + (attempt != 0 && attempt <= self.max_retries).then(|| self.delay_for_attempt(attempt)) + } } #[cfg(test)] @@ -300,4 +325,34 @@ mod tests { "Max retries did not match expected value" ); } + + #[test] + fn test_retry_delays_for_host_runtime() { + let retry_strategy = RetryStrategy::new(3, 100, RetryBackoffType::Exponential, None); + + assert_eq!(retry_strategy.retry_delay(0), None); + assert_eq!( + retry_strategy.retry_delay(1), + Some(Duration::from_millis(100)) + ); + assert_eq!( + retry_strategy.retry_delay(2), + Some(Duration::from_millis(200)) + ); + assert_eq!( + retry_strategy.retry_delay(3), + Some(Duration::from_millis(400)) + ); + assert_eq!(retry_strategy.retry_delay(4), None); + } + + #[test] + fn test_zero_jitter_is_supported() { + let retry_strategy = RetryStrategy::new(1, 100, RetryBackoffType::Constant, Some(0)); + + assert_eq!( + retry_strategy.retry_delay(1), + Some(Duration::from_millis(100)) + ); + } }