saluki_components/encoders/datadog/logs/
mod.rs

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

saluki_components/encoders/datadog/logs/
mod.rs
