saluki_components/decoders/datadog/
mod.rs

1//! Decoder for the Datadog v1.0 APM trace wire format (also called the `idx`/`etp` format).
2//!
3//! The v1.0 trace API (`Content-Type: application/msgpack`, endpoint `v1.0`) sends a tracer payload
4//! in a custom MessagePack layout rather than protobuf-over-msgpack. Its defining traits:
5//!
6//! - A **streaming string table**: a string appears once as a literal, then by `uint32` index
7//!   thereafter.
8//! - **Structs are maps keyed by field number** (`uint32`), not by name.
9//! - **Attribute maps are flat arrays** with three slots per entry (`key`, `type`, `value`).
10//! - **`AnyValue`** is a `uint32` type discriminant followed by the value.
11//!
12//! [`decode_v1_payload`](datadog::decode_v1_payload) decodes one tracer payload into the unified
13//! `Trace` model, producing one `Trace` per trace chunk. String references are resolved to owned
14//! strings during decode.
15
16use std::sync::Arc;
17
18use saluki_common::collections::FastHashMap;
19use saluki_core::data_model::event::trace::{
20    AttributeValue, PayloadFields, Span, SpanEvent, SpanLink, Trace, TraceIngestInfo,
21};
22use stringtheory::MetaString;
23
24mod error;
25mod read;
26mod string_table;
27mod value;
28
29#[cfg(test)]
30mod tests;
31
32pub use self::error::DecodeError;
33use self::string_table::StringTable;
34
35/// Plausible minimum average wire size of a trace chunk containing a span list and 16-byte trace ID.
36const MIN_BYTES_PER_TRACE_CHUNK: u32 = 22;
37
38/// Plausible minimum average wire size of a span containing a random span ID and epoch-nanosecond start time.
39const MIN_BYTES_PER_SPAN: u32 = 21;
40
41/// Decodes a single v1.0 tracer payload into unified trace events.
42///
43/// Each trace chunk in the payload becomes one [`Trace`]. Payload-level metadata (container ID,
44/// tracer language, environment, and so on) is cloned onto every trace, and payload-level
45/// attributes are merged into each trace's attributes with per-chunk attributes taking precedence.
46///
47/// # Errors
48///
49/// Returns a [`DecodeError`] if the input is not a well-formed v1.0 tracer payload (truncated,
50/// malformed MessagePack, oversized headers, out-of-range string references, and so on).
51pub fn decode_v1_payload(bytes: &[u8]) -> Result<Vec<Trace>, DecodeError> {
52    let mut reader = bytes;
53    let payload = decode_tracer_payload(&mut reader)?;
54    if !reader.is_empty() {
55        return Err(DecodeError::TrailingBytes { len: reader.len() });
56    }
57
58    let payload_fields = PayloadFields {
59        container_id: payload.container_id,
60        language_name: payload.language_name,
61        language_version: payload.language_version,
62        tracer_version: payload.tracer_version,
63        runtime_id: payload.runtime_id,
64        env: payload.env,
65        hostname: payload.hostname,
66        app_version: payload.app_version,
67        client_dropped_p0s_weight: 0.0,
68    };
69
70    let mut traces = Vec::with_capacity(payload.chunks.len());
71    for chunk in payload.chunks {
72        // Payload-level attributes form the base; per-chunk attributes override on key conflict.
73        let mut attributes = payload.attributes.clone();
74        attributes.extend(chunk.attributes);
75
76        let mut trace = Trace::new(chunk.spans, TraceIngestInfo::Datadog);
77        trace.trace_id_high = chunk.trace_id_high;
78        trace.trace_id_low = chunk.trace_id_low;
79        trace.origin = chunk.origin;
80        trace.payload = payload_fields.clone();
81        trace.attributes = Arc::new(attributes);
82        trace.priority = chunk.priority;
83        trace.dropped_trace = chunk.dropped_trace;
84        trace.sampling_mechanism = chunk.sampling_mechanism;
85        traces.push(trace);
86    }
87
88    Ok(traces)
89}
90
91/// Intermediate representation of a decoded tracer payload before chunks are lifted into `Trace`s.
92struct DecodedPayload {
93    container_id: MetaString,
94    language_name: MetaString,
95    language_version: MetaString,
96    tracer_version: MetaString,
97    runtime_id: MetaString,
98    env: MetaString,
99    hostname: MetaString,
100    app_version: MetaString,
101    attributes: FastHashMap<MetaString, AttributeValue>,
102    chunks: Vec<DecodedChunk>,
103}
104
105/// Intermediate representation of a decoded trace chunk.
106struct DecodedChunk {
107    priority: Option<i32>,
108    origin: MetaString,
109    attributes: FastHashMap<MetaString, AttributeValue>,
110    spans: Vec<Span>,
111    dropped_trace: bool,
112    trace_id_high: u64,
113    trace_id_low: u64,
114    sampling_mechanism: u32,
115}
116
117/// Splits a big-endian trace ID byte string into its high and low 64-bit halves.
118///
119/// IDs longer than 16 bytes use the final 16 bytes, matching `normalizeTraceChunkV1` in the
120/// reference decoder; shorter IDs are right-aligned (high-order bytes treated as zero).
121fn split_trace_id(id: &[u8]) -> (u64, u64) {
122    let mut buf = [0u8; 16];
123    if id.len() >= 16 {
124        buf.copy_from_slice(&id[id.len() - 16..]);
125    } else {
126        buf[16 - id.len()..].copy_from_slice(id);
127    }
128    let high = u64::from_be_bytes(buf[0..8].try_into().unwrap());
129    let low = u64::from_be_bytes(buf[8..16].try_into().unwrap());
130    (high, low)
131}
132
133/// Decodes the top-level `TracerPayload` map.
134fn decode_tracer_payload(r: &mut &[u8]) -> Result<DecodedPayload, DecodeError> {
135    let mut strings = StringTable::new();
136    let mut payload = DecodedPayload {
137        container_id: MetaString::empty(),
138        language_name: MetaString::empty(),
139        language_version: MetaString::empty(),
140        tracer_version: MetaString::empty(),
141        runtime_id: MetaString::empty(),
142        env: MetaString::empty(),
143        hostname: MetaString::empty(),
144        app_version: MetaString::empty(),
145        attributes: FastHashMap::default(),
146        chunks: Vec::new(),
147    };
148
149    let num_fields = read::read_map_len(r, "tracer payload")?;
150    for _ in 0..num_fields {
151        let field = read::read_u32(r, "tracer payload field")?;
152        match field {
153            1 => {
154                // The string table must arrive first so later references resolve. Anything beyond
155                // the seeded empty string means fields were decoded before it.
156                if strings.len() > 1 {
157                    return Err(DecodeError::StringsNotFirst);
158                }
159                decode_string_table(r, &mut strings)?;
160            }
161            2 => payload.container_id = value::read_streaming_string(r, &mut strings, "container ID")?,
162            3 => payload.language_name = value::read_streaming_string(r, &mut strings, "language name")?,
163            4 => payload.language_version = value::read_streaming_string(r, &mut strings, "language version")?,
164            5 => payload.tracer_version = value::read_streaming_string(r, &mut strings, "tracer version")?,
165            6 => payload.runtime_id = value::read_streaming_string(r, &mut strings, "runtime ID")?,
166            7 => payload.env = value::read_streaming_string(r, &mut strings, "env")?,
167            8 => payload.hostname = value::read_streaming_string(r, &mut strings, "hostname")?,
168            9 => payload.app_version = value::read_streaming_string(r, &mut strings, "app version")?,
169            10 => payload.attributes = value::read_attributes_map(r, &mut strings, "tracer payload attributes")?,
170            11 => payload.chunks = decode_chunk_list(r, &mut strings)?,
171            _ => value::harvest_unknown_field(r, &mut strings, "tracer payload field")?,
172        }
173    }
174
175    Ok(payload)
176}
177
178/// Decodes the string table array (field 1 of the tracer payload).
179///
180/// Empty strings are skipped: index 0 is always the pre-seeded empty string, matching the reference
181/// encoder which never emits duplicate or additional empty strings.
182fn decode_string_table(r: &mut &[u8], strings: &mut StringTable) -> Result<(), DecodeError> {
183    let num_strings = read::read_array_len(r, "string table")?;
184    for _ in 0..num_strings {
185        let s = read::read_str(r, "string table entry")?;
186        if s.is_empty() {
187            continue;
188        }
189        strings.add(s);
190    }
191    Ok(())
192}
193
194/// Decodes the list of trace chunks (field 11 of the tracer payload).
195fn decode_chunk_list(r: &mut &[u8], strings: &mut StringTable) -> Result<Vec<DecodedChunk>, DecodeError> {
196    let num_chunks = read::read_array_len_with_minimum(r, MIN_BYTES_PER_TRACE_CHUNK, "trace chunk list")?;
197    let mut chunks = Vec::with_capacity(num_chunks as usize);
198    for _ in 0..num_chunks {
199        chunks.push(decode_chunk(r, strings)?);
200    }
201    Ok(chunks)
202}
203
204/// Decodes a single `TraceChunk` map.
205fn decode_chunk(r: &mut &[u8], strings: &mut StringTable) -> Result<DecodedChunk, DecodeError> {
206    let mut priority = None;
207    let mut origin = MetaString::empty();
208    let mut attributes = FastHashMap::default();
209    let mut spans = Vec::new();
210    let mut dropped_trace = false;
211    let mut trace_id_high = 0;
212    let mut trace_id_low = 0;
213    let mut sampling_mechanism = 0;
214
215    let num_fields = read::read_map_len(r, "trace chunk")?;
216    for _ in 0..num_fields {
217        let field = read::read_u32(r, "trace chunk field")?;
218        match field {
219            1 => priority = Some(read::read_i32(r, "trace chunk priority")?),
220            2 => origin = value::read_streaming_string(r, strings, "trace chunk origin")?,
221            3 => attributes = value::read_attributes_map(r, strings, "trace chunk attributes")?,
222            4 => spans = decode_span_list(r, strings)?,
223            5 => dropped_trace = read::read_bool(r, "trace chunk droppedTrace")?,
224            6 => {
225                let id = read::read_bytes(r, "trace chunk traceID")?;
226                (trace_id_high, trace_id_low) = split_trace_id(&id);
227            }
228            7 => sampling_mechanism = read::read_u32(r, "trace chunk samplingMechanism")?,
229            _ => value::harvest_unknown_field(r, strings, "trace chunk field")?,
230        }
231    }
232
233    Ok(DecodedChunk {
234        priority,
235        origin,
236        attributes,
237        spans,
238        dropped_trace,
239        trace_id_high,
240        trace_id_low,
241        sampling_mechanism,
242    })
243}
244
245/// Decodes the list of spans within a trace chunk (field 4).
246fn decode_span_list(r: &mut &[u8], strings: &mut StringTable) -> Result<Vec<Span>, DecodeError> {
247    let num_spans = read::read_array_len_with_minimum(r, MIN_BYTES_PER_SPAN, "span list")?;
248    let mut spans = Vec::with_capacity(num_spans as usize);
249    for _ in 0..num_spans {
250        spans.push(decode_span(r, strings)?);
251    }
252    Ok(spans)
253}
254
255/// Decodes a single `Span` map.
256fn decode_span(r: &mut &[u8], strings: &mut StringTable) -> Result<Span, DecodeError> {
257    let mut service = MetaString::empty();
258    let mut name = MetaString::empty();
259    let mut resource = MetaString::empty();
260    let mut span_id = 0;
261    let mut parent_id = 0;
262    let mut start = 0;
263    let mut duration = 0;
264    let mut error = 0;
265    let mut attributes = FastHashMap::default();
266    let mut span_type = MetaString::empty();
267    let mut links = Vec::new();
268    let mut events = Vec::new();
269    let mut env = MetaString::empty();
270    let mut version = MetaString::empty();
271    let mut component = MetaString::empty();
272    let mut kind = 0;
273
274    let num_fields = read::read_map_len(r, "span")?;
275    for _ in 0..num_fields {
276        let field = read::read_u32(r, "span field")?;
277        match field {
278            1 => service = value::read_streaming_string(r, strings, "span service")?,
279            2 => name = value::read_streaming_string(r, strings, "span name")?,
280            3 => resource = value::read_streaming_string(r, strings, "span resource")?,
281            4 => span_id = read::read_u64(r, "span spanID")?,
282            5 => parent_id = read::read_u64(r, "span parentID")?,
283            6 => start = read::read_u64(r, "span start")?,
284            7 => duration = read::read_u64(r, "span duration")?,
285            8 => error = i32::from(read::read_bool(r, "span error")?),
286            9 => attributes = value::read_attributes_map(r, strings, "span attributes")?,
287            10 => span_type = value::read_streaming_string(r, strings, "span type")?,
288            11 => links = decode_span_link_list(r, strings)?,
289            12 => events = decode_span_event_list(r, strings)?,
290            13 => env = value::read_streaming_string(r, strings, "span env")?,
291            14 => version = value::read_streaming_string(r, strings, "span version")?,
292            15 => component = value::read_streaming_string(r, strings, "span component")?,
293            16 => kind = read::read_u32(r, "span kind")?,
294            _ => value::harvest_unknown_field(r, strings, "span field")?,
295        }
296    }
297
298    Ok(Span::new(
299        service, name, resource, span_type, span_id, parent_id, start, duration, error,
300    )
301    .with_attributes(attributes)
302    .with_span_links(links)
303    .with_span_events(events)
304    .with_env(env)
305    .with_version(version)
306    .with_component(component)
307    .with_kind(kind))
308}
309
310/// Decodes the list of span links within a span (field 11).
311fn decode_span_link_list(r: &mut &[u8], strings: &mut StringTable) -> Result<Vec<SpanLink>, DecodeError> {
312    let num_links = read::read_array_len(r, "span link list")?;
313    let mut links = Vec::with_capacity(num_links as usize);
314    for _ in 0..num_links {
315        links.push(decode_span_link(r, strings)?);
316    }
317    Ok(links)
318}
319
320/// Decodes a single `SpanLink` map.
321fn decode_span_link(r: &mut &[u8], strings: &mut StringTable) -> Result<SpanLink, DecodeError> {
322    let mut trace_id_high = 0;
323    let mut trace_id_low = 0;
324    let mut span_id = 0;
325    let mut attributes = FastHashMap::default();
326    let mut tracestate = MetaString::empty();
327    let mut flags = 0;
328
329    let num_fields = read::read_map_len(r, "span link")?;
330    for _ in 0..num_fields {
331        let field = read::read_u32(r, "span link field")?;
332        match field {
333            1 => {
334                let id = read::read_bytes(r, "span link traceID")?;
335                (trace_id_high, trace_id_low) = split_trace_id(&id);
336            }
337            2 => span_id = read::read_u64(r, "span link spanID")?,
338            3 => attributes = value::read_attributes_map(r, strings, "span link attributes")?,
339            4 => tracestate = value::read_streaming_string(r, strings, "span link tracestate")?,
340            5 => flags = read::read_u32(r, "span link flags")?,
341            _ => value::harvest_unknown_field(r, strings, "span link field")?,
342        }
343    }
344
345    Ok(SpanLink::new(trace_id_low, span_id)
346        .with_trace_id_high(trace_id_high)
347        .with_attributes(attributes)
348        .with_tracestate(tracestate)
349        .with_flags(flags))
350}
351
352/// Decodes the list of span events within a span (field 12).
353fn decode_span_event_list(r: &mut &[u8], strings: &mut StringTable) -> Result<Vec<SpanEvent>, DecodeError> {
354    let num_events = read::read_array_len(r, "span event list")?;
355    let mut events = Vec::with_capacity(num_events as usize);
356    for _ in 0..num_events {
357        events.push(decode_span_event(r, strings)?);
358    }
359    Ok(events)
360}
361
362/// Decodes a single `SpanEvent` map.
363fn decode_span_event(r: &mut &[u8], strings: &mut StringTable) -> Result<SpanEvent, DecodeError> {
364    let mut time = 0;
365    let mut name = MetaString::empty();
366    let mut attributes = FastHashMap::default();
367
368    let num_fields = read::read_map_len(r, "span event")?;
369    for _ in 0..num_fields {
370        let field = read::read_u32(r, "span event field")?;
371        match field {
372            1 => time = read::read_u64(r, "span event time")?,
373            2 => name = value::read_streaming_string(r, strings, "span event name")?,
374            3 => attributes = value::read_attributes_map(r, strings, "span event attributes")?,
375            _ => value::harvest_unknown_field(r, strings, "span event field")?,
376        }
377    }
378
379    Ok(SpanEvent::new(time, name).with_attributes(attributes))
380}
saluki_components/decoders/datadog/mod.rs

saluki_components/decoders/datadog/
mod.rs
