fix(remote-config)!: Expose HttpClientCapability in remote config

datadog-sidecar/src/shm_remote_config.rs

@@ -10,6 +10,7 @@ use datadog_ipc::one_way_shared_memory::{open_named_shm, OneWayShmReader, OneWay
 use datadog_ipc::platform::{FileBackedHandle, NamedShmHandle};
 use datadog_ipc::rate_limiter::ShmLimiter;
 use datadog_live_debugger::LiveDebuggingData;
+use libdd_capabilities_impl::NativeHttpClient;
 use libdd_common::{tag::Tag, MutexExt};
 use libdd_remote_config::config::dynamic::{parse_json, Configs};
 use libdd_remote_config::fetch::{
@@ -245,10 +246,12 @@ fn dynamic_instrumentation_is_enabled(apm_config: Option<bool>, info: &TargetInf
     }
 }
 
-impl<N: NotifyTarget + 'static> MultiTargetHandlers<N, Self> for ConfigFileStorage<N> {
+impl<N: NotifyTarget + 'static> MultiTargetHandlers<N, Self, NativeHttpClient>
+    for ConfigFileStorage<N>
+{
     fn fetched(
         &self,
-        fetcher: &Arc<MultiTargetFetcher<N, Self>>,
+        fetcher: &Arc<MultiTargetFetcher<N, Self, NativeHttpClient>>,
         runtime_id: &Arc<String>,
         target: &Arc<Target>,
         files: &[Arc<StoredShmFile>],
@@ -422,7 +425,7 @@ impl<N: NotifyTarget + 'static> Drop for ShmRemoteConfigsGuard<N> {
 
 #[derive(Clone)]
 pub struct ShmRemoteConfigs<N: NotifyTarget + 'static>(
-    Arc<MultiTargetFetcher<N, ConfigFileStorage<N>>>,
+    Arc<MultiTargetFetcher<N, ConfigFileStorage<N>, NativeHttpClient>>,
 );
 
 // we collect services per env, so that we always query, for each runtime + env, all the services
@@ -448,7 +451,8 @@ impl<N: NotifyTarget + 'static> ShmRemoteConfigs<N> {
             on_dead: Arc::new(Mutex::new(Some(on_dead))),
             _phantom: Default::default(),
         };
-        let fetcher = MultiTargetFetcher::new(storage, invariants);
+        let fetcher =
+            MultiTargetFetcher::new(storage, invariants, NativeHttpClient::new_periodic_client());
         fetcher
             .remote_config_interval
             .store(interval.as_nanos() as u64, Ordering::Relaxed);

libdd-capabilities-impl/src/http.rs

@@ -9,27 +9,47 @@ mod native {
     use libdd_capabilities::http::{HttpClientCapability, HttpError};
     use libdd_capabilities::maybe_send::MaybeSend;
     use libdd_common::connector::Connector;
-    use libdd_common::http_common::{new_default_client, Body, GenericHttpClient};
+    use libdd_common::http_common::{
+        new_client_periodic, new_default_client, Body, GenericHttpClient,
+    };
 
     use http_body_util::BodyExt;
 
     #[derive(Clone)]
     pub struct NativeHttpClient {
         client: Arc<OnceLock<GenericHttpClient<Connector>>>,
+        periodic: bool,
     }
 
     impl std::fmt::Debug for NativeHttpClient {
         fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
             f.debug_struct("NativeHttpClient")
                 .field("initialized", &self.client.get().is_some())
+                .field("periodic", &self.periodic)
                 .finish()
         }
     }
 
+    impl NativeHttpClient {
+        /// Like [`HttpClientCapability::new_client`], but disables connection pooling.
+        ///
+        /// Intended for clients that issue requests on a fixed interval (e.g. remote
+        /// config polling): the agent's low keep-alive setting can close an idle
+        /// connection between polls, which turns a pooled/reused connection into
+        /// intermittent request failures.
+        pub fn new_periodic_client() -> Self {
+            Self {
+                client: Arc::new(OnceLock::new()),
+                periodic: true,
+            }
+        }
+    }
+
     impl HttpClientCapability for NativeHttpClient {
         fn new_client() -> Self {
             Self {
                 client: Arc::new(OnceLock::new()),
+                periodic: false,
             }
         }
 
@@ -39,7 +59,17 @@ mod native {
             req: http::Request<bytes::Bytes>,
         ) -> impl std::future::Future<Output = Result<http::Response<bytes::Bytes>, HttpError>> + MaybeSend
         {
-            let client = self.client.get_or_init(new_default_client).clone();
+            let periodic = self.periodic;
+            let client = self
+                .client
+                .get_or_init(|| {
+                    if periodic {
+                        new_client_periodic()
+                    } else {
+                        new_default_client()
+                    }
+                })
+                .clone();
             async move {
                 let hyper_req = req.map(Body::from_bytes);
 

libdd-remote-config/Cargo.toml

@@ -13,6 +13,8 @@ description = "Datadog Remote Configuration client and config parsers"
 default = ["client", "https"]
 client = [
     "libdd-trace-protobuf",
+    "libdd-capabilities",
+    "bytes",
     "http-body-util",
     "http",
     "base64",
@@ -28,14 +30,16 @@ client = [
 
 regex-lite = ["libdd-common/regex-lite"]
 # Enable HTTPS support in the fetcher (rustls + ring via libdd-common).
-https = ["libdd-common/https"]
+https = ["libdd-common/https", "libdd-capabilities-impl/https"]
 # FIPS-compliant crypto provider (aws-lc-rs via libdd-common). Unix only.
-fips = ["libdd-common/fips"]
+fips = ["libdd-common/fips", "libdd-capabilities-impl/fips"]
 test = ["hyper/server", "hyper-util"]
 
 [dependencies]
 anyhow = { version = "1.0" }
+bytes = { version = "1", optional = true }
 libdd-common = { path = "../libdd-common", version = "5.1.0", default-features = false }
+libdd-capabilities = { path = "../libdd-capabilities", version = "2.1.0", optional = true }
 libdd-trace-protobuf = { path = "../libdd-trace-protobuf", version = "4.0.0", optional = true }
 hyper = { workspace = true, optional = true, default-features = false }
 http-body-util = {version = "0.1", optional = true }
@@ -61,6 +65,9 @@ strum_macros = "0.26"
 # Test feature
 hyper-util = { workspace = true, features = ["service"], optional = true }
 
+[target.'cfg(not(target_arch = "wasm32"))'.dependencies]
+libdd-capabilities-impl = { path = "../libdd-capabilities-impl", version = "3.0.0", default-features = false }
+
 [dev-dependencies]
 futures = "0.3"
 libdd-remote-config = { path = ".", features = ["test"] }

libdd-remote-config/examples/remote_config_fetch.rs

@@ -1,6 +1,7 @@
 // Copyright 2021-Present Datadog, Inc. https://www.datadoghq.com/
 // SPDX-License-Identifier: Apache-2.0
 
+use libdd_capabilities_impl::NativeHttpClient;
 use libdd_common::Endpoint;
 use libdd_remote_config::fetch::{ConfigInvariants, ConfigOptions, SingleChangesFetcher};
 use libdd_remote_config::file_change_tracker::{Change, FilePath};
@@ -50,6 +51,7 @@ async fn main() {
             products: vec![ApmTracing],
             capabilities: vec![],
         },
+        NativeHttpClient::new_periodic_client(),
     );
 
     loop {

libdd-remote-config/src/fetch/fetcher.rs

@@ -7,8 +7,8 @@ use base64::Engine;
 use hashbrown::HashMap;
 use http::uri::PathAndQuery;
 use http::StatusCode;
-use http_body_util::BodyExt;
-use libdd_common::{http_common, Endpoint, MutexExt};
+use libdd_capabilities::HttpClientCapability;
+use libdd_common::{Endpoint, MutexExt};
 use libdd_trace_protobuf::remoteconfig::{
     ClientGetConfigsRequest, ClientGetConfigsResponse, ClientState, ClientTracer, ConfigState,
     TargetFileHash, TargetFileMeta,
@@ -104,11 +104,12 @@ impl ConfigProductCapabilities {
     }
 }
 
-pub struct ConfigFetcherState<S> {
+pub struct ConfigFetcherState<S, C: HttpClientCapability> {
     target_files_by_path: Mutex<HashMap<Arc<RemoteConfigPath>, StoredTargetFile<S>>>,
     pub invariants: ConfigInvariants,
     endpoint: Endpoint,
     pub expire_unused_files: bool,
+    http_client: C,
 }
 
 #[derive(Debug, Default, Serialize, Deserialize)]
@@ -152,13 +153,14 @@ impl<S> ConfigFetcherFilesLock<'_, S> {
     }
 }
 
-impl<S> ConfigFetcherState<S> {
-    pub fn new(invariants: ConfigInvariants) -> Self {
+impl<S, C: HttpClientCapability> ConfigFetcherState<S, C> {
+    pub fn with_client(invariants: ConfigInvariants, http_client: C) -> Self {
         ConfigFetcherState {
             target_files_by_path: Default::default(),
             endpoint: get_agent_configs_endpoint(&invariants.endpoint),
             invariants,
             expire_unused_files: true,
+            http_client,
         }
     }
 
@@ -201,9 +203,9 @@ impl<S> ConfigFetcherState<S> {
     }
 }
 
-pub struct ConfigFetcher<S: FileStorage> {
+pub struct ConfigFetcher<S: FileStorage, C: HttpClientCapability> {
     pub file_storage: S,
-    state: Arc<ConfigFetcherState<S::StoredFile>>,
+    state: Arc<ConfigFetcherState<S::StoredFile, C>>,
 }
 
 pub struct ConfigClientState {
@@ -236,8 +238,8 @@ impl ConfigClientState {
     }
 }
 
-impl<S: FileStorage> ConfigFetcher<S> {
-    pub fn new(file_storage: S, state: Arc<ConfigFetcherState<S::StoredFile>>) -> Self {
+impl<S: FileStorage, C: HttpClientCapability> ConfigFetcher<S, C> {
+    pub fn new(file_storage: S, state: Arc<ConfigFetcherState<S::StoredFile, C>>) -> Self {
         ConfigFetcher {
             file_storage,
             state,
@@ -361,16 +363,16 @@ impl<S: FileStorage> ConfigFetcher<S> {
                 http::header::CONTENT_TYPE,
                 libdd_common::header::APPLICATION_JSON,
             )
-            .body(http_common::Body::from(serde_json::to_string(&config_req)?))?;
+            .body(bytes::Bytes::from(serde_json::to_string(&config_req)?))?;
         let response = tokio::time::timeout(
             Duration::from_millis(self.state.endpoint.timeout_ms),
-            http_common::new_default_client().request(req),
+            self.state.http_client.request(req),
         )
         .await
         .map_err(|e| anyhow::Error::msg(e).context(format!("Url: {:?}", self.state.endpoint)))?
         .map_err(|e| anyhow::Error::msg(e).context(format!("Url: {:?}", self.state.endpoint)))?;
         let status = response.status();
-        let body_bytes = response.into_body().collect().await?.to_bytes();
+        let body_bytes = response.into_body();
         if status != StatusCode::OK {
             // Not active
             if status == StatusCode::NOT_FOUND {
@@ -578,6 +580,8 @@ pub mod tests {
     use crate::fetch::test_server::RemoteConfigServer;
     use crate::RemoteConfigSource;
     use http::Response;
+    use libdd_capabilities_impl::NativeHttpClient;
+    use libdd_common::http_common;
     use std::mem::transmute;
     use std::sync::LazyLock;
 
@@ -690,7 +694,10 @@ pub mod tests {
         let storage = Arc::new(Storage::default());
         let mut fetcher = ConfigFetcher::new(
             storage.clone(),
-            Arc::new(ConfigFetcherState::new(server.dummy_options().invariants)),
+            Arc::new(ConfigFetcherState::with_client(
+                server.dummy_options().invariants,
+                NativeHttpClient::new_periodic_client(),
+            )),
         );
         let mut opaque_state = ConfigClientState::default();
 
@@ -740,7 +747,10 @@ pub mod tests {
 
         let mut fetcher = ConfigFetcher::new(
             storage.clone(),
-            Arc::new(ConfigFetcherState::new(invariants)),
+            Arc::new(ConfigFetcherState::with_client(
+                invariants,
+                NativeHttpClient::new_periodic_client(),
+            )),
         );
         let mut opaque_state = ConfigClientState::default();
 
@@ -924,7 +934,10 @@ pub mod tests {
         let storage = Arc::new(Storage::default());
         let mut fetcher = ConfigFetcher::new(
             storage,
-            Arc::new(ConfigFetcherState::new(server.dummy_options().invariants)),
+            Arc::new(ConfigFetcherState::with_client(
+                server.dummy_options().invariants,
+                NativeHttpClient::new_periodic_client(),
+            )),
         );
         let mut opaque_state = ConfigClientState::default();
 
@@ -1022,7 +1035,10 @@ pub mod tests {
         let storage = Arc::new(Storage::default());
         let mut fetcher = ConfigFetcher::new(
             storage,
-            Arc::new(ConfigFetcherState::new(server.dummy_options().invariants)),
+            Arc::new(ConfigFetcherState::with_client(
+                server.dummy_options().invariants,
+                NativeHttpClient::new_periodic_client(),
+            )),
         );
         let mut opaque_state = ConfigClientState::default();