fix(http): consolidate clients (#1062)

## Overview

The extension was constructing up to 6 separate HTTP clients across
different subsystems (metrics, logs, trace proxy, trace flusher, stats
flusher, and Lambda API). Three of the reqwest::Client instances
(metrics, logs, trace proxy) were identically configured via
get_client(config), and two HttpClient (hyper) instances (trace flusher,
stats flusher) were identically configured via
http_client::create_client().

Each independent client maintains its own connection pool and TLS
sessions, which means:
  - Duplicate TLS handshakes to the same Datadog intake hosts
  - No connection reuse across subsystems hitting the same endpoints
  - Unnecessary memory overhead from redundant connection pools

  This change consolidates to 3 clients total:
1. reqwest::Client (no-proxy) — Lambda Extensions API (intentionally
isolated, localhost only)
  2. reqwest::Client (shared) — Metrics, Logs, Trace proxy
  3. HttpClient (shared) — Trace flusher, Stats flusher

Both reqwest::Client and HttpClient are Arc-based internally, so cloning
is a refcount increment that shares the underlying connection pool.

This also removes the lazy OnceCell initialization from TraceFlusher and
StatsFlusher in favor of eager construction at startup. The lazy init
was retrying on failure, but the only failure modes (bad proxy URL,
unreadable cert file) are configuration errors that won't
  self-resolve between retries, so failing fast is preferable.

## Testing 

Unit tests, integration tests, and self monitoring

Author: duncanista

bottlecap/src/bin/bottlecap/main.rs

@@ -60,6 +60,7 @@ use bottlecap::{
         provider::Provider as TagProvider,
     },
     traces::{
+        http_client as trace_http_client,
         propagation::DatadogCompositePropagator,
         proxy_aggregator,
         proxy_flusher::Flusher as ProxyFlusher,
@@ -292,17 +293,28 @@ async fn extension_loop_active(
     let account_id = r.account_id.as_ref().unwrap_or(&"none".to_string()).clone();
     let tags_provider = setup_tag_provider(&Arc::clone(&aws_config), config, &account_id);
 
+    // Build one shared reqwest::Client for metrics, logs, and trace proxy flushing.
+    // reqwest::Client is Arc-based internally, so cloning just increments a refcount
+    // and shares the connection pool.
+    let shared_client = bottlecap::http::get_client(config);
+
     let (logs_agent_channel, logs_flusher, logs_agent_cancel_token, logs_aggregator_handle) =
         start_logs_agent(
             config,
             Arc::clone(&api_key_factory),
             &tags_provider,
             event_bus_tx.clone(),
             aws_config.is_managed_instance_mode(),
+            &shared_client,
         );
 
-    let (metrics_flushers, metrics_aggregator_handle, dogstatsd_cancel_token) =
-        start_dogstatsd(tags_provider.clone(), Arc::clone(&api_key_factory), config).await;
+    let (metrics_flushers, metrics_aggregator_handle, dogstatsd_cancel_token) = start_dogstatsd(
+        tags_provider.clone(),
+        Arc::clone(&api_key_factory),
+        config,
+        &shared_client,
+    )
+    .await;
 
     let propagator = Arc::new(DatadogCompositePropagator::new(Arc::clone(config)));
     // Lifecycle Invocation Processor
@@ -345,6 +357,7 @@ async fn extension_loop_active(
         &tags_provider,
         invocation_processor_handle.clone(),
         appsec_processor.clone(),
+        &shared_client,
     );
 
     let api_runtime_proxy_shutdown_signal = start_api_runtime_proxy(
@@ -1020,6 +1033,7 @@ fn start_logs_agent(
     tags_provider: &Arc<TagProvider>,
     event_bus: Sender<Event>,
     is_managed_instance_mode: bool,
+    client: &Client,
 ) -> (
     Sender<TelemetryEvent>,
     LogsFlusher,
@@ -1048,7 +1062,12 @@ fn start_logs_agent(
         drop(agent);
     });
 
-    let flusher = LogsFlusher::new(api_key_factory, aggregator_handle.clone(), config.clone());
+    let flusher = LogsFlusher::new(
+        api_key_factory,
+        aggregator_handle.clone(),
+        config.clone(),
+        client.clone(),
+    );
     (tx, flusher, cancel_token, aggregator_handle)
 }
 
@@ -1059,6 +1078,7 @@ fn start_trace_agent(
     tags_provider: &Arc<TagProvider>,
     invocation_processor_handle: InvocationProcessorHandle,
     appsec_processor: Option<Arc<TokioMutex<AppSecProcessor>>>,
+    client: &Client,
 ) -> (
     Sender<SendDataBuilderInfo>,
     Arc<trace_flusher::TraceFlusher>,
@@ -1069,6 +1089,14 @@ fn start_trace_agent(
     StatsConcentratorHandle,
     TraceAggregatorHandle,
 ) {
+    // Build one shared hyper-based HTTP client for trace and stats flushing.
+    // This client type is required by libdd_trace_utils for SendData::send().
+    let trace_http_client = trace_http_client::create_client(
+        config.proxy_https.as_ref(),
+        config.tls_cert_file.as_ref(),
+    )
+    .expect("Failed to create trace HTTP client");
+
     // Stats
     let (stats_concentrator_service, stats_concentrator_handle) =
         StatsConcentratorService::new(Arc::clone(config));
@@ -1080,6 +1108,7 @@ fn start_trace_agent(
         api_key_factory.clone(),
         stats_aggregator.clone(),
         Arc::clone(config),
+        trace_http_client.clone(),
     ));
 
     let stats_processor = Arc::new(stats_processor::ServerlessStatsProcessor {});
@@ -1092,6 +1121,7 @@ fn start_trace_agent(
         trace_aggregator_handle.clone(),
         config.clone(),
         api_key_factory.clone(),
+        trace_http_client,
     ));
 
     let obfuscation_config = obfuscation_config::ObfuscationConfig {
@@ -1117,6 +1147,7 @@ fn start_trace_agent(
         Arc::clone(&proxy_aggregator),
         Arc::clone(tags_provider),
         Arc::clone(config),
+        client.clone(),
     ));
 
     let trace_agent = trace_agent::TraceAgent::new(
@@ -1157,6 +1188,7 @@ async fn start_dogstatsd(
     tags_provider: Arc<TagProvider>,
     api_key_factory: Arc<ApiKeyFactory>,
     config: &Arc<Config>,
+    client: &Client,
 ) -> (
     Arc<Vec<MetricsFlusher>>,
     MetricsAggregatorHandle,
@@ -1180,12 +1212,11 @@ async fn start_dogstatsd(
     });
 
     // Get flushers with aggregator handle
-    let metrics_client = bottlecap::http::get_client(config);
     let flushers = Arc::new(start_metrics_flushers(
         Arc::clone(&api_key_factory),
         &aggregator_handle,
         config,
-        &metrics_client,
+        client,
     ));
 
     // Create Dogstatsd server

bottlecap/src/logs/flusher.rs

@@ -1,6 +1,5 @@
 use crate::FLUSH_RETRY_COUNT;
 use crate::config;
-use crate::http::get_client;
 use crate::logs::aggregator_service::AggregatorHandle;
 use dogstatsd::api_key::ApiKeyFactory;
 use futures::future::join_all;
@@ -36,8 +35,8 @@ impl Flusher {
         api_key_factory: Arc<ApiKeyFactory>,
         endpoint: String,
         config: Arc<config::Config>,
+        client: reqwest::Client,
     ) -> Self {
-        let client = get_client(&config);
         Flusher {
             client,
             endpoint,
@@ -199,6 +198,7 @@ impl LogsFlusher {
         api_key_factory: Arc<ApiKeyFactory>,
         aggregator_handle: AggregatorHandle,
         config: Arc<config::Config>,
+        client: reqwest::Client,
     ) -> Self {
         let mut flushers = Vec::new();
 
@@ -216,6 +216,7 @@ impl LogsFlusher {
             Arc::clone(&api_key_factory),
             endpoint,
             config.clone(),
+            client.clone(),
         ));
 
         // Create flushers for additional endpoints
@@ -227,6 +228,7 @@ impl LogsFlusher {
                 additional_api_key_factory,
                 endpoint_url,
                 config.clone(),
+                client.clone(),
             ));
         }
 

bottlecap/src/traces/proxy_flusher.rs

@@ -9,7 +9,6 @@ use tracing::{debug, error};
 
 use crate::{
     FLUSH_RETRY_COUNT, config,
-    http::get_client,
     tags::provider,
     traces::{
         DD_ADDITIONAL_TAGS_HEADER,
@@ -39,9 +38,8 @@ impl Flusher {
         aggregator: Arc<Mutex<Aggregator>>,
         tags_provider: Arc<provider::Provider>,
         config: Arc<config::Config>,
+        client: reqwest::Client,
     ) -> Self {
-        let client = get_client(&config);
-
         Flusher {
             client,
             aggregator,

bottlecap/src/traces/stats_flusher.rs

@@ -8,7 +8,7 @@ use tokio::sync::OnceCell;
 
 use crate::config;
 use crate::lifecycle::invocation::processor::S_TO_MS;
-use crate::traces::http_client::{self, HttpClient};
+use crate::traces::http_client::HttpClient;
 use crate::traces::stats_aggregator::StatsAggregator;
 use dogstatsd::api_key::ApiKeyFactory;
 use libdd_common::Endpoint;
@@ -21,10 +21,7 @@ pub struct StatsFlusher {
     config: Arc<config::Config>,
     api_key_factory: Arc<ApiKeyFactory>,
     endpoint: OnceCell<Endpoint>,
-    /// Cached HTTP client, lazily initialized on first use.
-    /// TODO: `StatsFlusher` and `TraceFlusher` both hit trace.agent.datadoghq.{site} and could
-    /// share a single HTTP client for better connection pooling.
-    http_client: OnceCell<HttpClient>,
+    http_client: HttpClient,
 }
 
 impl StatsFlusher {
@@ -33,13 +30,14 @@ impl StatsFlusher {
         api_key_factory: Arc<ApiKeyFactory>,
         aggregator: Arc<Mutex<StatsAggregator>>,
         config: Arc<config::Config>,
+        http_client: HttpClient,
     ) -> Self {
         StatsFlusher {
             aggregator,
             config,
             api_key_factory,
             endpoint: OnceCell::new(),
-            http_client: OnceCell::new(),
+            http_client,
         }
     }
 
@@ -97,18 +95,11 @@ impl StatsFlusher {
 
         let start = std::time::Instant::now();
 
-        // Get or create the cached HTTP client
-        let http_client = self.get_or_init_http_client().await;
-        let Some(http_client) = http_client else {
-            error!("STATS | Failed to create HTTP client, will retry");
-            return Some(stats);
-        };
-
         let resp = stats_utils::send_stats_payload_with_client(
             serialized_stats_payload,
             endpoint,
             api_key.as_str(),
-            Some(http_client),
+            Some(&self.http_client),
         )
         .await;
         let elapsed = start.elapsed();
@@ -170,29 +161,4 @@ impl StatsFlusher {
             Some(all_failed)
         }
     }
-    /// Returns a reference to the cached HTTP client, initializing it if necessary.
-    ///
-    /// The client is created once and reused for all subsequent flushes,
-    /// providing connection pooling and TLS session reuse.
-    ///
-    /// Returns `None` if client creation fails. The error is logged but not cached,
-    /// allowing retry on subsequent calls.
-    async fn get_or_init_http_client(&self) -> Option<&HttpClient> {
-        match self
-            .http_client
-            .get_or_try_init(|| async {
-                http_client::create_client(
-                    self.config.proxy_https.as_ref(),
-                    self.config.tls_cert_file.as_ref(),
-                )
-            })
-            .await
-        {
-            Ok(client) => Some(client),
-            Err(e) => {
-                error!("STATS_FLUSHER | Failed to create HTTP client: {e}");
-                None
-            }
-        }
-    }
 }

bottlecap/src/traces/trace_flusher.rs

@@ -11,13 +11,12 @@ use libdd_trace_utils::{
 };
 use std::str::FromStr;
 use std::sync::Arc;
-use tokio::sync::OnceCell;
 use tokio::task::JoinSet;
 use tracing::{debug, error};
 
 use crate::config::Config;
 use crate::lifecycle::invocation::processor::S_TO_MS;
-use crate::traces::http_client::{self, HttpClient};
+use crate::traces::http_client::HttpClient;
 use crate::traces::trace_aggregator_service::AggregatorHandle;
 
 pub struct TraceFlusher {
@@ -28,10 +27,7 @@ pub struct TraceFlusher {
     /// Configured via `DD_APM_ADDITIONAL_ENDPOINTS` (e.g., sending to both US and EU).
     /// Each trace batch is sent to the primary endpoint AND all additional endpoints.
     pub additional_endpoints: Vec<Endpoint>,
-    /// Cached HTTP client, lazily initialized on first use.
-    /// TODO: `TraceFlusher` and `StatsFlusher` both hit trace.agent.datadoghq.{site} and could
-    /// share a single HTTP client for better connection pooling.
-    http_client: OnceCell<HttpClient>,
+    http_client: HttpClient,
 }
 
 impl TraceFlusher {
@@ -40,6 +36,7 @@ impl TraceFlusher {
         aggregator_handle: AggregatorHandle,
         config: Arc<Config>,
         api_key_factory: Arc<ApiKeyFactory>,
+        http_client: HttpClient,
     ) -> Self {
         // Parse additional endpoints for dual-shipping from config.
         // Format: { "https://trace.agent.datadoghq.eu": ["api-key-1", "api-key-2"], ... }
@@ -65,7 +62,7 @@ impl TraceFlusher {
             config,
             api_key_factory,
             additional_endpoints,
-            http_client: OnceCell::new(),
+            http_client,
         }
     }
 
@@ -83,11 +80,7 @@ impl TraceFlusher {
             return None;
         };
 
-        // Get or create the cached HTTP client
-        let Some(http_client) = self.get_or_init_http_client().await else {
-            error!("TRACES | Failed to create HTTP client, skipping flush");
-            return None;
-        };
+        let http_client = &self.http_client;
 
         let mut failed_batch: Vec<SendData> = Vec::new();
 
@@ -167,32 +160,6 @@ impl TraceFlusher {
         None
     }
 
-    /// Returns a clone of the cached HTTP client, initializing it if necessary.
-    ///
-    /// The client is created once and reused for all subsequent flushes,
-    /// providing connection pooling and TLS session reuse.
-    ///
-    /// Returns `None` if client creation fails. The error is logged but not cached,
-    /// allowing retry on subsequent calls.
-    async fn get_or_init_http_client(&self) -> Option<HttpClient> {
-        match self
-            .http_client
-            .get_or_try_init(|| async {
-                http_client::create_client(
-                    self.config.proxy_https.as_ref(),
-                    self.config.tls_cert_file.as_ref(),
-                )
-            })
-            .await
-        {
-            Ok(client) => Some(client.clone()),
-            Err(e) => {
-                error!("TRACES | Failed to create HTTP client: {e}");
-                None
-            }
-        }
-    }
-
     /// Sends traces to the Datadog intake endpoint using the provided HTTP client.
     ///
     /// Each `SendData` is sent to its own configured target endpoint.

bottlecap/tests/logs_integration_test.rs

@@ -70,7 +70,9 @@ async fn test_logs() {
         false,
     );
     let api_key_factory = Arc::new(ApiKeyFactory::new(dd_api_key));
-    let logs_flusher = LogsFlusher::new(api_key_factory, logs_aggr_handle, arc_conf.clone());
+    let client = bottlecap::http::get_client(&Arc::clone(&arc_conf));
+    let logs_flusher =
+        LogsFlusher::new(api_key_factory, logs_aggr_handle, arc_conf.clone(), client);
 
     let telemetry_events: Vec<TelemetryEvent> = serde_json::from_str(
         r#"[{"time":"2022-10-21T14:05:03.165Z","type":"platform.start","record":{"requestId":"459921b5-681c-4a96-beb0-81e0aa586026","version":"$LATEST","tracing":{"spanId":"24cd7d670fa455f0","type":"X-Amzn-Trace-Id","value":"Root=1-6352a70e-1e2c502e358361800241fd45;Parent=35465b3a9e2f7c6a;Sampled=1"}}}]"#)