saluki_components/encoders/datadog/service_checks/
mod.rs

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

saluki_components/encoders/datadog/service_checks/
mod.rs
