feat(data-pipeline): add runtime-independent request preparation

.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

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

@@ -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)'] }
 

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<connect::HttpConnector>),
+}
+
+static DEFAULT_CONNECTOR: LazyLock<Connector> = 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<ConnStream, ConnStreamError>> {
+        match self {
+            Self::Http(c) => {
+                if require_tls {
+                    future::err::<ConnStream, ConnStreamError>(
+                        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<hyper_util::client::legacy::connect::HttpConnector>,
+    > {
+        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<hyper_util::client::legacy::connect::HttpConnector>,
+    > {
+        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<hyper::Uri> 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<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + 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<Result<(), Self::Error>> {
+        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}"
+        );
+    }
+}