feat(data-pipeline): add runtime-independent agentless sending

.github/workflows/test.yml

@@ -138,6 +138,20 @@ 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|trace-utils)"; then
+            cargo check -p libdd-dogstatsd-client --no-default-features
+            cargo check -p libdd-http-client --no-default-features \
+              --features hyper-backend
+            cargo check -p libdd-trace-utils --no-default-features \
+              --features mini_agent
+            cargo check -p libdd-trace-utils --no-default-features \
+              --features test-utils
+          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

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",

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]

libdd-common/Cargo.toml

@@ -55,15 +55,17 @@ 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"
+# Target gating keeps these dependencies out of WASM. They are also optional so
+# native callers using host-managed capabilities do not pull in an HTTP runtime.
+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 +94,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 +118,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)

libdd-common/src/connector/mod.rs

@@ -1,13 +1,20 @@
 // Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
+#[cfg(feature = "http-client")]
 use futures::future::BoxFuture;
+#[cfg(feature = "http-client")]
 use futures::{future, FutureExt};
+#[cfg(feature = "http-client")]
 use hyper_util::client::legacy::connect;
 
+#[cfg(feature = "http-client")]
 use core::future::Future;
+#[cfg(feature = "http-client")]
 use core::pin::Pin;
+#[cfg(feature = "http-client")]
 use core::task::{Context, Poll};
+#[cfg(feature = "http-client")]
 use std::sync::LazyLock;
 
 #[cfg(unix)]
@@ -17,24 +24,30 @@ pub mod named_pipe;
 
 pub mod errors;
 
+#[cfg(feature = "http-client")]
 mod conn_stream;
+#[cfg(feature = "http-client")]
 use conn_stream::{ConnStream, ConnStreamError};
 
+#[cfg(feature = "http-client")]
 #[derive(Clone)]
 pub enum Connector {
     Http(connect::HttpConnector),
     #[cfg(feature = "tls-core")]
     Https(hyper_rustls::HttpsConnector<connect::HttpConnector>),
 }
 
+#[cfg(feature = "http-client")]
 static DEFAULT_CONNECTOR: LazyLock<Connector> = LazyLock::new(Connector::new);
 
+#[cfg(feature = "http-client")]
 impl Default for Connector {
     fn default() -> Self {
         DEFAULT_CONNECTOR.clone()
     }
 }
 
+#[cfg(feature = "http-client")]
 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().
@@ -145,6 +158,7 @@ mod https {
     }
 }
 
+#[cfg(feature = "http-client")]
 impl tower_service::Service<hyper::Uri> for Connector {
     type Response = ConnStream;
     type Error = ConnStreamError;
@@ -172,7 +186,7 @@ impl tower_service::Service<hyper::Uri> for Connector {
     }
 }
 
-#[cfg(test)]
+#[cfg(all(test, feature = "http-client"))]
 mod tests {
     use crate::http_common;
     #[cfg(any(feature = "use_webpki_roots", target_os = "linux"))]

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<hyper::Uri, hyper::http::Error> {
+pub fn named_pipe_path_to_uri(path: &Path) -> Result<http::Uri, http::Error> {
     #[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<PathBuf> {
+pub fn named_pipe_path_from_uri(uri: &http::Uri) -> anyhow::Result<PathBuf> {
     if uri.scheme_str() != Some("windows") {
         return Err(super::errors::Error::InvalidUrl.into());
     }

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<hyper::Uri, hyper::http::Error> {
+pub fn socket_path_to_uri(path: &Path) -> Result<http::Uri, http::Error> {
     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<PathBuf> {
+pub fn socket_path_from_uri(uri: &http::Uri) -> anyhow::Result<PathBuf> {
     if uri.scheme_str() != Some("unix") {
         return Err(super::errors::Error::InvalidUrl.into());
     }

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<connector::Connector>;
-#[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<C: hyper_util::client::legacy::connect::Connect + Clone + Send + Sync + 'static> Connect
     for C
 {

libdd-data-pipeline-core/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "libdd-data-pipeline-core"
+version = "1.0.0"
+description = "Runtime-independent operations 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]
+http = "1"
+serde_json.workspace = true
+thiserror = "1.0"
+libdd-capabilities = { version = "3.0.0", path = "../libdd-capabilities" }
+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]
+bytes = "1.11.1"
+futures = { workspace = true, features = ["executor"] }
+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"]

libdd-data-pipeline-core/src/agentless/config.rs

@@ -0,0 +1,30 @@
+// 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", &"<redacted>")
+            .field("timeout", &self.timeout)
+            .finish()
+    }
+}