chore(flushing): standardize code with refactoring on some flushers and retries (#1018)

## Overview

Simplify code for flushing, trying to standardize everything by avoiding
code all over the place, ensuring that we only create one client and we
can reuse as much as possible for performance improvements

## Motivation

[SVLS-8507](https://datadoghq.atlassian.net/browse/SVLS-8507)


[SVLS-8507]:
https://datadoghq.atlassian.net/browse/SVLS-8507?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

Author: duncanista

bottlecap/src/bin/bottlecap/main.rs

@@ -66,14 +66,14 @@ use bottlecap::{
         span_dedup_service,
         stats_aggregator::StatsAggregator,
         stats_concentrator_service::{StatsConcentratorHandle, StatsConcentratorService},
-        stats_flusher::{self, StatsFlusher},
+        stats_flusher,
         stats_generator::StatsGenerator,
         stats_processor, trace_agent,
         trace_aggregator::SendDataBuilderInfo,
         trace_aggregator_service::{
             AggregatorHandle as TraceAggregatorHandle, AggregatorService as TraceAggregatorService,
         },
-        trace_flusher::{self, TraceFlusher},
+        trace_flusher,
         trace_processor::{self, SendingTraceProcessor},
     },
 };
@@ -1081,9 +1081,9 @@ fn start_trace_agent(
     appsec_processor: Option<Arc<TokioMutex<AppSecProcessor>>>,
 ) -> (
     Sender<SendDataBuilderInfo>,
-    Arc<trace_flusher::ServerlessTraceFlusher>,
+    Arc<trace_flusher::TraceFlusher>,
     Arc<trace_processor::ServerlessTraceProcessor>,
-    Arc<stats_flusher::ServerlessStatsFlusher>,
+    Arc<stats_flusher::StatsFlusher>,
     Arc<ProxyFlusher>,
     tokio_util::sync::CancellationToken,
     StatsConcentratorHandle,
@@ -1096,7 +1096,7 @@ fn start_trace_agent(
     let stats_aggregator: Arc<TokioMutex<StatsAggregator>> = Arc::new(TokioMutex::new(
         StatsAggregator::new_with_concentrator(stats_concentrator_handle.clone()),
     ));
-    let stats_flusher = Arc::new(stats_flusher::ServerlessStatsFlusher::new(
+    let stats_flusher = Arc::new(stats_flusher::StatsFlusher::new(
         api_key_factory.clone(),
         stats_aggregator.clone(),
         Arc::clone(config),
@@ -1108,7 +1108,7 @@ fn start_trace_agent(
     let (trace_aggregator_service, trace_aggregator_handle) = TraceAggregatorService::default();
     tokio::spawn(trace_aggregator_service.run());
 
-    let trace_flusher = Arc::new(trace_flusher::ServerlessTraceFlusher::new(
+    let trace_flusher = Arc::new(trace_flusher::TraceFlusher::new(
         trace_aggregator_handle.clone(),
         config.clone(),
         api_key_factory.clone(),
@@ -1394,6 +1394,7 @@ mod flush_handles_tests {
         let mut handles = FlushHandles::new();
         let handle = tokio::spawn(async {
             sleep(Duration::from_millis(5)).await;
+            Vec::new() // Return empty Vec for stats retry
         });
         handles.stats_flush_handles.push(handle);
 

bottlecap/src/flushing/handles.rs

@@ -2,6 +2,7 @@
 
 use datadog_protos::metrics::SketchPayload;
 use dogstatsd::datadog::Series;
+use libdd_trace_protobuf::pb;
 use libdd_trace_utils::send_data::SendData;
 use tokio::task::JoinHandle;
 
@@ -32,8 +33,8 @@ pub struct FlushHandles {
     pub metric_flush_handles: Vec<JoinHandle<MetricsRetryBatch>>,
     /// Handles for proxy flush operations. Returns failed request builders for retry.
     pub proxy_flush_handles: Vec<JoinHandle<Vec<reqwest::RequestBuilder>>>,
-    /// Handles for stats flush operations. Stats don't support retry.
-    pub stats_flush_handles: Vec<JoinHandle<()>>,
+    /// Handles for stats flush operations. Returns failed stats payloads for retry.
+    pub stats_flush_handles: Vec<JoinHandle<Vec<pb::ClientStatsPayload>>>,
 }
 
 impl FlushHandles {

bottlecap/src/flushing/service.rs

@@ -23,20 +23,11 @@ use crate::traces::{
 /// - Spawning non-blocking flush tasks
 /// - Awaiting pending flush handles with retry logic
 /// - Performing blocking flushes (spawn + await)
-///
-/// # Type Parameters
-///
-/// * `TF` - Trace flusher type implementing `TraceFlusher`
-/// * `SF` - Stats flusher type implementing `StatsFlusher`
-pub struct FlushingService<TF, SF>
-where
-    TF: TraceFlusher + Send + Sync + 'static,
-    SF: StatsFlusher + Send + Sync + 'static,
-{
+pub struct FlushingService {
     // Flushers
     logs_flusher: LogsFlusher,
-    trace_flusher: Arc<TF>,
-    stats_flusher: Arc<SF>,
+    trace_flusher: Arc<TraceFlusher>,
+    stats_flusher: Arc<StatsFlusher>,
     proxy_flusher: Arc<ProxyFlusher>,
     metrics_flushers: Arc<TokioMutex<Vec<MetricsFlusher>>>,
 
@@ -47,17 +38,13 @@ where
     handles: FlushHandles,
 }
 
-impl<TF, SF> FlushingService<TF, SF>
-where
-    TF: TraceFlusher + Send + Sync + 'static,
-    SF: StatsFlusher + Send + Sync + 'static,
-{
+impl FlushingService {
     /// Creates a new `FlushingService` with the given flushers.
     #[must_use]
     pub fn new(
         logs_flusher: LogsFlusher,
-        trace_flusher: Arc<TF>,
-        stats_flusher: Arc<SF>,
+        trace_flusher: Arc<TraceFlusher>,
+        stats_flusher: Arc<StatsFlusher>,
         proxy_flusher: Arc<ProxyFlusher>,
         metrics_flushers: Arc<TokioMutex<Vec<MetricsFlusher>>>,
         metrics_aggr_handle: MetricsAggregatorHandle,
@@ -135,11 +122,13 @@ where
             self.handles.metric_flush_handles.push(handle);
         }
 
-        // Spawn stats flush (fire-and-forget, no retry)
+        // Spawn stats flush
         let sf = Arc::clone(&self.stats_flusher);
         self.handles
             .stats_flush_handles
-            .push(tokio::spawn(async move { sf.flush(false).await }));
+            .push(tokio::spawn(async move {
+                sf.flush(false, None).await.unwrap_or_default()
+            }));
 
         // Spawn proxy flush
         let pf = self.proxy_flusher.clone();
@@ -166,11 +155,25 @@ where
         let mut joinset = tokio::task::JoinSet::new();
         let mut flush_error = false;
 
-        // Await stats handles (no retry)
+        // Await stats handles with retry
         for handle in self.handles.stats_flush_handles.drain(..) {
-            if let Err(e) = handle.await {
-                error!("FLUSHING_SERVICE | stats flush error {e:?}");
-                flush_error = true;
+            match handle.await {
+                Ok(retry) => {
+                    let sf = self.stats_flusher.clone();
+                    if !retry.is_empty() {
+                        debug!(
+                            "FLUSHING_SERVICE | redriving {:?} stats payloads",
+                            retry.len()
+                        );
+                        joinset.spawn(async move {
+                            sf.flush(false, Some(retry)).await;
+                        });
+                    }
+                }
+                Err(e) => {
+                    error!("FLUSHING_SERVICE | stats flush error {e:?}");
+                    flush_error = true;
+                }
             }
         }
 
@@ -325,7 +328,7 @@ where
             self.logs_flusher.flush(None),
             futures::future::join_all(metrics_futures),
             self.trace_flusher.flush(None),
-            self.stats_flusher.flush(force_stats),
+            self.stats_flusher.flush(force_stats, None),
             self.proxy_flusher.flush(None),
         );
     }
@@ -340,11 +343,7 @@ where
     }
 }
 
-impl<TF, SF> std::fmt::Debug for FlushingService<TF, SF>
-where
-    TF: TraceFlusher + Send + Sync + 'static,
-    SF: StatsFlusher + Send + Sync + 'static,
-{
+impl std::fmt::Debug for FlushingService {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("FlushingService")
             .field("handles", &self.handles)

bottlecap/src/traces/hyper_client.rs

@@ -0,0 +1,113 @@
+// Copyright 2023-Present Datadog, Inc. https://www.datadoghq.com/
+// SPDX-License-Identifier: Apache-2.0
+
+//! Hyper-based HTTP client for trace and stats flushers.
+//!
+//! This module provides the HTTP client type required by `libdd_trace_utils`
+//! for sending traces and stats to Datadog intake endpoints.
+
+use hyper_http_proxy;
+use hyper_rustls::HttpsConnectorBuilder;
+use libdd_common::{GenericHttpClient, hyper_migration};
+use rustls::RootCertStore;
+use rustls_pki_types::CertificateDer;
+use std::error::Error;
+use std::fs::File;
+use std::io::BufReader;
+use std::sync::LazyLock;
+use tracing::debug;
+
+/// Type alias for the HTTP client used by trace and stats flushers.
+///
+/// This is the client type expected by `libdd_trace_utils::SendData::send()`.
+pub type HyperClient =
+    GenericHttpClient<hyper_http_proxy::ProxyConnector<libdd_common::connector::Connector>>;
+
+/// Initialize the crypto provider needed for setting custom root certificates.
+fn ensure_crypto_provider_initialized() {
+    static INIT_CRYPTO_PROVIDER: LazyLock<()> = LazyLock::new(|| {
+        #[cfg(unix)]
+        rustls::crypto::aws_lc_rs::default_provider()
+            .install_default()
+            .expect("Failed to install default CryptoProvider");
+    });
+
+    let () = &*INIT_CRYPTO_PROVIDER;
+}
+
+/// Creates a new hyper-based HTTP client with the given configuration.
+///
+/// This client is compatible with `libdd_trace_utils` and supports:
+/// - HTTPS proxy configuration
+/// - Custom TLS root certificates
+///
+/// # Arguments
+///
+/// * `proxy_https` - Optional HTTPS proxy URL
+/// * `tls_cert_file` - Optional path to a PEM file containing root certificates
+///
+/// # Errors
+///
+/// Returns an error if:
+/// - The proxy URL cannot be parsed
+/// - The TLS certificate file cannot be read or parsed
+pub fn create_client(
+    proxy_https: Option<&String>,
+    tls_cert_file: Option<&String>,
+) -> Result<HyperClient, Box<dyn Error>> {
+    // Create the base connector with optional custom TLS config
+    let connector = if let Some(ca_cert_path) = tls_cert_file {
+        // Ensure crypto provider is initialized before creating TLS config
+        ensure_crypto_provider_initialized();
+
+        // Load the custom certificate
+        let cert_file = File::open(ca_cert_path)?;
+        let mut reader = BufReader::new(cert_file);
+        let certs: Vec<CertificateDer> =
+            rustls_pemfile::certs(&mut reader).collect::<Result<Vec<_>, _>>()?;
+
+        // Create a root certificate store and add custom certs
+        let mut root_store = RootCertStore::empty();
+        for cert in certs {
+            root_store.add(cert)?;
+        }
+
+        // Build the TLS config with custom root certificates
+        let tls_config = rustls::ClientConfig::builder()
+            .with_root_certificates(root_store)
+            .with_no_client_auth();
+
+        // Build the HTTPS connector with custom config
+        let https_connector = HttpsConnectorBuilder::new()
+            .with_tls_config(tls_config)
+            .https_or_http()
+            .enable_http1()
+            .build();
+
+        debug!(
+            "HYPER_CLIENT | Added root certificate from {}",
+            ca_cert_path
+        );
+
+        // Construct the Connector::Https variant directly
+        libdd_common::connector::Connector::Https(https_connector)
+    } else {
+        // Use default connector
+        libdd_common::connector::Connector::default()
+    };
+
+    if let Some(proxy) = proxy_https {
+        let proxy =
+            hyper_http_proxy::Proxy::new(hyper_http_proxy::Intercept::Https, proxy.parse()?);
+        let proxy_connector = hyper_http_proxy::ProxyConnector::from_proxy(connector, proxy)?;
+        let client = hyper_migration::client_builder().build(proxy_connector);
+        debug!(
+            "HYPER_CLIENT | Proxy connector created with proxy: {:?}",
+            proxy_https
+        );
+        Ok(client)
+    } else {
+        let proxy_connector = hyper_http_proxy::ProxyConnector::new(connector)?;
+        Ok(hyper_migration::client_builder().build(proxy_connector))
+    }
+}

bottlecap/src/traces/mod.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub mod context;
+pub mod hyper_client;
 pub mod propagation;
 pub mod proxy_aggregator;
 pub mod proxy_flusher;