saluki_components/encoders/datadog/logs/
mod.rs

1use async_trait::async_trait;
2use chrono::{SecondsFormat, Utc};
3use facet::Facet;
4use http::{uri::PathAndQuery, HeaderValue, Method, Uri};
5use saluki_common::iter::ReusableDeduplicator;
6use saluki_config::GenericConfiguration;
7use saluki_context::tags::Tag;
8use saluki_core::accounting::{MemoryBounds, MemoryBoundsBuilder};
9use saluki_core::{
10    components::{encoders::*, ComponentContext},
11    data_model::{
12        event::{log::Log, Event, EventType},
13        payload::{HttpPayload, Payload, PayloadMetadata, PayloadType},
14    },
15    observability::ComponentMetricsExt as _,
16    topology::PayloadsDispatcher,
17};
18use saluki_error::{ErrorContext as _, GenericError};
19use saluki_io::compression::CompressionScheme;
20use saluki_metrics::MetricsBuilder;
21use serde::Deserialize;
22use serde_json::{Map as JsonMap, Value as JsonValue};
23use tracing::{error, warn};
24
25use crate::common::datadog::{
26    io::RB_BUFFER_CHUNK_SIZE,
27    request_builder::{EndpointEncoder, RequestBuilder},
28    resolve_zstd_compressor_level,
29    telemetry::ComponentTelemetry,
30    DEFAULT_INTAKE_COMPRESSED_SIZE_LIMIT, DEFAULT_INTAKE_UNCOMPRESSED_SIZE_LIMIT,
31};
32
33const DEFAULT_SERIALIZER_COMPRESSOR_KIND: &str = "zstd";
34const MAX_LOGS_PER_PAYLOAD: usize = 1000;
35
36static CONTENT_TYPE_JSON: HeaderValue = HeaderValue::from_static("application/json");
37
38fn default_serializer_compressor_kind() -> String {
39    DEFAULT_SERIALIZER_COMPRESSOR_KIND.to_owned()
40}
41
42/// Datadog Logs incremental encoder.
43#[derive(Deserialize, Debug, Facet)]
44#[cfg_attr(test, derive(PartialEq, serde::Serialize))]
45pub struct DatadogLogsConfiguration {
46    /// Compression kind for Logs payloads. Defaults to `zstd`.
47    #[serde(
48        rename = "serializer_compressor_kind",
49        default = "default_serializer_compressor_kind"
50    )]
51    compressor_kind: String,
52
53    /// ADP-specific zstd compression level, taking precedence over `serializer_zstd_compressor_level`.
54    /// See [`resolve_zstd_compressor_level`] for how the effective level is determined.
55    #[serde(rename = "data_plane_serializer_zstd_compressor_level", default)]
56    data_plane_zstd_compressor_level: Option<i32>,
57
58    /// The Core Agent's zstd compression level, used only when set to a non-default value (not 1).
59    /// See [`resolve_zstd_compressor_level`] for how the effective level is determined.
60    #[serde(rename = "serializer_zstd_compressor_level", default)]
61    serializer_zstd_compressor_level: Option<i32>,
62}
63
64impl DatadogLogsConfiguration {
65    /// Creates a new `DatadogLogsConfiguration` from the given configuration.
66    pub fn from_configuration(config: &GenericConfiguration) -> Result<Self, GenericError> {
67        Ok(config.as_typed()?)
68    }
69}
70
71#[async_trait]
72impl IncrementalEncoderBuilder for DatadogLogsConfiguration {
73    type Output = DatadogLogs;
74
75    fn input_event_type(&self) -> EventType {
76        EventType::Log
77    }
78
79    fn output_payload_type(&self) -> PayloadType {
80        PayloadType::Http
81    }
82
83    async fn build(&self, context: ComponentContext) -> Result<Self::Output, GenericError> {
84        let metrics_builder = MetricsBuilder::from_component_context(&context);
85        let telemetry = ComponentTelemetry::from_builder(&metrics_builder);
86        let zstd_compressor_level = resolve_zstd_compressor_level(
87            self.data_plane_zstd_compressor_level,
88            self.serializer_zstd_compressor_level,
89        );
90        let compression_scheme = CompressionScheme::new(&self.compressor_kind, zstd_compressor_level);
91
92        let mut request_builder =
93            RequestBuilder::new(LogsEndpointEncoder::new(), compression_scheme, RB_BUFFER_CHUNK_SIZE).await?;
94        request_builder.with_max_inputs_per_payload(MAX_LOGS_PER_PAYLOAD);
95
96        Ok(DatadogLogs {
97            request_builder,
98            telemetry,
99        })
100    }
101}
102
103impl MemoryBounds for DatadogLogsConfiguration {
104    fn specify_bounds(&self, builder: &mut MemoryBoundsBuilder) {
105        // TODO: How do we properly represent the requests we can generate that may be sitting around in-flight?
106
107        builder.minimum().with_single_value::<DatadogLogs>("component struct");
108        builder.firm().with_array::<Log>("logs buffer", MAX_LOGS_PER_PAYLOAD);
109    }
110}
111
112pub struct DatadogLogs {
113    request_builder: RequestBuilder<LogsEndpointEncoder>,
114    telemetry: ComponentTelemetry,
115}
116
117#[async_trait]
118impl IncrementalEncoder for DatadogLogs {
119    async fn process_event(&mut self, event: Event) -> Result<ProcessResult, GenericError> {
120        let log: Log = match event {
121            Event::Log(log) => log,
122            _ => return Ok(ProcessResult::Continue),
123        };
124        match self.request_builder.encode(log).await {
125            Ok(None) => Ok(ProcessResult::Continue),
126            Ok(Some(log)) => Ok(ProcessResult::FlushRequired(Event::Log(log))),
127            Err(e) => {
128                if e.is_recoverable() {
129                    warn!(error = %e, "Failed to encode Datadog log due to recoverable error. Continuing...");
130
131                    // TODO: Get the actual number of events dropped from the error itself.
132                    self.telemetry.events_dropped_encoder().increment(1);
133                    Ok(ProcessResult::Continue)
134                } else {
135                    Err(e).error_context("Failed to encode Datadog log due to unrecoverable error.")
136                }
137            }
138        }
139    }
140
141    async fn flush(&mut self, dispatcher: &PayloadsDispatcher) -> Result<(), GenericError> {
142        let maybe_requests = self.request_builder.flush().await;
143        for maybe_request in maybe_requests {
144            match maybe_request {
145                Ok((events, _data_points, request)) => {
146                    let payload_meta = PayloadMetadata::from_event_count(events);
147                    let http_payload = HttpPayload::new(payload_meta, request);
148                    let payload = Payload::Http(http_payload);
149                    dispatcher.dispatch(payload).await?;
150                }
151                Err(e) => error!(error = %e, "Failed to build Datadog logs payload. Continuing..."),
152            }
153        }
154
155        Ok(())
156    }
157}
158
159#[derive(Debug)]
160struct LogsEndpointEncoder {
161    tags_deduplicator: ReusableDeduplicator<Tag>,
162}
163
164impl LogsEndpointEncoder {
165    fn new() -> Self {
166        Self {
167            tags_deduplicator: ReusableDeduplicator::new(),
168        }
169    }
170
171    fn build_agent_json(&mut self, log: &Log) -> JsonValue {
172        let mut obj = JsonMap::new();
173
174        // Encode a structured message object as a JSON string in the `message` field.
175        let mut message_inner = JsonMap::new();
176        message_inner.insert("message".to_string(), JsonValue::String(log.message().to_string()));
177        if !log.service().is_empty() {
178            message_inner.insert("service".to_string(), JsonValue::String(log.service().to_string()));
179        }
180        let message_str =
181            serde_json::to_string(&JsonValue::Object(message_inner)).unwrap_or_else(|_| log.message().to_string());
182        obj.insert("message".to_string(), JsonValue::String(message_str));
183
184        if let Some(status) = log.status() {
185            obj.insert("status".to_string(), JsonValue::String(status.as_str().to_string()));
186        }
187        if !log.hostname().is_empty() {
188            obj.insert("hostname".to_string(), JsonValue::String(log.hostname().to_string()));
189        }
190        if !log.service().is_empty() {
191            obj.insert("service".to_string(), JsonValue::String(log.service().to_string()));
192        }
193
194        if let Some(ddsource) = log.source().clone() {
195            obj.insert("ddsource".to_string(), JsonValue::String(ddsource.to_string()));
196        }
197
198        // ddtags: comma-separated, deduplicated
199        let tags_iter = self.tags_deduplicator.deduplicated(log.tags().into_iter());
200        let tags_vec: Vec<&str> = tags_iter.map(|t| t.as_str()).collect();
201        if !tags_vec.is_empty() {
202            obj.insert("ddtags".to_string(), JsonValue::String(tags_vec.join(",")));
203        }
204
205        // Default timestamp (RFC3339 with milliseconds, Z) unless user provided `timestamp` or `@timestamp`.
206        let user_provided_timestamp = log.additional_properties().contains_key("timestamp")
207            || log.additional_properties().contains_key("@timestamp");
208        if !user_provided_timestamp {
209            let now_rfc3339 = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true);
210            obj.insert("@timestamp".to_string(), JsonValue::String(now_rfc3339));
211        }
212
213        // Last-write-wins: merge AdditionalProperties last
214        for (k, v) in log.additional_properties() {
215            obj.insert(k.to_string(), v.clone());
216        }
217
218        JsonValue::Object(obj)
219    }
220}
221
222impl EndpointEncoder for LogsEndpointEncoder {
223    type Input = Log;
224    type EncodeError = serde_json::Error;
225
226    fn encoder_name() -> &'static str {
227        "logs"
228    }
229
230    fn compressed_size_limit(&self) -> usize {
231        DEFAULT_INTAKE_COMPRESSED_SIZE_LIMIT
232    }
233
234    fn uncompressed_size_limit(&self) -> usize {
235        DEFAULT_INTAKE_UNCOMPRESSED_SIZE_LIMIT
236    }
237
238    fn get_payload_prefix(&self) -> Option<&'static [u8]> {
239        Some(b"[")
240    }
241
242    fn get_payload_suffix(&self) -> Option<&'static [u8]> {
243        Some(b"]")
244    }
245
246    fn get_input_separator(&self) -> Option<&'static [u8]> {
247        Some(b",")
248    }
249
250    fn encode(&mut self, input: &Self::Input, buffer: &mut Vec<u8>) -> Result<(), Self::EncodeError> {
251        let json = self.build_agent_json(input);
252        serde_json::to_writer(buffer, &json)
253    }
254
255    fn endpoint_uri(&self) -> Uri {
256        PathAndQuery::from_static("/api/v2/logs").into()
257    }
258
259    fn endpoint_method(&self) -> Method {
260        Method::POST
261    }
262
263    fn content_type(&self) -> HeaderValue {
264        CONTENT_TYPE_JSON.clone()
265    }
266}
267
268#[cfg(test)]
269mod tests {
270    use std::collections::{BTreeSet, HashMap};
271
272    use saluki_context::tags::{Tag, TagSet};
273    use saluki_core::data_model::event::log::{Log, LogStatus};
274    use serde_json::json;
275    use stringtheory::MetaString;
276
277    use super::{JsonValue, LogsEndpointEncoder};
278
279    fn tag_set<const N: usize>(tags: [&'static str; N]) -> TagSet {
280        tags.into_iter().map(Tag::from_static).collect()
281    }
282
283    #[test]
284    fn build_agent_json_enriches_documented_fields() {
285        let mut encoder = LogsEndpointEncoder::new();
286        let log = Log::new("hello world")
287            .with_status(LogStatus::Error)
288            .with_source(MetaString::from_static("nginx"))
289            .with_hostname(MetaString::from_static("host-a"))
290            .with_service(MetaString::from_static("web"))
291            .with_tags(tag_set(["env:prod", "team:core"]));
292
293        let json = encoder.build_agent_json(&log);
294        let obj = json.as_object().expect("agent JSON should be an object");
295
296        // `message` is itself a JSON string carrying a `{message, service}` object.
297        let message = obj["message"].as_str().expect("message field should be a string");
298        let message_inner: JsonValue = serde_json::from_str(message).expect("message field should itself be JSON");
299        assert_eq!(json!("hello world"), message_inner["message"]);
300        assert_eq!(json!("web"), message_inner["service"]);
301
302        // Status renders via `LogStatus::as_str` (capitalized), and hostname/service/ddsource are lifted out.
303        assert_eq!(json!("Error"), obj["status"]);
304        assert_eq!(json!("host-a"), obj["hostname"]);
305        assert_eq!(json!("web"), obj["service"]);
306        assert_eq!(json!("nginx"), obj["ddsource"]);
307
308        // `ddtags` is the comma-joined tag set.
309        let ddtags = obj["ddtags"].as_str().expect("ddtags should be a string");
310        let ddtags = ddtags.split(',').collect::<BTreeSet<_>>();
311        assert_eq!(BTreeSet::from(["env:prod", "team:core"]), ddtags);
312    }
313
314    #[test]
315    fn build_agent_json_omits_empty_optional_fields() {
316        // A bare log has no status/hostname/service/ddsource/ddtags keys, and its `message` object carries only the
317        // message (no `service`).
318        let mut encoder = LogsEndpointEncoder::new();
319        let json = encoder.build_agent_json(&Log::new("bare"));
320        let obj = json.as_object().expect("agent JSON should be an object");
321
322        assert!(!obj.contains_key("status"));
323        assert!(!obj.contains_key("hostname"));
324        assert!(!obj.contains_key("service"));
325        assert!(!obj.contains_key("ddsource"));
326        assert!(!obj.contains_key("ddtags"));
327
328        let message = obj["message"].as_str().expect("message field should be a string");
329        let message_inner: JsonValue = serde_json::from_str(message).expect("message field should itself be JSON");
330        assert_eq!(json!("bare"), message_inner["message"]);
331        assert!(message_inner.get("service").is_none());
332    }
333
334    #[test]
335    fn build_agent_json_adds_default_timestamp_unless_user_supplied() {
336        let mut encoder = LogsEndpointEncoder::new();
337
338        // With no user timestamp, the encoder injects an RFC3339 `@timestamp` (milliseconds, UTC `Z`).
339        let json = encoder.build_agent_json(&Log::new("no ts"));
340        let ts = json["@timestamp"]
341            .as_str()
342            .expect("default @timestamp should be present");
343        assert!(ts.ends_with('Z'), "default timestamp should be UTC-suffixed: {ts}");
344        assert!(
345            chrono::DateTime::parse_from_rfc3339(ts).is_ok(),
346            "default timestamp should be RFC3339: {ts}"
347        );
348
349        // A user-provided `timestamp` suppresses the default `@timestamp`.
350        let mut props = HashMap::new();
351        props.insert(MetaString::from_static("timestamp"), json!(1_234_567));
352        let json = encoder.build_agent_json(&Log::new("user ts").with_additional_properties(props));
353        assert!(
354            !json.as_object().unwrap().contains_key("@timestamp"),
355            "a user-provided `timestamp` should suppress the default `@timestamp`"
356        );
357        assert_eq!(json!(1_234_567), json["timestamp"]);
358
359        // A user-provided `@timestamp` is preserved as-is instead of being overwritten.
360        let mut props = HashMap::new();
361        props.insert(MetaString::from_static("@timestamp"), json!("2020-01-01T00:00:00Z"));
362        let json = encoder.build_agent_json(&Log::new("user @ts").with_additional_properties(props));
363        assert_eq!(json!("2020-01-01T00:00:00Z"), json["@timestamp"]);
364    }
365
366    #[test]
367    fn build_agent_json_additional_properties_win_over_encoder_fields() {
368        // AdditionalProperties are merged last, so they override the encoder-populated fields (last-write-wins).
369        let mut encoder = LogsEndpointEncoder::new();
370        let mut props = HashMap::new();
371        props.insert(MetaString::from_static("hostname"), json!("override-host"));
372        props.insert(MetaString::from_static("custom"), json!(42));
373        let log = Log::new("msg")
374            .with_hostname(MetaString::from_static("original-host"))
375            .with_additional_properties(props);
376
377        let json = encoder.build_agent_json(&log);
378        assert_eq!(json!("override-host"), json["hostname"]);
379        assert_eq!(json!(42), json["custom"]);
380    }
381}
382
383#[cfg(test)]
384mod config_smoke {
385    use datadog_agent_config_testing::config_registry::structs;
386    use datadog_agent_config_testing::run_config_smoke_tests;
387    use serde_json::json;
388
389    use super::DatadogLogsConfiguration;
390    use crate::config::{DatadogRemapper, KEY_ALIASES};
391
392    #[tokio::test]
393    async fn smoke_test() {
394        run_config_smoke_tests(
395            structs::DATADOG_LOGS_CONFIGURATION,
396            &[],
397            json!({}),
398            |cfg| {
399                cfg.as_typed::<DatadogLogsConfiguration>()
400                    .expect("DatadogLogsConfiguration should deserialize")
401            },
402            KEY_ALIASES,
403            DatadogRemapper::from_env_vars,
404        )
405        .await
406    }
407}
saluki_components/encoders/datadog/logs/mod.rs

saluki_components/encoders/datadog/logs/
mod.rs
