Refactor data-pipeline to hold a client pool so the connection with the agent is reused.

Author: hoolioh

data-pipeline/src/stats_exporter.rs

@@ -14,7 +14,8 @@ use std::{
 use crate::{span_concentrator::SpanConcentrator, trace_exporter::TracerMetadata};
 use datadog_trace_protobuf::pb;
 use datadog_trace_utils::send_with_retry::{send_with_retry, RetryStrategy};
-use ddcommon::{worker::Worker, Endpoint};
+use ddcommon::{worker::Worker, Endpoint, HttpClient};
+use ddcommon::hyper_migration::new_default_client;
 use hyper;
 use tokio::select;
 use tokio_util::sync::CancellationToken;
@@ -31,6 +32,7 @@ pub struct StatsExporter {
     meta: TracerMetadata,
     sequence_id: AtomicU64,
     cancellation_token: CancellationToken,
+    client: HttpClient,
 }
 
 impl StatsExporter {
@@ -56,6 +58,7 @@ impl StatsExporter {
             meta,
             sequence_id: AtomicU64::new(0),
             cancellation_token,
+            client: new_default_client(),
         }
     }
 
@@ -89,11 +92,11 @@ impl StatsExporter {
         );
 
         let result = send_with_retry(
+            &self.client,
             &self.endpoint,
             body,
             &headers,
             &RetryStrategy::default(),
-            None,
         )
         .await;
 

data-pipeline/src/trace_exporter/builder.rs

@@ -12,6 +12,7 @@ use crate::trace_exporter::{
     INFO_ENDPOINT,
 };
 use arc_swap::ArcSwap;
+use ddcommon::hyper_migration::new_default_client;
 use ddcommon::{parse_uri, tag, Endpoint};
 use dogstatsd_client::new;
 use std::sync::{Arc, Mutex};
@@ -343,6 +344,7 @@ impl TraceExporterBuilder {
             agent_payload_response_version: self
                 .agent_rates_payload_version_enabled
                 .then(AgentResponsePayloadVersion::new),
+            http_client: new_default_client(),
         })
     }
 

data-pipeline/src/trace_exporter/mod.rs

@@ -36,7 +36,7 @@ use datadog_trace_utils::send_with_retry::{
 };
 use datadog_trace_utils::span::{Span, SpanText};
 use datadog_trace_utils::trace_utils::TracerHeaderTags;
-use ddcommon::MutexExt;
+use ddcommon::{HttpClient, MutexExt};
 use ddcommon::{hyper_migration, Endpoint};
 use ddcommon::{tag, tag::Tag};
 use ddtelemetry::worker::TelemetryWorker;
@@ -208,6 +208,7 @@ pub struct TraceExporter {
     health_metrics_enabled: bool,
     workers: Arc<Mutex<TraceExporterWorkers>>,
     agent_payload_response_version: Option<AgentResponsePayloadVersion>,
+    http_client: HttpClient,
 }
 
 impl TraceExporter {
@@ -627,7 +628,7 @@ impl TraceExporter {
         let payload_len = mp_payload.len();
 
         // Send traces to the agent
-        let result = send_with_retry(endpoint, mp_payload, &headers, &strategy, None).await;
+        let result = send_with_retry(&self.http_client, endpoint, mp_payload, &headers, &strategy).await;
 
         // Emit http.requests health metric based on number of attempts
         let requests_count = match &result {