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

saluki_components/decoders/datadog/
mod.rs
