libdd_trace_utils/json_log_encoder/
mod.rs

1// Copyright 2024-Present Datadog, Inc. https://www.datadoghq.com/
2// SPDX-License-Identifier: Apache-2.0
3
4//! JSON "log exporter" trace encoder.
5//!
6//! Emits traces in the newline-delimited JSON format consumed by the Datadog
7//! Forwarder Lambda (the legacy serverless path used when no Datadog Agent /
8//! Lambda Extension is reachable). Each emitted line is a self-contained JSON
9//! document of the form:
10//!
11//! ```text
12//! {"traces":[[ {span}, {span}, ... ]]}\n
13//! ```
14//!
15//! Spans are greedily packed into size-bounded lines. A single span that alone
16//! exceeds the line cap is dropped (and counted in [`EncodeStats::spans_dropped`])
17//! rather than emitted as a truncated, unparseable line.
18//!
19//! See the cross-language specification for the wire contract and the
20//! Forwarder's `is_trace` detection requirements.
21
22mod span;
23mod span_v1;
24
25use crate::span::v04::Span;
26use crate::span::v1::TracerPayload;
27use crate::span::TraceData;
28use span::LogSpan;
29use span_v1::{ChunkContextV1, LogSpanV1};
30use std::io::Write;
31
32/// Opening bytes of every emitted line: `{"traces":[[`.
33const TRACE_PREFIX: &[u8] = b"{\"traces\":[[";
34/// Closing bytes of every emitted line: `]]}` plus the terminating newline.
35const TRACE_SUFFIX: &[u8] = b"]]}\n";
36/// Fixed per-line overhead contributed by [`TRACE_PREFIX`] and [`TRACE_SUFFIX`].
37const TRACE_FORMAT_OVERHEAD: usize = TRACE_PREFIX.len() + TRACE_SUFFIX.len();
38
39/// Statistics reported by [`encode_traces`].
40#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
41pub struct EncodeStats {
42    /// Number of spans successfully written to `out`.
43    pub spans_written: usize,
44    /// Number of spans dropped because a single span exceeded `max_line_size`.
45    pub spans_dropped: usize,
46}
47
48/// Encodes `traces` into newline-delimited JSON "log exporter" lines, writing
49/// them to `out`.
50///
51/// All spans across all input traces are flattened and greedily packed into
52/// lines no larger than `max_line_size` bytes (including the `{"traces":[[` /
53/// `]]}\n` framing). Each line contains a single inner trace array, matching the
54/// reference dd-trace-js exporter. A span whose own serialized size plus the
55/// framing overhead exceeds `max_line_size` is dropped (counted in
56/// [`EncodeStats::spans_dropped`]) and never split across lines.
57///
58/// The caller is responsible for flushing `out` (e.g. stdout) after this returns.
59///
60/// # Errors
61///
62/// Returns any [`std::io::Error`] produced while writing to `out`.
63///
64/// # Examples
65///
66/// ```
67/// use libdd_trace_utils::json_log_encoder::encode_traces;
68/// use libdd_trace_utils::span::v04::SpanSlice;
69///
70/// let span = SpanSlice {
71///     service: "my-fn".into(),
72///     name: "aws.lambda".into(),
73///     resource: "my-fn".into(),
74///     trace_id: 1,
75///     span_id: 2,
76///     ..Default::default()
77/// };
78/// let traces = vec![vec![span]];
79///
80/// let mut out = Vec::new();
81/// let stats = encode_traces(&traces, &mut out, 64 * 1024).unwrap();
82///
83/// assert_eq!(stats.spans_written, 1);
84/// assert!(out.ends_with(b"\n"));
85/// ```
86pub fn encode_traces<T: TraceData>(
87    traces: &[Vec<Span<T>>],
88    out: &mut impl Write,
89    max_line_size: usize,
90) -> std::io::Result<EncodeStats> {
91    let mut stats = EncodeStats::default();
92
93    // Reusable buffers: `span_buf` holds the JSON for the span currently being
94    // considered; `line` accumulates the complete current line. It is primed
95    // with `TRACE_PREFIX` so that a single `write_all` per line can emit
96    // prefix + spans + suffix in one syscall (see `flush_line`).
97    let mut span_buf: Vec<u8> = Vec::with_capacity(512);
98    let mut line: Vec<u8> = Vec::with_capacity(max_line_size.min(64 * 1024));
99    line.extend_from_slice(TRACE_PREFIX);
100    let mut line_span_count: usize = 0;
101
102    for trace in traces {
103        for span in trace {
104            span_buf.clear();
105            serde_json::to_writer(&mut span_buf, &LogSpan(span)).map_err(std::io::Error::other)?;
106            let span_len = span_buf.len();
107
108            // A span that cannot fit on a line by itself is dropped rather than
109            // emitted as a truncated, unparseable line.
110            if span_len + TRACE_FORMAT_OVERHEAD > max_line_size {
111                stats.spans_dropped += 1;
112                tracing::debug!(
113                    span_len,
114                    max_line_size,
115                    "Span too large to send to logs, dropping"
116                );
117                continue;
118            }
119
120            // Flush the current line if appending this span would overflow.
121            // `line` already contains `TRACE_PREFIX`; the emitted line will also
122            // gain `TRACE_SUFFIX`, so account for both here.
123            let comma = usize::from(line_span_count > 0);
124            if line_span_count > 0
125                && line.len() + comma + span_len + TRACE_SUFFIX.len() > max_line_size
126            {
127                flush_line(out, &mut line)?;
128                line_span_count = 0;
129            }
130
131            if line_span_count > 0 {
132                line.push(b',');
133            }
134            line.extend_from_slice(&span_buf);
135            line_span_count += 1;
136            stats.spans_written += 1;
137        }
138    }
139
140    if line_span_count > 0 {
141        flush_line(out, &mut line)?;
142    }
143
144    Ok(stats)
145}
146
147/// Encodes a v1 [`TracerPayload`] into newline-delimited JSON "log exporter" lines,
148/// writing them to `out`. Same framing, packing, and oversized-span dropping behavior as
149/// [`encode_traces`] — see its docs — but takes v1
150/// [`TracerPayload`]/[`crate::span::v1::Span`] input and downgrades each span's unified
151/// attribute model back to the `meta`/`metrics`-shaped wire span via
152/// [`span_v1::LogSpanV1`], since the JSON log wire contract (consumed by the Datadog
153/// Forwarder Lambda) is unchanged by the v1 migration. Payload-level `env`/`app_version`/
154/// `attributes` are propagated into every span (lowest precedence, below chunk and span
155/// level) — same convention as the msgpack v0.4 downgrade encoder's
156/// `encode_payload_from_v1`.
157///
158/// # Errors
159///
160/// Returns any [`std::io::Error`] produced while writing to `out`.
161pub fn encode_traces_v1<T: TraceData>(
162    payload: &TracerPayload<T>,
163    out: &mut impl Write,
164    max_line_size: usize,
165) -> std::io::Result<EncodeStats> {
166    let mut stats = EncodeStats::default();
167
168    // Reusable buffers: `span_buf` holds the JSON for the span currently being
169    // considered; `line` accumulates the complete current line. It is primed
170    // with `TRACE_PREFIX` so that a single `write_all` per line can emit
171    // prefix + spans + suffix in one syscall (see `flush_line`).
172    let mut span_buf: Vec<u8> = Vec::with_capacity(512);
173    let mut line: Vec<u8> = Vec::with_capacity(max_line_size.min(64 * 1024));
174    line.extend_from_slice(TRACE_PREFIX);
175    let mut line_span_count: usize = 0;
176
177    for chunk in &payload.chunks {
178        let ctx = ChunkContextV1::new(
179            chunk,
180            &payload.env,
181            &payload.app_version,
182            &payload.attributes,
183        );
184        for span in &chunk.spans {
185            span_buf.clear();
186            serde_json::to_writer(&mut span_buf, &LogSpanV1(span, &ctx))
187                .map_err(std::io::Error::other)?;
188            let span_len = span_buf.len();
189
190            // A span that cannot fit on a line by itself is dropped rather than
191            // emitted as a truncated, unparseable line.
192            if span_len + TRACE_FORMAT_OVERHEAD > max_line_size {
193                stats.spans_dropped += 1;
194                tracing::debug!(
195                    span_len,
196                    max_line_size,
197                    "Span too large to send to logs, dropping"
198                );
199                continue;
200            }
201
202            // Flush the current line if appending this span would overflow.
203            // `line` already contains `TRACE_PREFIX`; the emitted line will also
204            // gain `TRACE_SUFFIX`, so account for both here.
205            let comma = usize::from(line_span_count > 0);
206            if line_span_count > 0
207                && line.len() + comma + span_len + TRACE_SUFFIX.len() > max_line_size
208            {
209                flush_line(out, &mut line)?;
210                line_span_count = 0;
211            }
212
213            if line_span_count > 0 {
214                line.push(b',');
215            }
216            line.extend_from_slice(&span_buf);
217            line_span_count += 1;
218            stats.spans_written += 1;
219        }
220    }
221
222    if line_span_count > 0 {
223        flush_line(out, &mut line)?;
224    }
225
226    Ok(stats)
227}
228
229/// Writes one complete line (`{"traces":[[` + joined spans + `]]}\n`) in a single
230/// `write_all`, then resets the line buffer (re-primed with `TRACE_PREFIX`) for
231/// reuse.
232///
233/// `line` is expected to already contain `TRACE_PREFIX` followed by the
234/// comma-joined spans; this appends `TRACE_SUFFIX` to complete the line.
235fn flush_line(out: &mut impl Write, line: &mut Vec<u8>) -> std::io::Result<()> {
236    line.extend_from_slice(TRACE_SUFFIX);
237    out.write_all(line)?;
238    line.clear();
239    line.extend_from_slice(TRACE_PREFIX);
240    Ok(())
241}
242
243#[cfg(test)]
244// `SpanSlice` fields are `Cow<'a, str>` (SliceData::Text), so the `"literal".into()`
245// conversions below are genuine `&str -> Cow` conversions required to compile. clippy on
246// the CI target nonetheless reports them as `useless_conversion` to `&str` (a false positive
247// not reproduced on all hosts); allow it here rather than dropping the necessary `.into()`.
248#[allow(clippy::useless_conversion)]
249mod tests {
250    use super::*;
251    use crate::span::v04::SpanSlice;
252    use serde_json::Value;
253
254    const MAX: usize = 64 * 1024;
255
256    fn lines(out: &[u8]) -> Vec<String> {
257        String::from_utf8(out.to_vec())
258            .unwrap()
259            .lines()
260            .map(|s| s.to_string())
261            .collect()
262    }
263
264    #[test]
265    fn golden_known_span() {
266        let span = SpanSlice {
267            service: "my-fn".into(),
268            name: "aws.lambda".into(),
269            resource: "my-fn".into(),
270            r#type: "serverless".into(),
271            trace_id: 1,
272            span_id: 2,
273            parent_id: 0,
274            start: 1717200000000000000,
275            duration: 1500000,
276            error: 0,
277            meta: [("env".into(), "prod".into())].into_iter().collect(),
278            metrics: [("_sampling_priority_v1".into(), 1.0)]
279                .into_iter()
280                .collect(),
281            ..Default::default()
282        };
283        let mut out = Vec::new();
284        let stats = encode_traces(&[vec![span]], &mut out, MAX).unwrap();
285        assert_eq!(stats.spans_written, 1);
286        assert_eq!(stats.spans_dropped, 0);
287
288        let expected = concat!(
289            "{\"traces\":[[",
290            "{\"trace_id\":\"0000000000000001\",",
291            "\"span_id\":\"0000000000000002\",",
292            "\"parent_id\":\"0000000000000000\",",
293            "\"service\":\"my-fn\",",
294            "\"name\":\"aws.lambda\",",
295            "\"resource\":\"my-fn\",",
296            "\"type\":\"serverless\",",
297            "\"error\":0,",
298            "\"start\":1717200000000000000,",
299            "\"duration\":1500000,",
300            "\"meta\":{\"env\":\"prod\"},",
301            "\"metrics\":{\"_sampling_priority_v1\":1.0}",
302            "}]]}\n",
303        );
304        assert_eq!(String::from_utf8(out).unwrap(), expected);
305    }
306
307    #[test]
308    fn meta_struct_is_omitted() {
309        let span = SpanSlice {
310            trace_id: 1,
311            span_id: 2,
312            meta_struct: [("_dd.appsec.json".into(), [0x81u8, 0xa4].as_slice())]
313                .into_iter()
314                .collect(),
315            ..Default::default()
316        };
317        let mut out = Vec::new();
318        encode_traces(&[vec![span]], &mut out, MAX).unwrap();
319        let text = String::from_utf8(out).unwrap();
320        assert!(
321            !text.contains("meta_struct"),
322            "meta_struct must not be emitted: {text}"
323        );
324        // Still a valid, Forwarder-parseable line.
325        let v: Value = serde_json::from_str(text.trim_end()).unwrap();
326        assert!(v["traces"][0][0]["trace_id"].is_string());
327    }
328
329    #[test]
330    fn hex_high_bits_and_root_parent() {
331        // trace_id with the high 64 bits set => 32 hex chars.
332        let trace_id: u128 = (0xABCDu128 << 64) | 0x1;
333        let span = SpanSlice {
334            trace_id,
335            span_id: 0xFF,
336            parent_id: 0,
337            ..Default::default()
338        };
339        let mut out = Vec::new();
340        encode_traces(&[vec![span]], &mut out, MAX).unwrap();
341        let text = String::from_utf8(out).unwrap();
342        assert!(
343            text.contains("\"trace_id\":\"000000000000abcd0000000000000001\""),
344            "got: {text}"
345        );
346        assert!(text.contains("\"span_id\":\"00000000000000ff\""));
347        assert!(text.contains("\"parent_id\":\"0000000000000000\""));
348    }
349
350    #[test]
351    fn error_is_integer() {
352        let span = SpanSlice {
353            error: 1,
354            ..Default::default()
355        };
356        let mut out = Vec::new();
357        encode_traces(&[vec![span]], &mut out, MAX).unwrap();
358        let text = String::from_utf8(out).unwrap();
359        assert!(text.contains("\"error\":1,"), "got: {text}");
360    }
361
362    #[test]
363    fn empty_maps_and_type_omitted() {
364        let span = SpanSlice {
365            name: "op".into(),
366            ..Default::default()
367        };
368        let mut out = Vec::new();
369        encode_traces(&[vec![span]], &mut out, MAX).unwrap();
370        let text = String::from_utf8(out).unwrap();
371        assert!(!text.contains("\"meta\""), "got: {text}");
372        assert!(!text.contains("\"metrics\""));
373        assert!(!text.contains("\"meta_struct\""));
374        assert!(!text.contains("\"span_links\""));
375        assert!(!text.contains("\"span_events\""));
376        assert!(!text.contains("\"type\""));
377    }
378
379    #[test]
380    fn string_escaping() {
381        let span = SpanSlice {
382            name: "say \"hi\"\n".into(),
383            ..Default::default()
384        };
385        let mut out = Vec::new();
386        encode_traces(&[vec![span]], &mut out, MAX).unwrap();
387        // Each line must remain valid JSON despite the embedded quote/newline.
388        for line in lines(&out) {
389            let parsed: Value = serde_json::from_str(&line).unwrap();
390            assert_eq!(
391                parsed["traces"][0][0]["name"].as_str().unwrap(),
392                "say \"hi\"\n"
393            );
394        }
395    }
396
397    #[test]
398    fn size_cap_batches_into_multiple_lines() {
399        // Build several spans that individually fit but together exceed a small cap.
400        let make = |id: u64| SpanSlice {
401            name: "x".into(),
402            span_id: id,
403            ..Default::default()
404        };
405        let trace: Vec<SpanSlice> = (1..=6).map(make).collect();
406
407        // Determine one span's serialized length to size the cap so ~2 fit per line.
408        let mut one = Vec::new();
409        encode_traces(&[vec![make(1)]], &mut one, MAX).unwrap();
410        // `one` = prefix + span + suffix. The bare span length:
411        let span_len = one.len() - TRACE_FORMAT_OVERHEAD;
412        // Cap fits two spans + a comma but not three.
413        let cap = TRACE_FORMAT_OVERHEAD + span_len * 2 + 1;
414
415        let mut out = Vec::new();
416        let stats = encode_traces(&[trace], &mut out, cap).unwrap();
417        assert_eq!(stats.spans_written, 6);
418        assert_eq!(stats.spans_dropped, 0);
419
420        let emitted = lines(&out);
421        assert_eq!(emitted.len(), 3, "expected 3 lines, got {emitted:?}");
422        for line in &emitted {
423            assert!(line.len() <= cap, "line over cap: {} > {cap}", line.len());
424            let parsed: Value = serde_json::from_str(line).unwrap();
425            assert_eq!(parsed["traces"][0].as_array().unwrap().len(), 2);
426        }
427    }
428
429    #[test]
430    fn oversize_single_span_dropped() {
431        let big = "a".repeat(10_000);
432        let span = SpanSlice {
433            name: big.as_str().into(),
434            span_id: 1,
435            ..Default::default()
436        };
437        let small = SpanSlice {
438            name: "ok".into(),
439            span_id: 2,
440            ..Default::default()
441        };
442        let mut out = Vec::new();
443        // Cap large enough for `small` but far too small for `big`.
444        let stats = encode_traces(&[vec![span, small]], &mut out, 1024).unwrap();
445        assert_eq!(stats.spans_dropped, 1);
446        assert_eq!(stats.spans_written, 1);
447
448        let emitted = lines(&out);
449        assert_eq!(emitted.len(), 1);
450        let parsed: Value = serde_json::from_str(&emitted[0]).unwrap();
451        assert_eq!(parsed["traces"][0][0]["name"].as_str().unwrap(), "ok");
452    }
453
454    #[test]
455    fn metric_non_finite_serializes_null() {
456        // serde_json renders non-finite f64 as JSON null; the line must remain
457        // parseable and the metric values must be null (not NaN/Infinity tokens).
458        let span = SpanSlice {
459            span_id: 1,
460            metrics: [("nan".into(), f64::NAN), ("inf".into(), f64::INFINITY)]
461                .into_iter()
462                .collect(),
463            ..Default::default()
464        };
465        let mut out = Vec::new();
466        encode_traces(&[vec![span]], &mut out, MAX).unwrap();
467        let emitted = lines(&out);
468        assert_eq!(emitted.len(), 1);
469        let parsed: Value = serde_json::from_str(&emitted[0]).unwrap();
470        let metrics = &parsed["traces"][0][0]["metrics"];
471        assert!(metrics["nan"].is_null(), "got: {metrics}");
472        assert!(metrics["inf"].is_null(), "got: {metrics}");
473    }
474
475    #[test]
476    fn multi_trace_flattened_into_one_line() {
477        // Two input traces, both well under the cap, flatten into a single line
478        // whose single inner trace array holds both spans.
479        let span_a = SpanSlice {
480            name: "a".into(),
481            span_id: 1,
482            ..Default::default()
483        };
484        let span_b = SpanSlice {
485            name: "b".into(),
486            span_id: 2,
487            ..Default::default()
488        };
489        let mut out = Vec::new();
490        let stats = encode_traces(&[vec![span_a], vec![span_b]], &mut out, MAX).unwrap();
491        assert_eq!(stats.spans_written, 2);
492        let emitted = lines(&out);
493        assert_eq!(emitted.len(), 1, "expected one line, got {emitted:?}");
494        let parsed: Value = serde_json::from_str(&emitted[0]).unwrap();
495        let inner = parsed["traces"][0].as_array().unwrap();
496        assert_eq!(inner.len(), 2);
497        assert_eq!(parsed["traces"].as_array().unwrap().len(), 1);
498    }
499
500    #[test]
501    fn span_links_and_events_emitted_when_present() {
502        use crate::span::v04::{SpanEvent, SpanLink};
503
504        let span = SpanSlice {
505            span_id: 1,
506            span_links: vec![SpanLink {
507                trace_id: 7,
508                span_id: 8,
509                ..Default::default()
510            }],
511            span_events: vec![SpanEvent {
512                time_unix_nano: 123,
513                name: "evt".into(),
514                ..Default::default()
515            }],
516            ..Default::default()
517        };
518        let mut out = Vec::new();
519        encode_traces(&[vec![span]], &mut out, MAX).unwrap();
520        let emitted = lines(&out);
521        assert_eq!(emitted.len(), 1);
522        let parsed: Value = serde_json::from_str(&emitted[0]).unwrap();
523        let span_json = &parsed["traces"][0][0];
524        // Lock the inner wire shape (field names/values), not just presence.
525        assert_eq!(span_json["span_links"][0]["trace_id"], 7);
526        assert_eq!(span_json["span_links"][0]["span_id"], 8);
527        assert_eq!(span_json["span_events"][0]["name"], "evt");
528        assert_eq!(span_json["span_events"][0]["time_unix_nano"], 123);
529    }
530
531    #[test]
532    fn span_link_flags_sentinel_bit_masked() {
533        // The internal "explicitly set" sentinel (bit 31) must never appear in the emitted
534        // JSON log, which downstream consumers treat as the real W3C trace-flags value.
535        // Covers both sentinel states: kept (0x8000_0001) and explicitly dropped (0x8000_0000).
536        use crate::span::v04::SpanLink;
537
538        fn encoded_flags(flags: u32) -> Value {
539            let span = SpanSlice {
540                span_id: 1,
541                span_links: vec![SpanLink {
542                    trace_id: 7,
543                    span_id: 8,
544                    flags,
545                    ..Default::default()
546                }],
547                ..Default::default()
548            };
549            let mut out = Vec::new();
550            encode_traces(&[vec![span]], &mut out, MAX).unwrap();
551            let emitted = lines(&out);
552            let parsed: Value = serde_json::from_str(&emitted[0]).unwrap();
553            parsed["traces"][0][0]["span_links"][0]["flags"].clone()
554        }
555
556        assert_eq!(
557            encoded_flags(0x8000_0001),
558            1,
559            "the sentinel bit must not leak into the JSON log"
560        );
561        assert_eq!(
562            encoded_flags(0x8000_0000),
563            0,
564            "an explicit drop decision must still emit flags: 0, not omit the field"
565        );
566    }
567
568    #[test]
569    fn empty_inner_trace_writes_nothing() {
570        // One trace containing zero spans: nothing is emitted and no spans counted.
571        let traces: Vec<Vec<SpanSlice>> = vec![vec![]];
572        let mut out = Vec::new();
573        let stats = encode_traces(&traces, &mut out, MAX).unwrap();
574        assert!(out.is_empty());
575        assert_eq!(stats.spans_written, 0);
576        assert_eq!(stats.spans_dropped, 0);
577    }
578
579    #[test]
580    fn empty_input_writes_nothing() {
581        let traces: Vec<Vec<SpanSlice>> = vec![];
582        let mut out = Vec::new();
583        let stats = encode_traces(&traces, &mut out, MAX).unwrap();
584        assert_eq!(stats, EncodeStats::default());
585        assert!(out.is_empty());
586    }
587
588    // Mirrors the Datadog Forwarder `is_trace` detection contract: every line
589    // parses as JSON, top-level `traces` is a non-empty array whose `[0]` is a
590    // non-empty array whose `[0]` has a non-null string `trace_id`, and every
591    // line ends with `\n`.
592    #[test]
593    fn forwarder_is_trace_contract() {
594        let trace: Vec<SpanSlice> = (1u64..=5)
595            .map(|i| SpanSlice {
596                service: "svc".into(),
597                name: "op".into(),
598                trace_id: u128::from(i),
599                span_id: i + 100,
600                ..Default::default()
601            })
602            .collect();
603
604        let mut out = Vec::new();
605        encode_traces(&[trace], &mut out, 200).unwrap();
606
607        let text = String::from_utf8(out).unwrap();
608        assert!(!text.is_empty());
609        // Every line is newline-terminated.
610        for chunk in text.split_inclusive('\n') {
611            assert!(chunk.ends_with('\n'), "line not newline-terminated");
612            let line = chunk.trim_end_matches('\n');
613            let parsed: Value = serde_json::from_str(line).unwrap();
614
615            let traces = parsed.get("traces").and_then(Value::as_array).unwrap();
616            assert!(!traces.is_empty());
617            let first = traces[0].as_array().unwrap();
618            assert!(!first.is_empty());
619            let trace_id = first[0].get("trace_id").unwrap();
620            assert!(trace_id.is_string());
621            assert!(!trace_id.as_str().unwrap().is_empty());
622        }
623    }
624}
libdd_trace_utils/json_log_encoder/mod.rs

libdd_trace_utils/json_log_encoder/
mod.rs
