feat: [SVLS-5992] support DD_URL and DD_DD_URL for metrics intake

Author: apiarian-datadog

bottlecap/Cargo.lock

@@ -9,13 +9,13 @@ async-trait = { version = "0.1", default-features = false }
 chrono = { version = "0.4", features = ["serde", "std", "now"], default-features = false }
 datadog-protos = { version = "0.1.0", default-features = false, git = "https://github.com/DataDog/saluki/" }
 ddsketch-agent = { version = "0.1.0", default-features = false, git = "https://github.com/DataDog/saluki/" }
-ddcommon = { git = "https://github.com/DataDog/libdatadog", rev = "dbc8869c6de4ccd78a770524d250420f397fc7b3" }
-datadog-trace-protobuf = { git = "https://github.com/DataDog/libdatadog", rev = "dbc8869c6de4ccd78a770524d250420f397fc7b3" }
-datadog-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "dbc8869c6de4ccd78a770524d250420f397fc7b3" }
-datadog-trace-mini-agent = { git = "https://github.com/DataDog/libdatadog", rev = "dbc8869c6de4ccd78a770524d250420f397fc7b3" }
-datadog-trace-normalization = { git = "https://github.com/DataDog/libdatadog", rev = "dbc8869c6de4ccd78a770524d250420f397fc7b3" }
-datadog-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "dbc8869c6de4ccd78a770524d250420f397fc7b3" }
-dogstatsd = { git = "https://github.com/DataDog/libdatadog", rev = "dbc8869c6de4ccd78a770524d250420f397fc7b3" }
+ddcommon = { git = "https://github.com/DataDog/libdatadog", rev = "6e2d4462873e35e49a2c9befde958d96c4ed9de4" }
+datadog-trace-protobuf = { git = "https://github.com/DataDog/libdatadog", rev = "6e2d4462873e35e49a2c9befde958d96c4ed9de4" }
+datadog-trace-utils = { git = "https://github.com/DataDog/libdatadog", rev = "6e2d4462873e35e49a2c9befde958d96c4ed9de4" }
+datadog-trace-mini-agent = { git = "https://github.com/DataDog/libdatadog", rev = "6e2d4462873e35e49a2c9befde958d96c4ed9de4" }
+datadog-trace-normalization = { git = "https://github.com/DataDog/libdatadog", rev = "6e2d4462873e35e49a2c9befde958d96c4ed9de4" }
+datadog-trace-obfuscation = { git = "https://github.com/DataDog/libdatadog", rev = "6e2d4462873e35e49a2c9befde958d96c4ed9de4" }
+dogstatsd = { git = "https://github.com/DataDog/libdatadog", rev = "6e2d4462873e35e49a2c9befde958d96c4ed9de4" }
 figment = { version = "0.10", default-features = false, features = ["yaml", "env"] }
 hyper = { version = "0.14", default-features = false, features = ["server"] }
 lazy_static = { version = "1.5", default-features = false }

bottlecap/Cargo.toml

@@ -46,8 +46,9 @@ use decrypt::resolve_secrets;
 use dogstatsd::{
     aggregator::Aggregator as MetricsAggregator,
     constants::CONTEXTS,
+    datadog::{DdDdUrl, DdUrl, MetricsIntakeUrlPrefix, MetricsIntakeUrlPrefixOverride, Site},
     dogstatsd::{DogStatsD, DogStatsDConfig},
-    flusher::{build_fqdn_metrics, Flusher as MetricsFlusher},
+    flusher::{Flusher as MetricsFlusher, FlusherConfig as MetricsFlusherConfig},
     metric::{SortedTags, EMPTY_TAGS},
 };
 use reqwest::Client;
@@ -284,13 +285,27 @@ async fn extension_loop_active(
         )
         .expect("failed to create aggregator"),
     ));
-    let mut metrics_flusher = MetricsFlusher::new(
-        resolved_api_key.clone(),
-        Arc::clone(&metrics_aggr),
-        build_fqdn_metrics(config.site.clone()),
-        config.https_proxy.clone(),
-        Duration::from_secs(config.flush_timeout),
-    );
+    let metrics_intake_url_prefix = MetricsIntakeUrlPrefix::new(
+        Some(Site::new(config.site.clone()).expect("failed to create site")),
+        MetricsIntakeUrlPrefixOverride::maybe_new(
+            config
+                .url
+                .clone()
+                .map(|url| DdUrl::new(url).expect("failed to create dd url")),
+            config
+                .dd_url
+                .clone()
+                .map(|dd_url| DdDdUrl::new(dd_url).expect("failed to create dd dd url")),
+        ),
+    )
+    .expect("failed to create intake url prefix");
+    let mut metrics_flusher = MetricsFlusher::new(MetricsFlusherConfig {
+        api_key: resolved_api_key.clone(),
+        aggregator: Arc::clone(&metrics_aggr),
+        metrics_intake_url_prefix,
+        https_proxy: config.https_proxy.clone(),
+        timeout: Duration::from_secs(config.flush_timeout),
+    });
 
     let trace_flusher = Arc::new(trace_flusher::ServerlessTraceFlusher {
         buffer: Arc::new(TokioMutex::new(Vec::new())),

bottlecap/src/bin/bottlecap/main.rs

@@ -39,8 +39,6 @@ pub struct FallbackConfig {
     otlp_config_receiver_protocols_http_endpoint: Option<String>,
     otlp_config_receiver_protocols_grpc_endpoint: Option<String>,
     // intake urls
-    url: Option<String>,
-    dd_url: Option<String>,
     logs_config_logs_dd_url: Option<String>,
     // APM, as opposed to logs, does not use the `apm_config` prefix for env vars
     apm_dd_url: Option<String>,
@@ -107,6 +105,11 @@ pub struct Config {
     pub trace_propagation_style_extract: Vec<TracePropagationStyle>,
     pub trace_propagation_extract_first: bool,
     pub trace_propagation_http_baggage_enabled: bool,
+    // Intake URLs
+    // These two are for metrics intake. The official config is `DD_DD_URL` but `DD_URL` can also
+    // be used for backwards compatibility. `DD_DD_URL` takes precedence.
+    pub url: Option<String>,
+    pub dd_url: Option<String>,
 }
 
 impl Default for Config {
@@ -142,6 +145,9 @@ impl Default for Config {
             trace_propagation_style_extract: vec![],
             trace_propagation_extract_first: false,
             trace_propagation_http_baggage_enabled: false,
+            // Intake URLs
+            url: None,
+            dd_url: None,
         }
     }
 }
@@ -208,9 +214,7 @@ fn fallback(figment: &Figment, yaml_figment: &Figment) -> Result<(), ConfigError
     }
 
     // Intake URLs
-    if config.url.is_some()
-        || config.dd_url.is_some()
-        || config.logs_config_logs_dd_url.is_some()
+    if config.logs_config_logs_dd_url.is_some()
         || config.apm_dd_url.is_some()
         || yaml_config
             .logs_config

bottlecap/src/config/mod.rs

@@ -1,7 +1,8 @@
 use bottlecap::config::Config;
 use bottlecap::metrics::enhanced::lambda::Lambda as enhanced_metrics;
 use dogstatsd::aggregator::Aggregator as MetricsAggregator;
-use dogstatsd::flusher::Flusher as MetricsFlusher;
+use dogstatsd::datadog::{DdDdUrl, MetricsIntakeUrlPrefix, MetricsIntakeUrlPrefixOverride};
+use dogstatsd::flusher::{Flusher as MetricsFlusher, FlusherConfig as MetricsFlusherConfig};
 use dogstatsd::metric::SortedTags;
 use httpmock::prelude::*;
 use std::sync::{Arc, Mutex};
@@ -36,13 +37,20 @@ async fn test_enhanced_metrics() {
         MetricsAggregator::new(SortedTags::parse("aTagKey:aTagValue").unwrap(), 1024)
             .expect("failed to create aggregator"),
     ));
-    let mut metrics_flusher = MetricsFlusher::new(
-        dd_api_key.to_string(),
-        metrics_aggr.clone(),
-        server.base_url(),
-        None,
-        std::time::Duration::from_secs(5),
-    );
+    let mut metrics_flusher = MetricsFlusher::new(MetricsFlusherConfig {
+        api_key: dd_api_key.to_string(),
+        aggregator: metrics_aggr.clone(),
+        metrics_intake_url_prefix: MetricsIntakeUrlPrefix::new(
+            None,
+            MetricsIntakeUrlPrefixOverride::maybe_new(
+                None,
+                Some(DdDdUrl::new(server.base_url()).expect("failed to create dd url")),
+            ),
+        )
+        .expect("failed to create metrics intake url prefix"),
+        https_proxy: None,
+        timeout: std::time::Duration::from_secs(5),
+    });
     let lambda_enhanced_metrics =
         enhanced_metrics::new(Arc::clone(&metrics_aggr), Arc::clone(&arc_config));