vector/sinks/datadog/events/
config.rs

1use http::Uri;
2use indoc::indoc;
3use tower::ServiceBuilder;
4use vector_lib::{config::proxy::ProxyConfig, configurable::configurable_component, schema};
5use vrl::value::Kind;
6
7use super::{
8    service::{DatadogEventsResponse, DatadogEventsService},
9    sink::DatadogEventsSink,
10};
11use crate::{
12    common::datadog,
13    config::{
14        AcknowledgementsConfig, DynValidatedSink, GenerateConfig, Input, SinkConfig, SinkContext,
15        ValidatedSink,
16    },
17    http::HttpClient,
18    sinks::{
19        Healthcheck, VectorSink,
20        datadog::{DatadogCommonConfig, LocalDatadogCommonConfig},
21        util::{
22            HttpEndpoint, ServiceBuilderExt, TowerRequestConfig, TowerRequestSettings,
23            http::{HttpStatusRetryLogic, RetryStrategy},
24        },
25    },
26    tls::MaybeTlsSettings,
27};
28
29/// Configuration for the `datadog_events` sink.
30#[configurable_component(sink(
31    "datadog_events",
32    "Publish observability events to the Datadog Events API."
33))]
34#[derive(Clone, Debug, Default)]
35#[serde(deny_unknown_fields)]
36pub struct DatadogEventsConfig {
37    #[serde(flatten)]
38    pub dd_common: LocalDatadogCommonConfig,
39
40    #[configurable(derived)]
41    #[serde(default)]
42    pub request: TowerRequestConfig,
43
44    #[configurable(derived)]
45    #[serde(default)]
46    pub retry_strategy: RetryStrategy,
47}
48
49impl GenerateConfig for DatadogEventsConfig {
50    fn generate_config() -> serde_json::Value {
51        serde_yaml::from_str(indoc! {r#"
52            default_api_key: ${DATADOG_API_KEY_ENV_VAR}
53        "#})
54        .unwrap()
55    }
56}
57
58impl DatadogEventsConfig {
59    /// Resolve the events API endpoint URI from the given endpoint/site.
60    fn events_endpoint(endpoint: Option<&str>, site: &str) -> crate::Result<Uri> {
61        let base = datadog::get_api_base_endpoint(endpoint, site);
62        Ok(HttpEndpoint::parse(&base)?
63            .append_path("/api/v1/events")?
64            .into_uri())
65    }
66
67    fn build_client(&self, proxy: &ProxyConfig) -> crate::Result<HttpClient> {
68        let tls = MaybeTlsSettings::from_config(self.dd_common.tls.as_ref(), false)?;
69        let client = HttpClient::new(tls, proxy)?;
70        Ok(client)
71    }
72
73    fn build_sink(
74        &self,
75        dd_common: &DatadogCommonConfig,
76        client: HttpClient,
77        validated: &ValidatedEvents,
78        endpoint: Uri,
79    ) -> crate::Result<VectorSink> {
80        let service =
81            DatadogEventsService::new(endpoint, dd_common.default_api_key.clone(), client);
82
83        let request_settings = validated.request_settings.clone();
84        let retry_logic = HttpStatusRetryLogic::new(
85            |req: &DatadogEventsResponse| req.http_status,
86            self.retry_strategy.clone(),
87        );
88
89        let service = ServiceBuilder::new()
90            .settings(request_settings, retry_logic)
91            .service(service);
92
93        let sink = DatadogEventsSink { service };
94
95        Ok(VectorSink::from_event_streamsink(sink))
96    }
97}
98
99#[async_trait::async_trait]
100#[typetag::serde(name = "datadog_events")]
101impl SinkConfig for DatadogEventsConfig {
102    fn input(&self) -> Input {
103        let requirement = schema::Requirement::empty()
104            .required_meaning("message", Kind::bytes())
105            .optional_meaning("host", Kind::bytes())
106            .optional_meaning("timestamp", Kind::timestamp());
107
108        Input::log().with_schema_requirement(requirement)
109    }
110
111    fn acknowledgements(&self) -> &AcknowledgementsConfig {
112        &self.dd_common.acknowledgements
113    }
114
115    fn as_dyn_validated(&self) -> Option<&dyn DynValidatedSink> {
116        Some(self)
117    }
118}
119
120#[derive(Clone, Debug)]
121pub struct ValidatedEvents {
122    request_settings: TowerRequestSettings,
123}
124
125#[async_trait::async_trait]
126impl ValidatedSink for DatadogEventsConfig {
127    type Validated = ValidatedEvents;
128
129    fn validate(&self) -> crate::Result<ValidatedEvents> {
130        let site = self
131            .dd_common
132            .site
133            .clone()
134            .unwrap_or_else(|| datadog::DD_US_SITE.to_owned());
135        let uri = Self::events_endpoint(self.dd_common.endpoint.as_deref(), &site)?;
136        if !matches!(uri.scheme_str(), Some("http" | "https")) || uri.authority().is_none() {
137            return Err("Datadog Events endpoint must be an absolute http(s) URL".into());
138        }
139        let request_settings = self.request.into_settings();
140        Ok(ValidatedEvents { request_settings })
141    }
142
143    async fn build(
144        &self,
145        validated: &ValidatedEvents,
146        cx: SinkContext,
147    ) -> crate::Result<(VectorSink, Healthcheck)> {
148        let client = self.build_client(cx.proxy())?;
149        let global = cx.extra_context.get_or_default::<datadog::Options>();
150        let dd_common = self.dd_common.with_globals(global)?;
151        let healthcheck = dd_common.build_healthcheck(client.clone())?;
152        let endpoint = Self::events_endpoint(dd_common.endpoint.as_deref(), &dd_common.site)?;
153        let sink = self.build_sink(&dd_common, client, validated, endpoint)?;
154
155        Ok((sink, healthcheck))
156    }
157}
158
159#[cfg(test)]
160mod tests {
161    use super::*;
162
163    #[test]
164    fn generate_config() {
165        crate::test_util::test_generate_config::<DatadogEventsConfig>();
166    }
167
168    #[test]
169    fn validate_produces_usable_state() {
170        let config = DatadogEventsConfig {
171            dd_common: LocalDatadogCommonConfig::new(
172                Some("http://127.0.0.1:8080".to_string()),
173                None,
174                None,
175            ),
176            ..Default::default()
177        };
178        config.validate().expect("validation should succeed");
179        // The delivery endpoint is derived at build time after globals; the
180        // pure endpoint resolution still produces the expected events URI.
181        assert_eq!(
182            DatadogEventsConfig::events_endpoint(
183                Some("http://127.0.0.1:8080"),
184                datadog::DD_US_SITE,
185            )
186            .expect("endpoint should parse")
187            .to_string(),
188            "http://127.0.0.1:8080/api/v1/events"
189        );
190    }
191
192    #[test]
193    fn validate_rejects_malformed_endpoint() {
194        let config = DatadogEventsConfig {
195            dd_common: LocalDatadogCommonConfig::new(Some("not a uri".to_string()), None, None),
196            ..Default::default()
197        };
198        assert!(config.validate().is_err());
199    }
200
201    #[test]
202    fn validate_accepts_endpoint_without_scheme() {
203        let config = DatadogEventsConfig {
204            dd_common: LocalDatadogCommonConfig::new(
205                Some("localhost:8080".to_string()),
206                None,
207                None,
208            ),
209            ..Default::default()
210        };
211        config.validate().expect("validation should succeed");
212        // A missing scheme is defaulted to https, matching the shared Datadog
213        // endpoint contract used by the healthcheck and other Datadog sinks.
214        assert_eq!(
215            DatadogEventsConfig::events_endpoint(Some("localhost:8080"), datadog::DD_US_SITE)
216                .expect("endpoint should parse")
217                .to_string(),
218            "https://localhost:8080/api/v1/events"
219        );
220    }
221}
vector/sinks/datadog/events/config.rs

vector/sinks/datadog/events/
config.rs
