enhancement(datadog service): Add datadog global config options

src/config/builder.rs

@@ -12,6 +12,12 @@ use crate::{enrichment_tables::EnrichmentTables, providers::Providers, secrets::
 
 #[cfg(feature = "api")]
 use super::api;
+#[cfg(any(
+    feature = "sinks-datadog_logs",
+    feature = "sinks-datadog_metrics",
+    feature = "sinks-datadog_traces",
+))]
+use super::datadog;
 #[cfg(feature = "enterprise")]
 use super::enterprise;
 use super::{
@@ -33,6 +39,15 @@ pub struct ConfigBuilder {
     #[serde(default)]
     pub api: api::Options,
 
+    #[cfg(any(
+        feature = "sinks-datadog_logs",
+        feature = "sinks-datadog_metrics",
+        feature = "sinks-datadog_traces",
+    ))]
+    #[configurable(derived)]
+    #[serde(default)]
+    pub datadog: datadog::Options,
+
     #[configurable(derived)]
     #[configurable(metadata(docs::hidden))]
     #[serde(default)]
@@ -92,6 +107,12 @@ struct ConfigBuilderHash<'a> {
     #[cfg(feature = "api")]
     api: &'a api::Options,
     schema: &'a schema::Options,
+    #[cfg(any(
+        feature = "sinks-datadog_logs",
+        feature = "sinks-datadog_metrics",
+        feature = "sinks-datadog_traces",
+    ))]
+    datadog: &'a datadog::Options,
     global: &'a GlobalOptions,
     healthchecks: &'a HealthcheckOptions,
     enrichment_tables: BTreeMap<&'a ComponentKey, &'a EnrichmentTableOuter>,
@@ -171,6 +192,12 @@ impl<'a> From<&'a ConfigBuilder> for ConfigBuilderHash<'a> {
             #[cfg(feature = "api")]
             api: &value.api,
             schema: &value.schema,
+            #[cfg(any(
+                feature = "sinks-datadog_logs",
+                feature = "sinks-datadog_metrics",
+                feature = "sinks-datadog_traces",
+            ))]
+            datadog: &value.datadog,
             global: &value.global,
             healthchecks: &value.healthchecks,
             enrichment_tables: value.enrichment_tables.iter().collect(),
@@ -190,6 +217,12 @@ impl From<Config> for ConfigBuilder {
             global,
             #[cfg(feature = "api")]
             api,
+            #[cfg(any(
+                feature = "sinks-datadog_logs",
+                feature = "sinks-datadog_metrics",
+                feature = "sinks-datadog_traces",
+            ))]
+            datadog,
             schema,
             #[cfg(feature = "enterprise")]
             enterprise,
@@ -220,6 +253,12 @@ impl From<Config> for ConfigBuilder {
             global,
             #[cfg(feature = "api")]
             api,
+            #[cfg(any(
+                feature = "sinks-datadog_logs",
+                feature = "sinks-datadog_metrics",
+                feature = "sinks-datadog_traces",
+            ))]
+            datadog,
             schema,
             #[cfg(feature = "enterprise")]
             enterprise,
@@ -434,6 +473,7 @@ mod tests {
         // reproducible across versions.
         let expected_keys = [
             "api",
+            "datadog",
             "enrichment_tables",
             "global",
             "healthchecks",
@@ -467,7 +507,7 @@ mod tests {
     /// the `ConfigBuilder` has changed what it serializes, or the implementation of `serde_json` has changed.
     /// If this test fails, we should ideally be able to fix so that the original hash passes!
     fn version_hash_match() {
-        let expected_hash = "6c98bea9d9e2f3133e2d39ba04592d17f96340a9bc4c8d697b09f5af388a76bd";
+        let expected_hash = "2c0ee2dfb8789e09dd953316fdef8b30b10fce3e259d44ca9461cf84ff63d12d";
         let builder = ConfigBuilder::default();
         let mut hash_builder = ConfigBuilderHash::from(&builder);
         hash_builder.version = "1.2.3".into();

src/config/compiler.rs

@@ -45,6 +45,12 @@ pub fn compile(mut builder: ConfigBuilder) -> Result<(Config, Vec<String>), Vec<
         global,
         #[cfg(feature = "api")]
         api,
+        #[cfg(any(
+            feature = "sinks-datadog_logs",
+            feature = "sinks-datadog_metrics",
+            feature = "sinks-datadog_traces",
+        ))]
+        datadog,
         schema,
         #[cfg(feature = "enterprise")]
         enterprise,
@@ -102,6 +108,12 @@ pub fn compile(mut builder: ConfigBuilder) -> Result<(Config, Vec<String>), Vec<
             #[cfg(feature = "api")]
             api,
             schema,
+            #[cfg(any(
+                feature = "sinks-datadog_logs",
+                feature = "sinks-datadog_metrics",
+                feature = "sinks-datadog_traces",
+            ))]
+            datadog,
             #[cfg(feature = "enterprise")]
             enterprise,
             hash,

src/config/datadog.rs

@@ -0,0 +1,38 @@
+//! Datadog global options.
+//! This is used to allow settings (api_key and site) to be specified in the configuration file
+//! globally which will apply to all datadog components. Each component can override the settings
+//! specified here if necessary.
+//!
+use vector_common::sensitive_string::SensitiveString;
+use vector_config::configurable_component;
+
+use crate::common::datadog::DD_US_SITE;
+
+/// Default settings to use for Datadog components.
+#[configurable_component]
+#[derive(Clone, Debug, Derivative)]
+#[derivative(Default)]
+#[serde(default, deny_unknown_fields)]
+pub struct Options {
+    /// Default Datadog API key to use for Datadog components.
+    #[serde(default = "default_api_key")]
+    #[derivative(Default(value = "default_api_key()"))]
+    #[configurable(metadata(docs::examples = "${DATADOG_API_KEY_ENV_VAR}"))]
+    #[configurable(metadata(docs::examples = "ef8d5de700e7989468166c40fc8a0ccd"))]
+    pub api_key: Option<SensitiveString>,
+
+    /// Default site to use for DataDog components.
+    #[serde(default = "default_site")]
+    #[derivative(Default(value = "default_site()"))]
+    #[configurable(metadata(docs::examples = "us3.datadoghq.com"))]
+    #[configurable(metadata(docs::examples = "datadoghq.eu"))]
+    pub site: String,
+}
+
+fn default_api_key() -> Option<SensitiveString> {
+    std::env::var("DD_API_KEY").ok().map(Into::into)
+}
+
+pub fn default_site() -> String {
+    std::env::var("DD_SITE").unwrap_or(DD_US_SITE.to_string())
+}

src/config/enterprise.rs

@@ -17,18 +17,15 @@ use url::{ParseError, Url};
 use vector_core::config::proxy::ProxyConfig;
 
 use super::{
-    load_source_from_paths, process_paths, ComponentKey, Config, ConfigPath, OutputId, SinkOuter,
-    SourceOuter, TransformOuter,
+    datadog::default_site, load_source_from_paths, process_paths, ComponentKey, Config, ConfigPath,
+    OutputId, SinkOuter, SourceOuter, TransformOuter,
 };
 use crate::{
     common::datadog::{get_api_base_endpoint, get_base_domain_region, Region},
     conditions::AnyCondition,
     http::{HttpClient, HttpError},
     sinks::{
-        datadog::{
-            default_site, logs::DatadogLogsConfig, metrics::DatadogMetricsConfig,
-            DatadogCommonConfig,
-        },
+        datadog::{logs::DatadogLogsConfig, metrics::DatadogMetricsConfig, DatadogCommonConfig},
         util::{http::RequestConfig, retries::ExponentialBackoff},
     },
     sources::{
@@ -481,12 +478,11 @@ fn setup_logs_reporting(
 
     // Create a Datadog logs sink to consume and emit internal logs.
     let datadog_logs = DatadogLogsConfig {
-        dd_common: DatadogCommonConfig {
-            endpoint: datadog.endpoint.clone(),
-            site: datadog.site.clone(),
-            default_api_key: api_key.into(),
-            ..Default::default()
-        },
+        dd_common: DatadogCommonConfig::new(
+            datadog.endpoint.clone(),
+            Some(datadog.site.clone()),
+            Some(api_key.into()),
+        ),
         region: datadog.region,
         request: RequestConfig {
             headers: IndexMap::from([(
@@ -589,12 +585,11 @@ fn setup_metrics_reporting(
 
     // Create a Datadog metrics sink to consume and emit internal + host metrics.
     let datadog_metrics = DatadogMetricsConfig {
-        dd_common: DatadogCommonConfig {
-            endpoint: datadog.endpoint.clone(),
-            site: datadog.site.clone(),
-            default_api_key: api_key.into(),
-            ..Default::default()
-        },
+        dd_common: DatadogCommonConfig::new(
+            datadog.endpoint.clone(),
+            Some(datadog.site.clone()),
+            Some(api_key.into()),
+        ),
         region: datadog.region,
         ..Default::default()
     };

src/config/mod.rs

@@ -23,6 +23,12 @@ pub mod api;
 mod builder;
 mod cmd;
 mod compiler;
+#[cfg(any(
+    feature = "sinks-datadog_logs",
+    feature = "sinks-datadog_metrics",
+    feature = "sinks-datadog_traces",
+))]
+pub mod datadog;
 mod diff;
 mod enrichment_table;
 #[cfg(feature = "enterprise")]
@@ -106,6 +112,12 @@ pub struct Config {
     #[cfg(feature = "api")]
     pub api: api::Options,
     pub schema: schema::Options,
+    #[cfg(any(
+        feature = "sinks-datadog_logs",
+        feature = "sinks-datadog_metrics",
+        feature = "sinks-datadog_traces",
+    ))]
+    pub datadog: datadog::Options,
     pub hash: Option<String>,
     #[cfg(feature = "enterprise")]
     pub enterprise: Option<enterprise::Options>,

src/config/sink.rs

@@ -14,6 +14,7 @@ use vector_core::{
     sink::VectorSink,
 };
 
+use super::datadog;
 use super::{id::Inputs, schema, ComponentKey, ProxyConfig, Resource};
 use crate::sinks::{util::UriSerde, Healthcheck};
 
@@ -239,6 +240,12 @@ dyn_clone::clone_trait_object!(SinkConfig);
 pub struct SinkContext {
     pub healthcheck: SinkHealthcheckOptions,
     pub globals: GlobalOptions,
+    #[cfg(any(
+        feature = "sinks-datadog_logs",
+        feature = "sinks-datadog_metrics",
+        feature = "sinks-datadog_traces",
+    ))]
+    pub datadog: datadog::Options,
     pub proxy: ProxyConfig,
     pub schema: schema::Options,
     pub app_name: String,
@@ -250,6 +257,7 @@ impl Default for SinkContext {
         Self {
             healthcheck: Default::default(),
             globals: Default::default(),
+            datadog: Default::default(),
             proxy: Default::default(),
             schema: Default::default(),
             app_name: crate::get_app_name().to_string(),

src/sinks/datadog/events/config.rs

@@ -7,7 +7,7 @@ use vrl::value::Kind;
 
 use crate::{
     common::datadog::{get_base_domain_region, Region},
-    config::{AcknowledgementsConfig, GenerateConfig, Input, SinkConfig, SinkContext},
+    config::{datadog, AcknowledgementsConfig, GenerateConfig, Input, SinkConfig, SinkContext},
     http::HttpClient,
     sinks::{
         datadog::{
@@ -54,10 +54,10 @@ impl GenerateConfig for DatadogEventsConfig {
 }
 
 impl DatadogEventsConfig {
-    fn get_api_events_endpoint(&self) -> http::Uri {
+    fn get_api_events_endpoint(&self, global: &datadog::Options) -> http::Uri {
         let api_base_endpoint = get_api_base_endpoint(
             self.dd_common.endpoint.as_ref(),
-            get_base_domain_region(self.dd_common.site.as_str(), self.region.as_ref()),
+            get_base_domain_region(self.dd_common.site(global), self.region.as_ref()),
         );
 
         // We know this URI will be valid since we have just built it up ourselves.
@@ -70,10 +70,14 @@ impl DatadogEventsConfig {
         Ok(client)
     }
 
-    fn build_sink(&self, client: HttpClient) -> crate::Result<VectorSink> {
+    fn build_sink(
+        &self,
+        global: &datadog::Options,
+        client: HttpClient,
+    ) -> crate::Result<VectorSink> {
         let service = DatadogEventsService::new(
-            self.get_api_events_endpoint(),
-            self.dd_common.default_api_key.clone().into(),
+            self.get_api_events_endpoint(global),
+            self.dd_common.default_api_key(global),
             client,
         );
 
@@ -96,10 +100,10 @@ impl DatadogEventsConfig {
 impl SinkConfig for DatadogEventsConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
         let client = self.build_client(cx.proxy())?;
-        let healthcheck = self
-            .dd_common
-            .build_healthcheck(client.clone(), self.region.as_ref())?;
-        let sink = self.build_sink(client)?;
+        let healthcheck =
+            self.dd_common
+                .build_healthcheck(&cx.datadog, client.clone(), self.region.as_ref())?;
+        let sink = self.build_sink(&cx.datadog, client)?;
 
         Ok((sink, healthcheck))
     }

src/sinks/datadog/events/service.rs

@@ -8,7 +8,10 @@ use futures::{
 use http::Request;
 use hyper::Body;
 use tower::{Service, ServiceExt};
-use vector_common::request_metadata::{GroupedCountByteSize, MetaDescriptive};
+use vector_common::{
+    request_metadata::{GroupedCountByteSize, MetaDescriptive},
+    sensitive_string::SensitiveString,
+};
 use vector_stream::DriverResponse;
 
 use crate::{
@@ -53,15 +56,15 @@ pub struct DatadogEventsService {
 impl DatadogEventsService {
     pub fn new(
         endpoint: http::Uri,
-        default_api_key: String,
+        default_api_key: SensitiveString,
         http_client: HttpClient<Body>,
     ) -> Self {
         let batch_http_service = HttpBatchService::new(http_client, move |req| {
             let req: DatadogEventsRequest = req;
 
             let api_key = match req.metadata.api_key.as_ref() {
                 Some(x) => x.as_ref(),
-                None => default_api_key.as_str(),
+                None => default_api_key.inner(),
             };
 
             let request = Request::post(&endpoint)