saluki_components/encoders/datadog/service_checks/
mod.rs

1use async_trait::async_trait;
2use facet::Facet;
3use http::{uri::PathAndQuery, HeaderValue, Method, Uri};
4use saluki_config::GenericConfiguration;
5use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
6use saluki_core::{
7    components::{encoders::*, ComponentContext},
8    data_model::{
9        event::{service_check::ServiceCheck, Event, EventType},
10        payload::{HttpPayload, Payload, PayloadMetadata, PayloadType},
11    },
12    observability::ComponentMetricsExt as _,
13    topology::PayloadsDispatcher,
14};
15use saluki_error::{ErrorContext as _, GenericError};
16use saluki_io::compression::CompressionScheme;
17use saluki_metrics::MetricsBuilder;
18use serde::Deserialize;
19use tracing::{debug, error, warn};
20
21use crate::common::datadog::{
22    clamp_payload_limits,
23    io::RB_BUFFER_CHUNK_SIZE,
24    request_builder::{EndpointEncoder, RequestBuilder},
25    resolve_zstd_compressor_level,
26    telemetry::ComponentTelemetry,
27    DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT, DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT,
28};
29
30const DEFAULT_SERIALIZER_COMPRESSOR_KIND: &str = "zstd";
31const MAX_SERVICE_CHECKS_PER_PAYLOAD: usize = 100;
32
33static CONTENT_TYPE_JSON: HeaderValue = HeaderValue::from_static("application/json");
34
35fn default_serializer_compressor_kind() -> String {
36    DEFAULT_SERIALIZER_COMPRESSOR_KIND.to_owned()
37}
38
39const fn default_max_payload_size() -> usize {
40    DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT
41}
42
43const fn default_max_uncompressed_payload_size() -> usize {
44    DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT
45}
46
47const fn default_log_payloads() -> bool {
48    false
49}
50
51/// Datadog Service Checks incremental encoder.
52///
53/// Generates Datadog Service Checks payloads for the Datadog platform.
54#[derive(Deserialize, Facet)]
55#[cfg_attr(test, derive(Debug, PartialEq, serde::Serialize))]
56pub struct DatadogServiceChecksConfiguration {
57    /// Maximum compressed size, in bytes, of a service check payload.
58    ///
59    /// This matches the Datadog Agent's generic payload limit for service checks. The effective value is
60    /// clamped to the Agent's default intake-safe limit of 2,621,440 bytes, so larger configured values do not allow
61    /// payloads that intake may reject. If set to `0`, every non-empty compressed payload exceeds the limit and is
62    /// dropped during flush.
63    ///
64    /// Defaults to 2,621,440 bytes.
65    #[serde(rename = "serializer_max_payload_size", default = "default_max_payload_size")]
66    max_payload_size: usize,
67
68    /// Maximum uncompressed size, in bytes, of a service check payload.
69    ///
70    /// This matches the Datadog Agent's generic payload limit for service checks. The effective value is
71    /// clamped to the Agent's default intake-safe limit of 4,194,304 bytes, so larger configured values do not allow
72    /// payloads that intake may reject. Values smaller than the minimum endpoint framing size prevent the request
73    /// builder from starting.
74    ///
75    /// Defaults to 4,194,304 bytes.
76    #[serde(
77        rename = "serializer_max_uncompressed_payload_size",
78        default = "default_max_uncompressed_payload_size"
79    )]
80    max_uncompressed_payload_size: usize,
81
82    /// Compression kind to use for the request payloads.
83    ///
84    /// Defaults to `zstd`.
85    #[serde(
86        rename = "serializer_compressor_kind",
87        default = "default_serializer_compressor_kind"
88    )]
89    compressor_kind: String,
90
91    /// ADP-specific zstd compression level, taking precedence over `serializer_zstd_compressor_level`.
92    /// See [`resolve_zstd_compressor_level`] for how the effective level is determined.
93    #[serde(rename = "data_plane_serializer_zstd_compressor_level", default)]
94    data_plane_zstd_compressor_level: Option<i32>,
95
96    /// The Core Agent's zstd compression level, used only when set to a non-default value (not 1).
97    /// See [`resolve_zstd_compressor_level`] for how the effective level is determined.
98    #[serde(rename = "serializer_zstd_compressor_level", default)]
99    serializer_zstd_compressor_level: Option<i32>,
100
101    /// Whether to log service check payload contents before encoding.
102    ///
103    /// This logs decoded service check objects, not the encoded HTTP body.
104    ///
105    /// Defaults to `false`.
106    #[serde(default = "default_log_payloads")]
107    log_payloads: bool,
108}
109
110impl DatadogServiceChecksConfiguration {
111    /// Creates a new `DatadogServiceChecksConfiguration` from the given configuration.
112    pub fn from_configuration(config: &GenericConfiguration) -> Result<Self, GenericError> {
113        Ok(config.as_typed()?)
114    }
115}
116
117#[async_trait]
118impl IncrementalEncoderBuilder for DatadogServiceChecksConfiguration {
119    type Output = DatadogServiceChecks;
120
121    fn input_event_type(&self) -> EventType {
122        EventType::ServiceCheck
123    }
124
125    fn output_payload_type(&self) -> PayloadType {
126        PayloadType::Http
127    }
128
129    async fn build(&self, context: ComponentContext) -> Result<Self::Output, GenericError> {
130        let metrics_builder = MetricsBuilder::from_component_context(&context);
131        let telemetry = ComponentTelemetry::from_builder(&metrics_builder);
132        let zstd_compressor_level = resolve_zstd_compressor_level(
133            self.data_plane_zstd_compressor_level,
134            self.serializer_zstd_compressor_level,
135        );
136        let compression_scheme = CompressionScheme::new(&self.compressor_kind, zstd_compressor_level);
137
138        // Create our request builder.
139        let mut request_builder =
140            RequestBuilder::new(ServiceChecksEndpointEncoder, compression_scheme, RB_BUFFER_CHUNK_SIZE).await?;
141        let (uncompressed_limit, compressed_limit) = clamp_payload_limits(
142            self.max_uncompressed_payload_size,
143            self.max_payload_size,
144            DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT,
145            DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT,
146        );
147        request_builder.with_len_limits(uncompressed_limit, compressed_limit)?;
148        request_builder.with_max_inputs_per_payload(MAX_SERVICE_CHECKS_PER_PAYLOAD);
149
150        Ok(DatadogServiceChecks {
151            request_builder,
152            telemetry,
153            log_payloads: self.log_payloads,
154        })
155    }
156}
157
158impl MemoryBounds for DatadogServiceChecksConfiguration {
159    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
160        // TODO: How do we properly represent the requests we can generate that may be sitting around in-flight?
161        //
162        // Theoretically, we'll end up being limited by the size of the downstream forwarder's interconnect, and however
163        // many payloads it will buffer internally... so realistically the firm limit boils down to the forwarder itself
164        // but we'll have a hard time in the forwarder knowing the maximum size of any given payload being sent in, which
165        // then makes it hard to calculate a proper firm bound even though we know the rest of the values required to
166        // calculate the firm bound.
167        builder
168            .minimum()
169            .with_single_value::<DatadogServiceChecks>("component struct");
170
171        builder
172            .firm()
173            // Capture the size of the "split re-encode" buffer in the request builder, which is where we keep owned
174            // versions of events that we encode in case we need to actually re-encode them during a split operation.
175            .with_array::<ServiceCheck>("service checks split re-encode buffer", MAX_SERVICE_CHECKS_PER_PAYLOAD);
176    }
177}
178
179pub struct DatadogServiceChecks {
180    request_builder: RequestBuilder<ServiceChecksEndpointEncoder>,
181    telemetry: ComponentTelemetry,
182    log_payloads: bool,
183}
184
185#[async_trait]
186impl IncrementalEncoder for DatadogServiceChecks {
187    async fn process_event(&mut self, event: Event) -> Result<ProcessResult, GenericError> {
188        let service_check = match event.try_into_service_check() {
189            Some(eventd) => eventd,
190            None => return Ok(ProcessResult::Continue),
191        };
192
193        if self.log_payloads {
194            debug!(?service_check, "Flushing service check.");
195        }
196
197        match self.request_builder.encode(service_check).await {
198            Ok(None) => Ok(ProcessResult::Continue),
199            Ok(Some(service_check)) => Ok(ProcessResult::FlushRequired(Event::ServiceCheck(service_check))),
200            Err(e) => {
201                if e.is_recoverable() {
202                    warn!(error = %e, "Failed to encode Datadog service check due to recoverable error. Continuing...");
203
204                    // TODO: Get the actual number of events dropped from the error itself.
205                    self.telemetry.events_dropped_encoder().increment(1);
206
207                    Ok(ProcessResult::Continue)
208                } else {
209                    Err(e).error_context("Failed to encode Datadog service check due to unrecoverable error.")
210                }
211            }
212        }
213    }
214
215    async fn flush(&mut self, dispatcher: &PayloadsDispatcher) -> Result<(), GenericError> {
216        let maybe_requests = self.request_builder.flush().await;
217        for maybe_request in maybe_requests {
218            match maybe_request {
219                Ok((events, _data_points, request)) => {
220                    let payload_meta = PayloadMetadata::from_event_count(events);
221                    let http_payload = HttpPayload::new(payload_meta, request);
222                    let payload = Payload::Http(http_payload);
223
224                    dispatcher.dispatch(payload).await?;
225                }
226                Err(e) => error!(error = %e, "Failed to build Datadog service checks payload. Continuing..."),
227            }
228        }
229
230        Ok(())
231    }
232}
233
234#[derive(Debug)]
235struct ServiceChecksEndpointEncoder;
236
237impl EndpointEncoder for ServiceChecksEndpointEncoder {
238    type Input = ServiceCheck;
239    type EncodeError = serde_json::Error;
240
241    fn encoder_name() -> &'static str {
242        "service_check"
243    }
244
245    fn compressed_size_limit(&self) -> usize {
246        DEFAULT_SERIALIZER_COMPRESSED_SIZE_LIMIT
247    }
248
249    fn uncompressed_size_limit(&self) -> usize {
250        DEFAULT_SERIALIZER_UNCOMPRESSED_SIZE_LIMIT
251    }
252
253    fn encode(&mut self, input: &Self::Input, buffer: &mut Vec<u8>) -> Result<(), Self::EncodeError> {
254        serde_json::to_writer(buffer, input)
255    }
256
257    fn get_payload_prefix(&self) -> Option<&'static [u8]> {
258        Some(b"[")
259    }
260
261    fn get_payload_suffix(&self) -> Option<&'static [u8]> {
262        Some(b"]")
263    }
264
265    fn get_input_separator(&self) -> Option<&'static [u8]> {
266        Some(b",")
267    }
268
269    fn endpoint_uri(&self) -> Uri {
270        PathAndQuery::from_static("/api/v1/check_run").into()
271    }
272
273    fn endpoint_method(&self) -> Method {
274        Method::POST
275    }
276
277    fn content_type(&self) -> HeaderValue {
278        CONTENT_TYPE_JSON.clone()
279    }
280}
281
282#[cfg(test)]
283mod config_smoke {
284    use datadog_agent_config_testing::config_registry::structs;
285    use datadog_agent_config_testing::run_config_smoke_tests;
286    use serde_json::json;
287
288    use super::DatadogServiceChecksConfiguration;
289    use crate::config::{DatadogRemapper, KEY_ALIASES};
290
291    #[tokio::test]
292    async fn smoke_test() {
293        run_config_smoke_tests(
294            structs::DATADOG_SERVICE_CHECKS_CONFIGURATION,
295            &[],
296            json!({}),
297            |cfg| {
298                cfg.as_typed::<DatadogServiceChecksConfiguration>()
299                    .expect("DatadogServiceChecksConfiguration should deserialize")
300            },
301            KEY_ALIASES,
302            DatadogRemapper::from_env_vars,
303        )
304        .await
305    }
306}
saluki_components/encoders/datadog/service_checks/mod.rs

saluki_components/encoders/datadog/service_checks/
mod.rs
