vector/common/
datadog.rs

1//! Functionality shared between Datadog sources and sinks.
2// Allow unused imports here, since use of these functions will differ depending on the
3// Datadog component type, whether it's used in integration tests, etc.
4#![allow(dead_code)]
5#![allow(unreachable_pub)]
6
7use std::sync::LazyLock;
8
9use regex::Regex;
10use serde::{Deserialize, Serialize};
11use vector_lib::{
12    event::DatadogMetricOriginMetadata, schema::meaning, sensitive_string::SensitiveString,
13};
14
15pub(crate) const DD_US_SITE: &str = "datadoghq.com";
16pub(crate) const DD_EU_SITE: &str = "datadoghq.eu";
17
18/// The datadog tags event path.
19pub const DDTAGS: &str = "ddtags";
20/// The datadog message event path.
21pub const MESSAGE: &str = "message";
22
23/// Mapping of the semantic meaning of well known Datadog reserved attributes
24/// to the field name that Datadog intake expects.
25// https://docs.datadoghq.com/logs/log_configuration/attributes_naming_convention/?s=severity#reserved-attributes
26pub const DD_RESERVED_SEMANTIC_ATTRS: [(&str, &str); 6] = [
27    (meaning::SEVERITY, "status"), // status is intentionally semantically defined as severity
28    (meaning::TIMESTAMP, "timestamp"),
29    (meaning::HOST, "hostname"),
30    (meaning::SERVICE, "service"),
31    (meaning::SOURCE, "ddsource"),
32    (meaning::TAGS, DDTAGS),
33];
34
35/// Returns true if the parameter `attr` is one of the reserved Datadog log attributes
36pub fn is_reserved_attribute(attr: &str) -> bool {
37    DD_RESERVED_SEMANTIC_ATTRS
38        .iter()
39        .any(|(_, attr_str)| &attr == attr_str)
40}
41
42/// DatadogSeriesMetric
43#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
44pub struct DatadogSeriesMetric {
45    /// metric
46    pub metric: String,
47    /// metric type
48    pub r#type: DatadogMetricType,
49    /// interval
50    pub interval: Option<u32>,
51    /// points
52    pub points: Vec<DatadogPoint<f64>>,
53    /// tags
54    pub tags: Option<Vec<String>>,
55    /// host
56    #[serde(skip_serializing_if = "Option::is_none")]
57    pub host: Option<String>,
58    /// source_type_name
59    #[serde(skip_serializing_if = "Option::is_none")]
60    pub source_type_name: Option<String>,
61    /// device
62    #[serde(skip_serializing_if = "Option::is_none")]
63    pub device: Option<String>,
64    /// metadata
65    #[serde(skip_serializing_if = "Option::is_none")]
66    pub metadata: Option<DatadogSeriesMetricMetadata>,
67}
68
69/// Datadog series metric metadata
70#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
71pub struct DatadogSeriesMetricMetadata {
72    #[serde(skip_serializing_if = "Option::is_none")]
73    pub(crate) origin: Option<DatadogMetricOriginMetadata>,
74}
75
76/// Datadog Metric Type
77#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
78#[serde(rename_all = "snake_case")]
79pub enum DatadogMetricType {
80    /// Gauge
81    Gauge,
82    /// Count
83    Count,
84    /// Rate
85    Rate,
86}
87
88/// Datadog Point
89#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
90pub struct DatadogPoint<T>(pub i64, pub T);
91
92/// Gets the base API endpoint to use for any calls to Datadog.
93///
94/// If `endpoint` is not specified, we fallback to `site`.
95pub(crate) fn get_api_base_endpoint(endpoint: Option<&str>, site: &str) -> String {
96    endpoint.map_or_else(|| format!("https://api.{site}"), compute_api_endpoint)
97}
98
99/// Computes the Datadog API endpoint from a given endpoint string.
100///
101/// This scans the given endpoint for the common Datadog domain names; and, if found, rewrites the
102/// endpoint string using the standard API URI. If not found, the endpoint is used as-is.
103fn compute_api_endpoint(endpoint: &str) -> String {
104    // This mechanism is derived from the forwarder health check in the Datadog Agent:
105    // https://github.com/DataDog/datadog-agent/blob/cdcf0fc809b9ac1cd6e08057b4971c7dbb8dbe30/comp/forwarder/defaultforwarder/forwarder_health.go#L45-L47
106    // https://github.com/DataDog/datadog-agent/blob/cdcf0fc809b9ac1cd6e08057b4971c7dbb8dbe30/comp/forwarder/defaultforwarder/forwarder_health.go#L188-L190
107    static DOMAIN_REGEX: LazyLock<Regex> = LazyLock::new(|| {
108        Regex::new(r"((?:[a-z]{2}\d\.)?(?:datadoghq\.[a-z]+|ddog-gov\.com))/*$")
109            .expect("Could not build Datadog domain regex")
110    });
111
112    if let Some(caps) = DOMAIN_REGEX.captures(endpoint) {
113        format!("https://api.{}", &caps[1])
114    } else {
115        endpoint.into()
116    }
117}
118
119/// Default settings to use for Datadog components.
120#[derive(Clone, Debug, Derivative)]
121#[derivative(Default)]
122pub struct Options {
123    /// Default Datadog API key to use for Datadog components.
124    ///
125    /// This can also be specified with the `DD_API_KEY` environment variable.
126    #[derivative(Default(value = "default_api_key()"))]
127    pub api_key: Option<SensitiveString>,
128
129    /// Default site to use for Datadog components.
130    ///
131    /// This can also be specified with the `DD_SITE` environment variable.
132    #[derivative(Default(value = "default_site()"))]
133    pub site: String,
134}
135
136fn default_api_key() -> Option<SensitiveString> {
137    std::env::var("DD_API_KEY").ok().map(Into::into)
138}
139
140pub(crate) fn default_site() -> String {
141    std::env::var("DD_SITE").unwrap_or(DD_US_SITE.to_string())
142}
143
144#[cfg(test)]
145mod tests {
146    use similar_asserts::assert_eq;
147
148    use super::*;
149
150    #[test]
151    fn computes_correct_api_endpoint() {
152        assert_eq!(
153            compute_api_endpoint("https://http-intake.logs.datadoghq.com"),
154            "https://api.datadoghq.com"
155        );
156        assert_eq!(
157            compute_api_endpoint("https://http-intake.logs.datadoghq.com/"),
158            "https://api.datadoghq.com"
159        );
160        assert_eq!(
161            compute_api_endpoint("http://http-intake.logs.datadoghq.com/"),
162            "https://api.datadoghq.com"
163        );
164        assert_eq!(
165            compute_api_endpoint("https://anythingelse.datadoghq.com/"),
166            "https://api.datadoghq.com"
167        );
168        assert_eq!(
169            compute_api_endpoint("https://this.datadoghq.eu/"),
170            "https://api.datadoghq.eu"
171        );
172        assert_eq!(
173            compute_api_endpoint("http://datadog.com/"),
174            "http://datadog.com/"
175        );
176    }
177
178    #[test]
179    fn preserves_site_prefix_in_api_endpoint() {
180        for (prefix, tld) in [
181            ("us3", "com"),
182            ("us5", "com"),
183            ("ap1", "com"),
184            ("eu1", "eu"),
185        ] {
186            assert_eq!(
187                compute_api_endpoint(&format!(
188                    "https://http-intake.logs.{prefix}.datadoghq.{tld}"
189                )),
190                format!("https://api.{prefix}.datadoghq.{tld}")
191            );
192        }
193        assert_eq!(
194            compute_api_endpoint("https://1-2-3-observability-pipelines.agent.us3.datadoghq.com"),
195            "https://api.us3.datadoghq.com"
196        );
197    }
198
199    #[test]
200    fn gets_correct_api_base_endpoint() {
201        assert_eq!(
202            get_api_base_endpoint(None, DD_US_SITE),
203            "https://api.datadoghq.com"
204        );
205        assert_eq!(
206            get_api_base_endpoint(None, "datadog.net"),
207            "https://api.datadog.net"
208        );
209        assert_eq!(
210            get_api_base_endpoint(Some("https://logs.datadoghq.eu"), DD_US_SITE),
211            "https://api.datadoghq.eu"
212        );
213        assert_eq!(
214            get_api_base_endpoint(
215                Some("https://http-intake.logs.us3.datadoghq.com"),
216                DD_US_SITE
217            ),
218            "https://api.us3.datadoghq.com"
219        );
220    }
221}
vector/common/datadog.rs

vector/common/
datadog.rs
