chore(dogstatsd): use pre built client (#1061)

## Overview

Upgrades dogstatsd which allows us to use our own reqwest client instead
of expecting the server to create one

This gives us control over how we are building a specialized client
which might need a proxy, certificates, and more.

## Testing

Unit tests, integration tests, and self monitoring

Author: duncanista

bottlecap/Cargo.lock

@@ -76,8 +76,8 @@ libdd-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "c812
 libdd-trace-normalization = { git = "https://github.com/DataDog/libdatadog",  rev = "c8121f422d2c8d219f8d421ff3cdb1fcbe9e8b09" }
 libdd-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "c8121f422d2c8d219f8d421ff3cdb1fcbe9e8b09"  }
 libdd-trace-stats = { git = "https://github.com/DataDog/libdatadog", rev = "c8121f422d2c8d219f8d421ff3cdb1fcbe9e8b09"  }
-dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "2eb009a59ed07ffcaf174db1c31af413365e9bc6", default-features = false }
-datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "2eb009a59ed07ffcaf174db1c31af413365e9bc6", default-features = false }
+dogstatsd = { git = "https://github.com/DataDog/serverless-components", rev = "28f796bf767fff56caf08153ade5cd80c8e8f705", default-features = false }
+datadog-fips = { git = "https://github.com/DataDog/serverless-components", rev = "28f796bf767fff56caf08153ade5cd80c8e8f705", default-features = false }
 libddwaf = { version = "1.28.1", git = "https://github.com/DataDog/libddwaf-rust", rev = "d1534a158d976bd4f747bf9fcc58e0712d2d17fc", default-features = false, features = ["serde"] }
 
 [dev-dependencies]
@@ -118,11 +118,10 @@ opt-level = 3
 tikv-jemallocator = "0.5"
 
 [features]
-default = ["reqwest/rustls-tls-native-roots", "dogstatsd/default", "datadog-fips/default" ]
+default = ["reqwest/rustls-tls-native-roots", "datadog-fips/default" ]
 fips = [
     "libdd-common/fips",
     "libdd-trace-utils/fips",
-    "dogstatsd/fips",
     "datadog-fips/fips",
     "libddwaf/fips",
     "reqwest/rustls-tls-native-roots-no-provider",

bottlecap/Cargo.toml

@@ -80,8 +80,9 @@ use bottlecap::{
 use datadog_fips::reqwest_adapter::create_reqwest_client_builder;
 use decrypt::resolve_secrets;
 use dogstatsd::{
-    aggregator_service::AggregatorHandle as MetricsAggregatorHandle,
-    aggregator_service::AggregatorService as MetricsAggregatorService,
+    aggregator::{
+        AggregatorHandle as MetricsAggregatorHandle, AggregatorService as MetricsAggregatorService,
+    },
     api_key::ApiKeyFactory,
     constants::CONTEXTS,
     datadog::{
@@ -95,7 +96,7 @@ use dogstatsd::{
 use libdd_trace_obfuscation::obfuscation_config;
 use reqwest::Client;
 use std::{collections::hash_map, env, path::Path, str::FromStr, sync::Arc};
-use tokio::time::{Duration, Instant};
+use tokio::time::Instant;
 use tokio::{sync::Mutex as TokioMutex, sync::mpsc::Sender};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, warn};
@@ -1179,10 +1180,12 @@ 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,
     ));
 
     // Create Dogstatsd server
@@ -1214,6 +1217,7 @@ fn start_metrics_flushers(
     api_key_factory: Arc<ApiKeyFactory>,
     metrics_aggr_handle: &MetricsAggregatorHandle,
     config: &Arc<Config>,
+    client: &Client,
 ) -> Vec<MetricsFlusher> {
     let mut flushers = Vec::new();
 
@@ -1237,9 +1241,7 @@ fn start_metrics_flushers(
         api_key_factory,
         aggregator_handle: metrics_aggr_handle.clone(),
         metrics_intake_url_prefix: metrics_intake_url.expect("can't parse site or override"),
-        https_proxy: config.proxy_https.clone(),
-        ca_cert_path: config.tls_cert_file.clone(),
-        timeout: Duration::from_secs(config.flush_timeout),
+        client: client.clone(),
         retry_strategy: DsdRetryStrategy::Immediate(3),
         compression_level: config.metrics_config_compression_level,
     };
@@ -1267,9 +1269,7 @@ fn start_metrics_flushers(
                 api_key_factory: additional_api_key_factory,
                 aggregator_handle: metrics_aggr_handle.clone(),
                 metrics_intake_url_prefix: metrics_intake_url.clone(),
-                https_proxy: config.proxy_https.clone(),
-                ca_cert_path: config.tls_cert_file.clone(),
-                timeout: Duration::from_secs(config.flush_timeout),
+                client: client.clone(),
                 retry_strategy: DsdRetryStrategy::Immediate(3),
                 compression_level: config.metrics_config_compression_level,
             };

bottlecap/src/bin/bottlecap/main.rs

@@ -5,8 +5,7 @@ use std::sync::Arc;
 use tracing::{debug, error};
 
 use dogstatsd::{
-    aggregator_service::AggregatorHandle as MetricsAggregatorHandle,
-    flusher::Flusher as MetricsFlusher,
+    aggregator::AggregatorHandle as MetricsAggregatorHandle, flusher::Flusher as MetricsFlusher,
 };
 
 use crate::flushing::handles::{FlushHandles, MetricsRetryBatch};

bottlecap/src/flushing/service.rs

@@ -96,7 +96,7 @@ impl Processor {
         tags_provider: Arc<provider::Provider>,
         config: Arc<config::Config>,
         aws_config: Arc<AwsConfig>,
-        metrics_aggregator: dogstatsd::aggregator_service::AggregatorHandle,
+        metrics_aggregator: dogstatsd::aggregator::AggregatorHandle,
         propagator: Arc<DatadogCompositePropagator>,
     ) -> Self {
         let resource = tags_provider
@@ -1346,7 +1346,7 @@ mod tests {
     use crate::traces::stats_generator::StatsGenerator;
     use crate::traces::trace_processor;
     use base64::{Engine, engine::general_purpose::STANDARD};
-    use dogstatsd::aggregator_service::AggregatorService;
+    use dogstatsd::aggregator::AggregatorService;
     use dogstatsd::metric::EMPTY_TAGS;
     use serde_json::json;
 

bottlecap/src/lifecycle/invocation/processor.rs

@@ -1,7 +1,7 @@
 use std::{collections::HashMap, sync::Arc};
 
 use chrono::{DateTime, Utc};
-use dogstatsd::aggregator_service::AggregatorHandle;
+use dogstatsd::aggregator::AggregatorHandle;
 use libdd_trace_protobuf::pb::Span;
 use serde_json::Value;
 use thiserror::Error;

bottlecap/src/lifecycle/invocation/processor_service.rs

@@ -5,7 +5,7 @@ use crate::metrics::enhanced::usage_metrics::{EnhancedMetricsHandle, EnhancedMet
 use crate::proc::{self, CPUData, NetworkData};
 use dogstatsd::metric::SortedTags;
 use dogstatsd::metric::{Metric, MetricValue};
-use dogstatsd::{aggregator_service::AggregatorHandle, metric};
+use dogstatsd::{aggregator::AggregatorHandle, metric};
 use std::collections::HashMap;
 use std::env::consts::ARCH;
 use std::sync::Arc;
@@ -809,7 +809,7 @@ mod tests {
     use super::*;
     use crate::config;
     use crate::extension::telemetry::events::{OnDemandReportMetrics, ReportMetrics};
-    use dogstatsd::aggregator_service::AggregatorService;
+    use dogstatsd::aggregator::AggregatorService;
     use dogstatsd::metric::EMPTY_TAGS;
     const PRECISION: f64 = 0.000_000_01;
 

bottlecap/src/metrics/enhanced/lambda.rs

@@ -436,7 +436,7 @@ mod tests {
     use std::{collections::HashMap, time::Duration};
     use tokio::{sync::Mutex as TokioMutex, time::Instant};
 
-    use dogstatsd::{aggregator_service::AggregatorService, metric::EMPTY_TAGS};
+    use dogstatsd::{aggregator::AggregatorService, metric::EMPTY_TAGS};
     use http_body_util::Full;
     use hyper::{server::conn::http1, service::service_fn};
     use hyper_util::rt::TokioIo;

bottlecap/src/proxy/interceptor.rs

@@ -1,6 +1,6 @@
 use bottlecap::config::Config;
 use bottlecap::metrics::enhanced::lambda::Lambda as enhanced_metrics;
-use dogstatsd::aggregator_service::AggregatorService;
+use dogstatsd::aggregator::AggregatorService;
 use dogstatsd::api_key::ApiKeyFactory;
 use dogstatsd::datadog::{DdDdUrl, MetricsIntakeUrlPrefix, MetricsIntakeUrlPrefixOverride};
 use dogstatsd::flusher::Flusher as MetricsFlusher;
@@ -51,9 +51,10 @@ async fn test_enhanced_metrics() {
         aggregator_handle: metrics_aggr_handle.clone(),
         metrics_intake_url_prefix: MetricsIntakeUrlPrefix::new(None, Some(metrics_site_override))
             .expect("can't parse metrics intake URL from site"),
-        https_proxy: None,
-        ca_cert_path: None,
-        timeout: std::time::Duration::from_secs(5),
+        client: datadog_fips::reqwest_adapter::create_reqwest_client_builder()
+            .expect("failed to create client builder")
+            .build()
+            .expect("failed to build client"),
         retry_strategy: dogstatsd::datadog::RetryStrategy::Immediate(1),
         compression_level: 6,
     };