saluki_components/forwarders/datadog/
mod.rs

1use async_trait::async_trait;
2use http::Uri;
3use memory_accounting::{MemoryBounds, MemoryBoundsBuilder, UsageExpr};
4use saluki_common::buf::FrozenChunkedBytesBuffer;
5use saluki_config::GenericConfiguration;
6use saluki_core::{
7    components::{forwarders::*, ComponentContext},
8    data_model::payload::PayloadType,
9    observability::ComponentMetricsExt as _,
10};
11use saluki_error::GenericError;
12use saluki_metrics::MetricsBuilder;
13use serde::Deserialize;
14use stringtheory::MetaString;
15use tokio::select;
16use tracing::debug;
17
18use crate::common::datadog::{
19    config::ForwarderConfiguration,
20    io::TransactionForwarder,
21    telemetry::ComponentTelemetry,
22    transaction::{Metadata, Transaction},
23    DEFAULT_INTAKE_COMPRESSED_SIZE_LIMIT,
24};
25
26/// Datadog forwarder.
27///
28/// Forwards Datadog-specific payloads to the Datadog platform. Handles the standard Datadog Agent configuration,
29/// in terms of specifying additional endpoints, adding the necessary HTTP request headers for authentication,
30/// identification, and more.
31#[derive(Deserialize)]
32pub struct DatadogConfiguration {
33    /// Forwarder configuration settings.
34    ///
35    /// See [`ForwarderConfiguration`] for more information about the available settings.
36    #[serde(flatten)]
37    forwarder_config: ForwarderConfiguration,
38
39    #[serde(skip)]
40    configuration: Option<GenericConfiguration>,
41}
42
43impl DatadogConfiguration {
44    /// Creates a new `DatadogConfiguration` from the given configuration.
45    pub fn from_configuration(config: &GenericConfiguration) -> Result<Self, GenericError> {
46        let mut forwarder_config: DatadogConfiguration = config.as_typed()?;
47        forwarder_config.configuration = Some(config.clone());
48        Ok(forwarder_config)
49    }
50
51    /// Overrides the default endpoint that payloads are sent to.
52    ///
53    /// This overrides any existing endpoint configuration, and manually sets the base endpoint (e.g.,
54    /// `https://api.datad0g.com`) to be used for all payloads.
55    ///
56    /// This can be used to preserve other configuration settings (forwarder settings, retry, etc) while still allowing
57    /// for overriding _where_ payloads are sent to.
58    ///
59    /// # Errors
60    ///
61    /// If the given request path is not valid, an error is returned.
62    pub fn with_endpoint_override(mut self, dd_url: String, api_key: String) -> Self {
63        // Clear any existing additional endpoints, and set the new DD URL and API key.
64        //
65        // This ensures that the only endpoint we'll send to is this one.
66        let endpoint = self.forwarder_config.endpoint_mut();
67        endpoint.clear_additional_endpoints();
68        endpoint.set_dd_url(dd_url);
69        endpoint.set_api_key(api_key);
70
71        self
72    }
73}
74
75#[async_trait]
76impl ForwarderBuilder for DatadogConfiguration {
77    fn input_payload_type(&self) -> PayloadType {
78        PayloadType::Http
79    }
80
81    async fn build(&self, context: ComponentContext) -> Result<Box<dyn Forwarder + Send>, GenericError> {
82        let metrics_builder = MetricsBuilder::from_component_context(&context);
83        let telemetry = ComponentTelemetry::from_builder(&metrics_builder);
84        let forwarder = TransactionForwarder::from_config(
85            context,
86            self.forwarder_config.clone(),
87            self.configuration.clone(),
88            get_dd_endpoint_name,
89            telemetry.clone(),
90            metrics_builder,
91        )?;
92
93        Ok(Box::new(Datadog { forwarder }))
94    }
95}
96
97impl MemoryBounds for DatadogConfiguration {
98    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
99        builder
100            .minimum()
101            .with_single_value::<Datadog>("component struct")
102            .with_array::<Transaction<FrozenChunkedBytesBuffer>>("requests channel", 8);
103
104        builder
105            .firm()
106            // TODO: This is a little wonky because we're accounting for the firm bound portion of connected encoders here, as this
107            // is the only place where we can calculate how many requests we'll hold on to in memory, which is what ultimately influences
108            // the firm usage.
109            //
110            // We're also cheating by knowing what the largest possible payload is that we'll potentially see, based on the limits on the
111            // Datadog encoders. This won't necessarily hold up for future sources/encoders, but is good enough for now.
112            .with_expr(UsageExpr::sum(
113                "in-flight requests",
114                UsageExpr::config(
115                    "forwarder_retry_queue_payloads_max_size",
116                    self.forwarder_config.retry().queue_max_size_bytes() as usize,
117                ),
118                UsageExpr::product(
119                    "high priority queue",
120                    UsageExpr::config(
121                        "forwarder_high_prio_buffer_size",
122                        self.forwarder_config.endpoint_buffer_size(),
123                    ),
124                    // TODO: The default compressed size limit just so happens to be the biggest one we currently default with on our side,
125                    // but it's not clear that this will always be the case.
126                    UsageExpr::constant("maximum compressed payload size", DEFAULT_INTAKE_COMPRESSED_SIZE_LIMIT),
127                ),
128            ));
129    }
130}
131
132pub struct Datadog {
133    forwarder: TransactionForwarder<FrozenChunkedBytesBuffer>,
134}
135
136#[async_trait]
137impl Forwarder for Datadog {
138    async fn run(mut self: Box<Self>, mut context: ForwarderContext) -> Result<(), GenericError> {
139        let Self { forwarder } = *self;
140
141        let mut health = context.take_health_handle();
142
143        // Spawn our forwarder task to handle sending requests.
144        let forwarder = forwarder.spawn().await;
145
146        health.mark_ready();
147        debug!("Datadog forwarder started.");
148
149        loop {
150            select! {
151                _ = health.live() => continue,
152                maybe_payload = context.payloads().next() => match maybe_payload {
153                    Some(payload) => if let Some(http_payload) = payload.try_into_http_payload() {
154                        let (payload_meta, request) = http_payload.into_parts();
155                        let transaction_meta = Metadata::from_event_count(payload_meta.event_count());
156                        let transaction = Transaction::from_original(transaction_meta, request);
157
158                        forwarder.send_transaction(transaction).await?;
159                    },
160                    None => break,
161                },
162            }
163        }
164
165        // Shutdown the forwarder gracefully.
166        forwarder.shutdown().await;
167
168        debug!("Datadog forwarder stopped.");
169
170        Ok(())
171    }
172}
173
174fn get_dd_endpoint_name(uri: &Uri) -> Option<MetaString> {
175    match uri.path() {
176        "/api/v2/logs" => Some(MetaString::from_static("logs_v2")),
177        "/api/v2/series" => Some(MetaString::from_static("series_v2")),
178        "/api/beta/sketches" => Some(MetaString::from_static("sketches_v2")),
179        "/api/v1/check_run" => Some(MetaString::from_static("check_run_v1")),
180        "/api/v1/events_batch" => Some(MetaString::from_static("events_batch_v1")),
181        _ => None,
182    }
183}
saluki_components/forwarders/datadog/mod.rs

saluki_components/forwarders/datadog/
mod.rs
