1mod span_v04;
5mod span_v1;
6
7use crate::span::v04::Span;
8use crate::span::v1::TracerPayload;
9use crate::span::TraceData;
10use crate::tracer_metadata::TracerMetadata;
11use libdd_common::ResultInfallibleExt;
12use rmp::encode::{
13 write_array_len, write_bin, write_map_len, write_sint, write_str, write_uint, write_uint8,
14 ByteBuf, RmpWrite, ValueWriteError,
15};
16use std::borrow::Borrow;
17use std::collections::HashMap;
18
19mod trace_key {
21 pub const LANGUAGE_NAME: u8 = 3;
22 pub const LANGUAGE_VERSION: u8 = 4;
23 pub const TRACER_VERSION: u8 = 5;
24 pub const RUNTIME_ID: u8 = 6;
25 pub const ENV_REF: u8 = 7;
26 pub const HOSTNAME_REF: u8 = 8;
27 pub const APP_VERSION_REF: u8 = 9;
28 pub const ATTRIBUTES: u8 = 10;
30 pub const CHUNKS: u8 = 11;
31}
32
33mod chunk_key {
35 pub const PRIORITY: u8 = 1;
36 pub const ORIGIN: u8 = 2;
37 pub const ATTRIBUTES: u8 = 3;
38 pub const SPANS: u8 = 4;
39 pub const DROPPED_TRACE: u8 = 5;
40 pub const TRACE_ID: u8 = 6;
41 pub const SAMPLING_MECHANISM: u8 = 7;
43}
44
45#[repr(u8)]
47pub(super) enum SpanKey {
48 Service = 1,
49 Name = 2,
50 Resource = 3,
51 SpanId = 4,
52 ParentId = 5,
53 Start = 6,
54 Duration = 7,
55 Error = 8,
56 Attributes = 9,
57 Type = 10,
58 SpanLinks = 11,
59 SpanEvents = 12,
60 Env = 13,
61 Version = 14,
62 Component = 15,
63 Kind = 16,
64}
65
66#[repr(u8)]
68pub(super) enum SpanLinkKey {
69 TraceId = 1,
70 SpanId = 2,
71 Attributes = 3,
72 TraceState = 4,
73 Flags = 5,
74}
75
76#[repr(u8)]
78pub(super) enum SpanEventKey {
79 Time = 1,
80 Name = 2,
81 Attributes = 3,
82}
83
84#[repr(u8)]
87pub(super) enum AnyValueKey {
88 String = 1,
89 Bool = 2,
90 Double = 3,
91 Int64 = 4,
92 Bytes = 5,
93 Array = 6,
94 KeyValueList = 7,
95}
96
97pub(super) const TYPED_VALUE_STRIDE: u32 = 2;
100
101pub(super) const FLAT_ATTR_STRIDE: u32 = 3;
105
106pub(crate) struct StringTable {
115 seen: HashMap<String, u32>,
116}
117
118impl StringTable {
119 fn new() -> Self {
120 let mut seen = HashMap::new();
121 seen.insert(String::new(), 0);
122 Self { seen }
123 }
124
125 pub(crate) fn write_interned<W: RmpWrite, S: AsRef<str>>(
130 &mut self,
131 writer: &mut W,
132 s: S,
133 ) -> Result<(), ValueWriteError<W::Error>> {
134 let s = s.as_ref();
135 if let Some(&id) = self.seen.get(s) {
136 write_uint(writer, id as u64)?;
137 } else {
138 let id = self.seen.len() as u32;
139 self.seen.insert(s.to_string(), id);
140 write_str(writer, s)?;
141 }
142 Ok(())
143 }
144}
145
146pub(super) fn normalize_span_start(start: i64) -> u64 {
151 if start < 0 {
152 std::time::SystemTime::now()
153 .duration_since(std::time::UNIX_EPOCH)
154 .map(|d| d.as_nanos() as u64)
155 .unwrap_or(0)
156 } else {
157 start as u64
158 }
159}
160
161struct PayloadAttrs<'a> {
163 env: Option<&'a str>,
164 hostname: Option<&'a str>,
165 app_version: Option<&'a str>,
166 apm_mode: Option<&'a str>,
168 git_commit_sha: Option<&'a str>,
170}
171
172fn extract_payload_attrs<'a, T: TraceData + 'a, S: AsRef<[Span<T>]>>(
173 traces: &'a [S],
174 metadata: &'a TracerMetadata,
175) -> PayloadAttrs<'a>
176where
177 T::Text: 'a,
178{
179 let mut env = (!metadata.env.is_empty()).then_some(metadata.env.as_str());
183 let mut hostname = (!metadata.hostname.is_empty()).then_some(metadata.hostname.as_str());
184 let mut app_version =
185 (!metadata.app_version.is_empty()).then_some(metadata.app_version.as_str());
186 let mut git_commit_sha =
187 (!metadata.git_commit_sha.is_empty()).then_some(metadata.git_commit_sha.as_str());
188 let mut apm_mode = None;
189
190 'outer: for trace in traces {
191 for span in trace.as_ref() {
192 if env.is_none() {
193 env = span.meta.get("env").map(|v| v.borrow());
194 }
195 if hostname.is_none() {
196 hostname = span.meta.get("_dd.hostname").map(|v| v.borrow());
197 }
198 if app_version.is_none() {
199 app_version = span.meta.get("version").map(|v| v.borrow());
200 }
201 if apm_mode.is_none() {
202 apm_mode = span.meta.get("_dd.apm_mode").map(|v| v.borrow());
203 }
204 if git_commit_sha.is_none() {
205 git_commit_sha = span.meta.get("_dd.git.commit.sha").map(|v| v.borrow());
206 }
207 if env.is_some()
208 && hostname.is_some()
209 && app_version.is_some()
210 && apm_mode.is_some()
211 && git_commit_sha.is_some()
212 {
213 break 'outer;
214 }
215 }
216 }
217
218 PayloadAttrs {
219 env,
220 hostname,
221 app_version,
222 apm_mode,
223 git_commit_sha,
224 }
225}
226
227struct ChunkAttrs<'a> {
229 trace_id: u128,
231 sampling_priority: Option<i32>,
233 origin: Option<&'a str>,
235 sampling_mechanism: Option<u32>,
237}
238
239fn extract_chunk_attrs<'a, T: TraceData>(spans: &'a [Span<T>]) -> ChunkAttrs<'a>
240where
241 T::Text: 'a,
242{
243 let trace_id = spans
246 .first()
247 .map(|s| {
248 let high = s
249 .meta
250 .get("_dd.p.tid")
251 .and_then(|v| u64::from_str_radix(v.borrow(), 16).ok())
252 .unwrap_or(0);
253 ((high as u128) << 64) | s.trace_id
254 })
255 .unwrap_or(0);
256
257 let mut sampling_priority = None;
258 let mut origin = None;
259 let mut sampling_mechanism = None;
260
261 for span in spans {
262 let is_root =
264 span.parent_id == 0 || span.metrics.get("_dd.top_level").copied().unwrap_or(0.0) == 1.0;
265
266 if is_root {
267 sampling_priority = span.metrics.get("_sampling_priority_v1").map(|v| *v as i32);
270 origin = span.meta.get("_dd.origin").map(|v| v.borrow());
271 sampling_mechanism = span
274 .meta
275 .get("_dd.p.dm")
276 .and_then(|v| v.borrow().parse::<i32>().ok())
277 .map(|dm| dm.unsigned_abs());
278 break;
279 }
280
281 if sampling_priority.is_none() {
284 sampling_priority = span.metrics.get("_sampling_priority_v1").map(|v| *v as i32);
285 }
286 if origin.is_none() {
287 origin = span.meta.get("_dd.origin").map(|v| v.borrow());
288 }
289 if sampling_mechanism.is_none() {
290 sampling_mechanism = span
291 .meta
292 .get("_dd.p.dm")
293 .and_then(|v| v.borrow().parse::<i32>().ok())
294 .map(|dm| dm.unsigned_abs());
295 }
296 }
297
298 ChunkAttrs {
299 trace_id,
300 sampling_priority,
301 origin,
302 sampling_mechanism,
303 }
304}
305
306fn encode_payload<W: RmpWrite, T: TraceData, S: AsRef<[Span<T>]>>(
319 writer: &mut W,
320 traces: &[S],
321 metadata: &TracerMetadata,
322) -> Result<(), ValueWriteError<W::Error>> {
323 let mut table = StringTable::new();
324 let payload_attrs = extract_payload_attrs(traces, metadata);
325
326 let attr_count =
327 payload_attrs.apm_mode.is_some() as u32 + payload_attrs.git_commit_sha.is_some() as u32;
328 let has_attributes = attr_count > 0;
329
330 let map_len = 1u32 + (!metadata.language.is_empty()) as u32
332 + (!metadata.language_version.is_empty()) as u32
333 + (!metadata.tracer_version.is_empty()) as u32
334 + (!metadata.runtime_id.is_empty()) as u32
335 + payload_attrs.env.is_some() as u32
336 + payload_attrs.hostname.is_some() as u32
337 + payload_attrs.app_version.is_some() as u32
338 + has_attributes as u32;
339
340 write_map_len(writer, map_len)?;
341
342 write_uint8(writer, trace_key::CHUNKS)?;
343 write_array_len(writer, traces.len() as u32)?;
344 for trace in traces {
345 encode_chunk(writer, trace.as_ref(), &mut table)?;
346 }
347
348 if !metadata.language.is_empty() {
349 write_uint8(writer, trace_key::LANGUAGE_NAME)?;
350 table.write_interned(writer, &metadata.language)?;
351 }
352
353 if !metadata.language_version.is_empty() {
354 write_uint8(writer, trace_key::LANGUAGE_VERSION)?;
355 table.write_interned(writer, &metadata.language_version)?;
356 }
357
358 if !metadata.tracer_version.is_empty() {
359 write_uint8(writer, trace_key::TRACER_VERSION)?;
360 table.write_interned(writer, &metadata.tracer_version)?;
361 }
362
363 if !metadata.runtime_id.is_empty() {
364 write_uint8(writer, trace_key::RUNTIME_ID)?;
365 table.write_interned(writer, &metadata.runtime_id)?;
366 }
367
368 if let Some(env) = payload_attrs.env {
369 write_uint8(writer, trace_key::ENV_REF)?;
370 table.write_interned(writer, env)?;
371 }
372
373 if let Some(hostname) = payload_attrs.hostname {
374 write_uint8(writer, trace_key::HOSTNAME_REF)?;
375 table.write_interned(writer, hostname)?;
376 }
377
378 if let Some(app_version) = payload_attrs.app_version {
379 write_uint8(writer, trace_key::APP_VERSION_REF)?;
380 table.write_interned(writer, app_version)?;
381 }
382
383 if has_attributes {
384 write_uint8(writer, trace_key::ATTRIBUTES)?;
387 write_array_len(writer, attr_count * FLAT_ATTR_STRIDE)?;
388 if let Some(v) = payload_attrs.apm_mode {
389 table.write_interned(writer, "_dd.apm_mode")?;
390 write_uint8(writer, AnyValueKey::String as u8)?;
391 table.write_interned(writer, v)?;
392 }
393 if let Some(v) = payload_attrs.git_commit_sha {
394 table.write_interned(writer, "_dd.git.commit.sha")?;
395 write_uint8(writer, AnyValueKey::String as u8)?;
396 table.write_interned(writer, v)?;
397 }
398 }
399
400 Ok(())
401}
402
403fn encode_chunk<W: RmpWrite, T: TraceData>(
415 writer: &mut W,
416 spans: &[Span<T>],
417 table: &mut StringTable,
418) -> Result<(), ValueWriteError<W::Error>> {
419 let attrs = extract_chunk_attrs(spans);
420
421 let fields = 2u32 + attrs.origin.is_some() as u32
423 + attrs.sampling_priority.is_some() as u32
424 + attrs.sampling_mechanism.is_some() as u32;
425
426 write_map_len(writer, fields)?;
427
428 write_uint8(writer, chunk_key::TRACE_ID)?;
429 write_bin(writer, &attrs.trace_id.to_be_bytes())?;
430
431 write_uint8(writer, chunk_key::SPANS)?;
432 write_array_len(writer, spans.len() as u32)?;
433 for span in spans {
434 span_v04::encode_span(writer, span, table)?;
435 }
436
437 if let Some(origin) = attrs.origin {
438 write_uint8(writer, chunk_key::ORIGIN)?;
439 table.write_interned(writer, origin)?;
440 }
441
442 if let Some(priority) = attrs.sampling_priority {
443 write_uint8(writer, chunk_key::PRIORITY)?;
444 write_sint(writer, priority as i64)?;
445 }
446
447 if let Some(mechanism) = attrs.sampling_mechanism {
448 write_uint8(writer, chunk_key::SAMPLING_MECHANISM)?;
449 write_uint(writer, mechanism as u64)?;
450 }
451
452 Ok(())
453}
454
455pub fn write_to_slice<T: TraceData, S: AsRef<[Span<T>]>>(
460 slice: &mut &mut [u8],
462 traces: &[S],
463 metadata: &TracerMetadata,
464) -> Result<(), ValueWriteError> {
465 encode_payload(slice, traces, metadata)
466}
467
468pub fn to_vec<T: TraceData, S: AsRef<[Span<T>]>>(
470 traces: &[S],
471 metadata: &TracerMetadata,
472) -> Vec<u8> {
473 to_vec_with_capacity(traces, 0, metadata)
474}
475
476pub fn to_vec_with_capacity<T: TraceData, S: AsRef<[Span<T>]>>(
478 traces: &[S],
479 capacity: u32,
480 metadata: &TracerMetadata,
481) -> Vec<u8> {
482 let mut buf = ByteBuf::with_capacity(capacity as usize);
483 encode_payload(&mut buf, traces, metadata)
484 .map_err(super::flatten_value_write_infallible)
485 .unwrap_infallible();
486 buf.into_vec()
487}
488
489pub fn to_encoded_byte_len<T: TraceData, S: AsRef<[Span<T>]>>(
491 traces: &[S],
492 metadata: &TracerMetadata,
493) -> u32 {
494 let mut counter = super::CountLength(0);
495 let _ = encode_payload(&mut counter, traces, metadata);
501 counter.0
502}
503
504fn encode_payload_v1<W: RmpWrite, T: TraceData>(
506 writer: &mut W,
507 payload: &TracerPayload<T>,
508) -> Result<(), ValueWriteError<W::Error>> {
509 let mut table = StringTable::new();
510
511 let has_attributes = !payload.attributes.is_empty();
512
513 let map_len = 1u32 + (!payload.language_name.borrow().is_empty()) as u32
515 + (!payload.language_version.borrow().is_empty()) as u32
516 + (!payload.tracer_version.borrow().is_empty()) as u32
517 + (!payload.runtime_id.borrow().is_empty()) as u32
518 + (!payload.env.borrow().is_empty()) as u32
519 + (!payload.hostname.borrow().is_empty()) as u32
520 + (!payload.app_version.borrow().is_empty()) as u32
521 + has_attributes as u32;
522
523 write_map_len(writer, map_len)?;
524
525 write_uint8(writer, trace_key::CHUNKS)?;
526 write_array_len(writer, payload.chunks.len() as u32)?;
527 for chunk in &payload.chunks {
528 encode_chunk_v1(writer, chunk, &mut table)?;
529 }
530
531 if !payload.language_name.borrow().is_empty() {
532 write_uint8(writer, trace_key::LANGUAGE_NAME)?;
533 table.write_interned(writer, payload.language_name.borrow())?;
534 }
535
536 if !payload.language_version.borrow().is_empty() {
537 write_uint8(writer, trace_key::LANGUAGE_VERSION)?;
538 table.write_interned(writer, payload.language_version.borrow())?;
539 }
540
541 if !payload.tracer_version.borrow().is_empty() {
542 write_uint8(writer, trace_key::TRACER_VERSION)?;
543 table.write_interned(writer, payload.tracer_version.borrow())?;
544 }
545
546 if !payload.runtime_id.borrow().is_empty() {
547 write_uint8(writer, trace_key::RUNTIME_ID)?;
548 table.write_interned(writer, payload.runtime_id.borrow())?;
549 }
550
551 if !payload.env.borrow().is_empty() {
552 write_uint8(writer, trace_key::ENV_REF)?;
553 table.write_interned(writer, payload.env.borrow())?;
554 }
555
556 if !payload.hostname.borrow().is_empty() {
557 write_uint8(writer, trace_key::HOSTNAME_REF)?;
558 table.write_interned(writer, payload.hostname.borrow())?;
559 }
560
561 if !payload.app_version.borrow().is_empty() {
562 write_uint8(writer, trace_key::APP_VERSION_REF)?;
563 table.write_interned(writer, payload.app_version.borrow())?;
564 }
565
566 if has_attributes {
567 write_uint8(writer, trace_key::ATTRIBUTES)?;
568 span_v1::encode_attributes_map(writer, &payload.attributes, &mut table)?;
569 }
570
571 Ok(())
572}
573
574fn encode_chunk_v1<W: RmpWrite, T: TraceData>(
576 writer: &mut W,
577 chunk: &crate::span::v1::TraceChunk<T>,
578 table: &mut StringTable,
579) -> Result<(), ValueWriteError<W::Error>> {
580 let origin = <T::Text as Borrow<str>>::borrow(&chunk.origin);
581 let has_attributes = !chunk.attributes.is_empty();
582 let has_dropped = chunk.dropped_trace;
583
584 let fields = 2u32 + !origin.is_empty() as u32
586 + chunk.priority.is_some() as u32
587 + chunk.sampling_mechanism.is_some() as u32
588 + has_attributes as u32
589 + has_dropped as u32;
590
591 write_map_len(writer, fields)?;
592
593 write_uint8(writer, chunk_key::TRACE_ID)?;
594 write_bin(writer, &chunk.trace_id)?;
595
596 write_uint8(writer, chunk_key::SPANS)?;
597 write_array_len(writer, chunk.spans.len() as u32)?;
598 for span in &chunk.spans {
599 span_v1::encode_span(writer, span, table)?;
600 }
601
602 if !origin.is_empty() {
603 write_uint8(writer, chunk_key::ORIGIN)?;
604 table.write_interned(writer, origin)?;
605 }
606
607 if let Some(priority) = chunk.priority {
608 write_uint8(writer, chunk_key::PRIORITY)?;
609 write_sint(writer, priority as i64)?;
610 }
611
612 if let Some(mechanism) = chunk.sampling_mechanism {
613 write_uint8(writer, chunk_key::SAMPLING_MECHANISM)?;
614 write_uint(writer, mechanism as u64)?;
615 }
616
617 if has_attributes {
618 write_uint8(writer, chunk_key::ATTRIBUTES)?;
619 span_v1::encode_attributes_map(writer, &chunk.attributes, table)?;
620 }
621
622 if has_dropped {
623 write_uint8(writer, chunk_key::DROPPED_TRACE)?;
624 rmp::encode::write_bool(writer, true).map_err(ValueWriteError::InvalidDataWrite)?;
625 }
626
627 Ok(())
628}
629
630pub fn to_vec_from_payload_v1<T: TraceData>(payload: &TracerPayload<T>) -> Vec<u8> {
655 to_vec_from_payload_with_capacity_v1(payload, 0)
656}
657
658pub fn to_vec_from_payload_with_capacity_v1<T: TraceData>(
684 payload: &TracerPayload<T>,
685 capacity: u32,
686) -> Vec<u8> {
687 let mut buf = ByteBuf::with_capacity(capacity as usize);
688 encode_payload_v1(&mut buf, payload)
689 .map_err(super::flatten_value_write_infallible)
690 .unwrap_infallible();
691 buf.into_vec()
692}
693
694pub fn write_payload_to_slice_v1<T: TraceData>(
725 slice: &mut &mut [u8],
726 payload: &TracerPayload<T>,
727) -> Result<(), ValueWriteError> {
728 encode_payload_v1(slice, payload)
729}
730
731pub fn to_encoded_byte_len_from_payload_v1<T: TraceData>(payload: &TracerPayload<T>) -> u32 {
759 let mut counter = super::CountLength(0);
760 let _ = encode_payload_v1(&mut counter, payload);
761 counter.0
762}
763
764#[cfg(test)]
765mod tests {
766 use super::*;
767 use crate::span::v04::SpanBytes;
768 use libdd_tinybytes::BytesString;
769
770 fn make_span(
771 service: &str,
772 name: &str,
773 trace_id: u128,
774 span_id: u64,
775 parent_id: u64,
776 ) -> SpanBytes {
777 SpanBytes {
778 service: BytesString::from_slice(service.as_bytes()).unwrap(),
779 name: BytesString::from_slice(name.as_bytes()).unwrap(),
780 resource: BytesString::from_slice(b"res").unwrap(),
781 trace_id,
782 span_id,
783 parent_id,
784 start: 1_000_000,
785 duration: 500,
786 ..Default::default()
787 }
788 }
789
790 #[test]
791 fn test_to_vec_non_empty() {
792 let spans = vec![make_span("svc", "op", 42, 1, 0)];
793 let traces = vec![spans];
794 let encoded = to_vec(&traces, &TracerMetadata::default());
795 assert!(!encoded.is_empty());
796 }
797
798 #[test]
799 fn test_to_vec_empty_traces() {
800 let traces: Vec<Vec<SpanBytes>> = vec![];
801 let encoded = to_vec(&traces, &TracerMetadata::default());
802 assert!(!encoded.is_empty());
804 }
805
806 #[test]
807 fn test_string_interning_reduces_size() {
808 let s1 = make_span("my-service", "op1", 1, 1, 0);
810 let s2 = make_span("my-service", "op2", 2, 2, 0);
811 let traces_two = vec![vec![s1], vec![s2]];
812
813 let s_single = make_span("my-service", "op1", 1, 1, 0);
815 let traces_single = vec![vec![s_single]];
816
817 let encoded_two = to_vec(&traces_two, &TracerMetadata::default());
818 let encoded_single = to_vec(&traces_single, &TracerMetadata::default());
819
820 assert!(
823 encoded_two.len() < 2 * encoded_single.len(),
824 "Interning should reduce size: two={} single={}",
825 encoded_two.len(),
826 encoded_single.len()
827 );
828 }
829
830 #[test]
831 fn test_chunk_level_attrs_origin_and_priority() {
832 let meta = vec![(
833 BytesString::from_static("_dd.origin"),
834 BytesString::from_static("lambda"),
835 )]
836 .into();
837 let metrics = vec![(BytesString::from_static("_sampling_priority_v1"), 1.0f64)].into();
838
839 let root = SpanBytes {
840 service: BytesString::from_slice(b"svc").unwrap(),
841 name: BytesString::from_slice(b"op").unwrap(),
842 resource: BytesString::from_slice(b"res").unwrap(),
843 trace_id: 99,
844 span_id: 1,
845 parent_id: 0,
846 start: 1000,
847 duration: 100,
848 meta,
849 metrics,
850 ..Default::default()
851 };
852
853 let encoded = to_vec(&[vec![root]], &TracerMetadata::default());
854 assert!(!encoded.is_empty());
855 let lambda_bytes = b"lambda";
857 assert!(
858 encoded
859 .windows(lambda_bytes.len())
860 .any(|w| w == lambda_bytes),
861 "origin 'lambda' should appear in payload"
862 );
863 }
864
865 #[test]
866 fn test_to_encoded_byte_len_matches_to_vec() {
867 let spans = vec![
868 make_span("svc", "op", 1, 1, 0),
869 make_span("svc", "child", 1, 2, 1),
870 ];
871 let traces = vec![spans];
872 let meta = TracerMetadata::default();
873 let encoded = to_vec(&traces, &meta);
874 let len = to_encoded_byte_len(&traces, &meta);
875 assert_eq!(encoded.len() as u32, len);
876 }
877
878 #[test]
879 fn test_remote_parent_root_span_top_level() {
880 let metrics = vec![
882 (BytesString::from_static("_dd.top_level"), 1.0f64),
883 (BytesString::from_static("_sampling_priority_v1"), 2.0f64),
884 ]
885 .into();
886
887 let root = SpanBytes {
888 service: BytesString::from_slice(b"svc").unwrap(),
889 name: BytesString::from_slice(b"op").unwrap(),
890 resource: BytesString::from_slice(b"res").unwrap(),
891 trace_id: 123,
892 span_id: 42,
893 parent_id: 999, start: 1000,
895 duration: 100,
896 metrics,
897 ..Default::default()
898 };
899
900 let encoded = to_vec(&[vec![root]], &TracerMetadata::default());
901 assert!(!encoded.is_empty());
902 }
903
904 #[test]
905 fn test_payload_promoted_fields() {
906 let meta = vec![
907 (
908 BytesString::from_static("env"),
909 BytesString::from_static("prod"),
910 ),
911 (
912 BytesString::from_static("version"),
913 BytesString::from_static("1.2.3"),
914 ),
915 (
916 BytesString::from_static("_dd.hostname"),
917 BytesString::from_static("my-host"),
918 ),
919 ]
920 .into();
921
922 let span = SpanBytes {
923 service: BytesString::from_slice(b"svc").unwrap(),
924 name: BytesString::from_slice(b"op").unwrap(),
925 resource: BytesString::from_slice(b"res").unwrap(),
926 trace_id: 1,
927 span_id: 1,
928 parent_id: 0,
929 start: 1000,
930 duration: 100,
931 meta,
932 ..Default::default()
933 };
934
935 let encoded = to_vec(&[vec![span]], &TracerMetadata::default());
936 let prod_bytes = b"prod";
937 assert!(
938 encoded.windows(prod_bytes.len()).any(|w| w == prod_bytes),
939 "env 'prod' should appear in payload"
940 );
941 let host_bytes = b"my-host";
942 assert!(
943 encoded.windows(host_bytes.len()).any(|w| w == host_bytes),
944 "hostname 'my-host' should appear in payload"
945 );
946 }
947
948 #[test]
949 fn test_payload_attributes_apm_mode_and_git_commit_sha() {
950 let meta = vec![
951 (
952 BytesString::from_static("_dd.apm_mode"),
953 BytesString::from_static("ssi"),
954 ),
955 (
956 BytesString::from_static("_dd.git.commit.sha"),
957 BytesString::from_static("abc123"),
958 ),
959 ]
960 .into();
961
962 let span = SpanBytes {
963 service: BytesString::from_slice(b"svc").unwrap(),
964 name: BytesString::from_slice(b"op").unwrap(),
965 resource: BytesString::from_slice(b"res").unwrap(),
966 trace_id: 1,
967 span_id: 1,
968 parent_id: 0,
969 start: 1000,
970 duration: 100,
971 meta,
972 ..Default::default()
973 };
974
975 let encoded = to_vec(&[vec![span]], &TracerMetadata::default());
976
977 let ssi_bytes = b"ssi";
979 assert!(
980 encoded.windows(ssi_bytes.len()).any(|w| w == ssi_bytes),
981 "apm_mode 'ssi' should appear in payload"
982 );
983 let sha_bytes = b"abc123";
984 assert!(
985 encoded.windows(sha_bytes.len()).any(|w| w == sha_bytes),
986 "git commit sha 'abc123' should appear in payload"
987 );
988 let apm_key = b"_dd.apm_mode";
990 assert!(
991 encoded.windows(apm_key.len()).any(|w| w == apm_key),
992 "_dd.apm_mode key should appear in payload"
993 );
994 let git_key = b"_dd.git.commit.sha";
995 assert!(
996 encoded.windows(git_key.len()).any(|w| w == git_key),
997 "_dd.git.commit.sha key should appear in payload"
998 );
999 }
1000
1001 #[test]
1002 fn test_payload_attributes_absent_when_no_relevant_tags() {
1003 let span = make_span("svc", "op", 1, 1, 0);
1005 let encoded = to_vec(&[vec![span]], &TracerMetadata::default());
1006 let apm_key = b"_dd.apm_mode";
1007 assert!(
1008 !encoded.windows(apm_key.len()).any(|w| w == apm_key),
1009 "key 10 should be absent when no relevant tags are set"
1010 );
1011 }
1012
1013 #[test]
1014 fn test_payload_metadata_fields_present() {
1015 let span = make_span("svc", "op", 1, 1, 0);
1016 let metadata = TracerMetadata {
1017 language: "python".to_string(),
1018 language_version: "3.11".to_string(),
1019 tracer_version: "2.0.0".to_string(),
1020 runtime_id: "abc-123-uuid".to_string(),
1021 ..Default::default()
1022 };
1023 let encoded = to_vec(&[vec![span]], &metadata);
1024
1025 for s in &[b"python" as &[u8], b"3.11", b"2.0.0", b"abc-123-uuid"] {
1026 assert!(
1027 encoded.windows(s.len()).any(|w| w == *s),
1028 "{} should appear in payload",
1029 std::str::from_utf8(s).unwrap()
1030 );
1031 }
1032 }
1033
1034 #[test]
1035 fn test_payload_metadata_absent_when_empty() {
1036 let span = make_span("svc", "op", 1, 1, 0);
1037 let encoded_with = to_vec(
1038 &[vec![span.clone()]],
1039 &TracerMetadata {
1040 language: "go".to_string(),
1041 ..Default::default()
1042 },
1043 );
1044 let encoded_without = to_vec(&[vec![span]], &TracerMetadata::default());
1045 assert!(encoded_with.len() > encoded_without.len());
1047 }
1048
1049 #[test]
1050 fn test_128bit_trace_id_from_dd_p_tid() {
1051 let meta = vec![(
1052 BytesString::from_static("_dd.p.tid"),
1053 BytesString::from_static("640cfd5400000000"),
1054 )]
1055 .into();
1056 let span = SpanBytes {
1057 service: BytesString::from_slice(b"svc").unwrap(),
1058 name: BytesString::from_slice(b"op").unwrap(),
1059 resource: BytesString::from_slice(b"res").unwrap(),
1060 trace_id: 0x0123456789abcdef,
1061 span_id: 1,
1062 parent_id: 0,
1063 start: 1000,
1064 duration: 100,
1065 meta,
1066 ..Default::default()
1067 };
1068 let encoded = to_vec(&[vec![span]], &TracerMetadata::default());
1069
1070 let expected = [
1072 0x64, 0x0c, 0xfd, 0x54, 0x00, 0x00, 0x00, 0x00, 0x01, 0x23, 0x45, 0x67, 0x89, 0xab,
1073 0xcd, 0xef,
1074 ];
1075 assert!(
1076 encoded.windows(16).any(|w| w == expected),
1077 "128-bit trace_id big-endian bytes should appear in payload"
1078 );
1079 let tid_key = b"_dd.p.tid";
1081 assert!(
1082 !encoded.windows(tid_key.len()).any(|w| w == tid_key),
1083 "_dd.p.tid should be consumed, not encoded as a span attribute"
1084 );
1085 }
1086
1087 #[test]
1088 fn test_128bit_trace_id_without_dd_p_tid() {
1089 let span = make_span("svc", "op", 0x0123456789abcdef, 1, 0);
1091 let encoded = to_vec(&[vec![span]], &TracerMetadata::default());
1092 let expected = [
1093 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x23, 0x45, 0x67, 0x89, 0xab,
1094 0xcd, 0xef,
1095 ];
1096 assert!(
1097 encoded.windows(16).any(|w| w == expected),
1098 "absent _dd.p.tid should yield zero high 64 bits"
1099 );
1100 }
1101
1102 #[test]
1103 fn test_sampling_mechanism_negative_value() {
1104 let meta = vec![(
1107 BytesString::from_static("_dd.p.dm"),
1108 BytesString::from_static("-4"),
1109 )]
1110 .into();
1111 let root = SpanBytes {
1112 service: BytesString::from_slice(b"svc").unwrap(),
1113 name: BytesString::from_slice(b"op").unwrap(),
1114 resource: BytesString::from_slice(b"res").unwrap(),
1115 trace_id: 1,
1116 span_id: 1,
1117 parent_id: 0,
1118 start: 1000,
1119 duration: 100,
1120 meta,
1121 ..Default::default()
1122 };
1123 let encoded = to_vec(&[vec![root]], &TracerMetadata::default());
1124
1125 let expected = [chunk_key::SAMPLING_MECHANISM, 0x04];
1129 assert!(
1130 encoded.windows(2).any(|w| w == expected),
1131 "sampling_mechanism should be encoded as unsigned_abs(\"-4\") = 4"
1132 );
1133 }
1134
1135 #[test]
1136 fn test_chunk_attrs_fallback_no_root_span() {
1137 let meta1 = vec![(
1140 BytesString::from_static("_dd.origin"),
1141 BytesString::from_static("lambda"),
1142 )]
1143 .into();
1144 let metrics2 = vec![(BytesString::from_static("_sampling_priority_v1"), 2.0f64)].into();
1145 let meta3 = vec![(
1146 BytesString::from_static("_dd.p.dm"),
1147 BytesString::from_static("-3"),
1148 )]
1149 .into();
1150
1151 let s1 = SpanBytes {
1152 service: BytesString::from_slice(b"svc").unwrap(),
1153 name: BytesString::from_slice(b"op1").unwrap(),
1154 resource: BytesString::from_slice(b"res").unwrap(),
1155 trace_id: 1,
1156 span_id: 11,
1157 parent_id: 10, start: 1000,
1159 duration: 100,
1160 meta: meta1,
1161 ..Default::default()
1162 };
1163 let s2 = SpanBytes {
1164 service: BytesString::from_slice(b"svc").unwrap(),
1165 name: BytesString::from_slice(b"op2").unwrap(),
1166 resource: BytesString::from_slice(b"res").unwrap(),
1167 trace_id: 1,
1168 span_id: 12,
1169 parent_id: 11,
1170 start: 1000,
1171 duration: 100,
1172 metrics: metrics2,
1173 ..Default::default()
1174 };
1175 let s3 = SpanBytes {
1176 service: BytesString::from_slice(b"svc").unwrap(),
1177 name: BytesString::from_slice(b"op3").unwrap(),
1178 resource: BytesString::from_slice(b"res").unwrap(),
1179 trace_id: 1,
1180 span_id: 13,
1181 parent_id: 12,
1182 start: 1000,
1183 duration: 100,
1184 meta: meta3,
1185 ..Default::default()
1186 };
1187 let encoded = to_vec(&[vec![s1, s2, s3]], &TracerMetadata::default());
1188
1189 let lambda = b"lambda";
1192 assert!(
1193 encoded.windows(lambda.len()).any(|w| w == lambda),
1194 "origin 'lambda' from span 1 should appear in payload"
1195 );
1196 let prio = [chunk_key::PRIORITY, 0x02];
1198 assert!(
1199 encoded.windows(2).any(|w| w == prio),
1200 "sampling_priority 2 from span 2 should appear"
1201 );
1202 let mech = [chunk_key::SAMPLING_MECHANISM, 0x03];
1204 assert!(
1205 encoded.windows(2).any(|w| w == mech),
1206 "sampling_mechanism 3 from span 3 should appear"
1207 );
1208 }
1209}
1210
1211#[cfg(test)]
1212mod v1_payload_tests {
1213 use super::*;
1220 use crate::span::v1::{
1221 AttributeValue, Span as V1Span, SpanBytes as V1SpanBytes, SpanKind, TraceChunkBytes,
1222 TracerPayloadBytes,
1223 };
1224 use crate::span::vec_map::VecMap;
1225 use libdd_tinybytes::BytesString;
1226
1227 fn bs(s: &str) -> BytesString {
1228 BytesString::from_slice(s.as_bytes()).expect("test string must fit in BytesString")
1229 }
1230
1231 fn make_span(service: &str, name: &str, span_id: u64) -> V1SpanBytes {
1232 V1Span {
1233 service: bs(service),
1234 name: bs(name),
1235 resource: bs("res"),
1236 span_id,
1237 start: 1_000_000,
1238 duration: 500,
1239 ..Default::default()
1240 }
1241 }
1242
1243 fn make_chunk(spans: Vec<V1SpanBytes>, trace_id: [u8; 16]) -> TraceChunkBytes {
1244 TraceChunkBytes {
1245 trace_id,
1246 spans,
1247 ..Default::default()
1248 }
1249 }
1250
1251 #[test]
1252 fn empty_payload_is_valid_msgpack_map() {
1253 let payload = TracerPayloadBytes::default();
1254 let encoded = to_vec_from_payload_v1(&payload);
1255 assert_eq!(encoded, vec![0x81, 0x0b, 0x90]);
1258 }
1259
1260 #[test]
1261 fn payload_byte_len_matches_to_vec() {
1262 let chunk = make_chunk(vec![make_span("svc", "op", 1)], [0u8; 16]);
1263 let payload = TracerPayloadBytes {
1264 chunks: vec![chunk],
1265 ..Default::default()
1266 };
1267 let encoded = to_vec_from_payload_v1(&payload);
1268 let len = to_encoded_byte_len_from_payload_v1(&payload);
1269 assert_eq!(encoded.len() as u32, len);
1270 }
1271
1272 #[test]
1273 fn span_kind_is_always_emitted_as_uint() {
1274 let chunk = make_chunk(vec![make_span("svc", "op", 1)], [0u8; 16]);
1277 let payload = TracerPayloadBytes {
1278 chunks: vec![chunk],
1279 ..Default::default()
1280 };
1281 let encoded = to_vec_from_payload_v1(&payload);
1282 let pat = [0x10u8, 0x01u8];
1283 assert!(
1284 encoded.windows(2).any(|w| w == pat),
1285 "Kind (key=16) Internal (=1) must be emitted"
1286 );
1287 }
1288
1289 #[test]
1290 fn typed_attributes_carry_correct_type_discriminants() {
1291 let mut attrs = VecMap::new();
1292 attrs.insert(bs("k_str"), AttributeValue::String(bs("v")));
1293 let span = V1Span {
1294 service: bs("svc"),
1295 name: bs("op"),
1296 resource: bs("res"),
1297 span_id: 1,
1298 start: 1,
1299 duration: 1,
1300 attributes: attrs,
1301 ..Default::default()
1302 };
1303 let chunk = make_chunk(vec![span], [0u8; 16]);
1304 let payload = TracerPayloadBytes {
1305 chunks: vec![chunk],
1306 ..Default::default()
1307 };
1308 let encoded = to_vec_from_payload_v1(&payload);
1309 assert!(
1311 encoded.windows(b"k_str".len()).any(|w| w == b"k_str"),
1312 "attribute key must appear"
1313 );
1314 }
1315
1316 #[test]
1317 fn bytes_attribute_uses_bin_marker() {
1318 let mut attrs = VecMap::new();
1320 attrs.insert(
1321 bs("payload"),
1322 AttributeValue::Bytes(libdd_tinybytes::Bytes::copy_from_slice(b"\xde\xad")),
1323 );
1324 let span = V1Span {
1325 service: bs("svc"),
1326 name: bs("op"),
1327 resource: bs("res"),
1328 span_id: 1,
1329 start: 1,
1330 duration: 1,
1331 attributes: attrs,
1332 ..Default::default()
1333 };
1334 let payload = TracerPayloadBytes {
1335 chunks: vec![make_chunk(vec![span], [0u8; 16])],
1336 ..Default::default()
1337 };
1338 let encoded = to_vec_from_payload_v1(&payload);
1339 let want = [0xc4u8, 0x02, 0xde, 0xad];
1341 assert!(
1342 encoded.windows(4).any(|w| w == want),
1343 "Bytes attribute must be encoded as msgpack bin"
1344 );
1345 }
1346
1347 #[test]
1348 fn list_and_keyvalue_attributes_round_trip_through_recursion() {
1349 let mut nested = VecMap::new();
1350 nested.insert(bs("nk"), AttributeValue::Int(7));
1351 let mut attrs = VecMap::new();
1352 attrs.insert(
1353 bs("list"),
1354 AttributeValue::List(vec![
1355 AttributeValue::String(bs("a")),
1356 AttributeValue::Bool(true),
1357 ]),
1358 );
1359 attrs.insert(bs("kv"), AttributeValue::KeyValue(nested));
1360 let span = V1Span {
1361 service: bs("svc"),
1362 name: bs("op"),
1363 resource: bs("res"),
1364 span_id: 1,
1365 start: 1,
1366 duration: 1,
1367 attributes: attrs,
1368 ..Default::default()
1369 };
1370 let payload = TracerPayloadBytes {
1371 chunks: vec![make_chunk(vec![span], [0u8; 16])],
1372 ..Default::default()
1373 };
1374 let encoded = to_vec_from_payload_v1(&payload);
1375 for s in &[b"list" as &[u8], b"kv", b"a", b"nk"] {
1377 assert!(
1378 encoded.windows(s.len()).any(|w| w == *s),
1379 "{} should appear in payload",
1380 std::str::from_utf8(s).unwrap()
1381 );
1382 }
1383 }
1384
1385 #[test]
1386 fn promoted_fields_at_payload_level() {
1387 let payload = TracerPayloadBytes {
1388 language_name: bs("python"),
1389 language_version: bs("3.11"),
1390 tracer_version: bs("2.0.0"),
1391 runtime_id: bs("rt-1"),
1392 env: bs("prod"),
1393 hostname: bs("h"),
1394 app_version: bs("1.2.3"),
1395 chunks: vec![make_chunk(vec![make_span("svc", "op", 1)], [0u8; 16])],
1396 ..Default::default()
1397 };
1398 let encoded = to_vec_from_payload_v1(&payload);
1399 for s in &[
1400 b"python" as &[u8],
1401 b"3.11",
1402 b"2.0.0",
1403 b"rt-1",
1404 b"prod",
1405 b"1.2.3",
1406 ] {
1407 assert!(
1408 encoded.windows(s.len()).any(|w| w == *s),
1409 "{} should appear",
1410 std::str::from_utf8(s).unwrap()
1411 );
1412 }
1413 }
1414
1415 #[test]
1416 fn chunk_level_attrs_emitted_when_set() {
1417 let chunk = TraceChunkBytes {
1418 trace_id: [0u8; 16],
1419 priority: Some(1),
1420 origin: bs("lambda"),
1421 sampling_mechanism: Some(4),
1422 spans: vec![make_span("svc", "op", 1)],
1423 ..Default::default()
1424 };
1425 let payload = TracerPayloadBytes {
1426 chunks: vec![chunk],
1427 ..Default::default()
1428 };
1429 let encoded = to_vec_from_payload_v1(&payload);
1430 assert!(
1431 encoded.windows(b"lambda".len()).any(|w| w == b"lambda"),
1432 "chunk origin should appear"
1433 );
1434 let want = [chunk_key::SAMPLING_MECHANISM, 0x04];
1436 assert!(encoded.windows(2).any(|w| w == want));
1437 }
1438
1439 #[test]
1440 fn chunk_dropped_trace_emitted_when_true() {
1441 let chunk = TraceChunkBytes {
1442 trace_id: [0u8; 16],
1443 dropped_trace: true,
1444 spans: vec![make_span("svc", "op", 1)],
1445 ..Default::default()
1446 };
1447 let payload = TracerPayloadBytes {
1448 chunks: vec![chunk],
1449 ..Default::default()
1450 };
1451 let encoded = to_vec_from_payload_v1(&payload);
1452 let want = [chunk_key::DROPPED_TRACE, 0xc3];
1454 assert!(
1455 encoded.windows(2).any(|w| w == want),
1456 "DROPPED_TRACE marker + true should appear in payload"
1457 );
1458 }
1459
1460 #[test]
1461 fn chunk_dropped_trace_skipped_when_false() {
1462 let chunk = TraceChunkBytes {
1463 trace_id: [0u8; 16],
1464 dropped_trace: false,
1465 spans: vec![make_span("svc", "op", 1)],
1466 ..Default::default()
1467 };
1468 let payload = TracerPayloadBytes {
1469 chunks: vec![chunk],
1470 ..Default::default()
1471 };
1472 let encoded = to_vec_from_payload_v1(&payload);
1473 assert!(
1474 !encoded.contains(&chunk_key::DROPPED_TRACE),
1475 "DROPPED_TRACE key should not be emitted when false"
1476 );
1477 }
1478
1479 #[test]
1480 fn chunk_attributes_emitted_when_set() {
1481 let mut attrs = VecMap::new();
1482 attrs.insert(bs("region"), AttributeValue::String(bs("us-east-1")));
1483 let chunk = TraceChunkBytes {
1484 trace_id: [0u8; 16],
1485 attributes: attrs,
1486 spans: vec![make_span("svc", "op", 1)],
1487 ..Default::default()
1488 };
1489 let payload = TracerPayloadBytes {
1490 chunks: vec![chunk],
1491 ..Default::default()
1492 };
1493 let encoded = to_vec_from_payload_v1(&payload);
1494 let want = [chunk_key::ATTRIBUTES, 0x93];
1496 assert!(
1497 encoded.windows(2).any(|w| w == want),
1498 "ATTRIBUTES key + flat-triplet array header should appear"
1499 );
1500 assert!(
1501 encoded
1502 .windows(b"us-east-1".len())
1503 .any(|w| w == b"us-east-1"),
1504 "chunk attribute value should be in the payload"
1505 );
1506 }
1507
1508 #[test]
1509 fn span_kind_otel_values() {
1510 for (kind, expected_byte) in [
1511 (SpanKind::Internal, 0x01u8),
1512 (SpanKind::Server, 0x02),
1513 (SpanKind::Client, 0x03),
1514 (SpanKind::Producer, 0x04),
1515 (SpanKind::Consumer, 0x05),
1516 ] {
1517 let span = V1Span {
1518 service: bs("svc"),
1519 name: bs("op"),
1520 resource: bs("res"),
1521 span_id: 1,
1522 start: 1,
1523 duration: 1,
1524 span_kind: kind,
1525 ..Default::default()
1526 };
1527 let payload = TracerPayloadBytes {
1528 chunks: vec![make_chunk(vec![span], [0u8; 16])],
1529 ..Default::default()
1530 };
1531 let encoded = to_vec_from_payload_v1(&payload);
1532 let want = [0x10u8, expected_byte];
1533 assert!(
1534 encoded.windows(2).any(|w| w == want),
1535 "SpanKind {kind:?} should produce byte {expected_byte:#x}"
1536 );
1537 }
1538 }
1539
1540 #[test]
1541 fn string_interning_works_across_chunks() {
1542 let chunk_with_two = TracerPayloadBytes {
1547 chunks: vec![
1548 make_chunk(vec![make_span("shared", "op1", 1)], [0u8; 16]),
1549 make_chunk(vec![make_span("shared", "op2", 2)], [0u8; 16]),
1550 ],
1551 ..Default::default()
1552 };
1553 let single = TracerPayloadBytes {
1554 chunks: vec![make_chunk(vec![make_span("shared", "op1", 1)], [0u8; 16])],
1555 ..Default::default()
1556 };
1557 let two = to_vec_from_payload_v1(&chunk_with_two);
1558 let one = to_vec_from_payload_v1(&single);
1559 let shared_occurrences = two
1560 .windows(b"shared".len())
1561 .filter(|w| *w == b"shared")
1562 .count();
1563 assert_eq!(
1564 shared_occurrences, 1,
1565 "the literal bytes \"shared\" must appear exactly once; subsequent uses must be \
1566 encoded as interning IDs"
1567 );
1568 assert!(
1569 two.len() < 2 * one.len(),
1570 "interning should reduce repeated payload size"
1571 );
1572 }
1573}