feat(trace-utils)!: add v1 decoder#2174
Conversation
📚 Documentation Check Results📦
|
Clippy Allow Annotation ReportComparing clippy allow annotations between branches:
Summary by Rule
Annotation Counts by File
Annotation Stats by Crate
About This ReportThis report tracks Clippy allow annotations for specific rules, showing how they've changed in this PR. Decreasing the number of these annotations generally improves code quality. |
🔒 Cargo Deny Results📦
|
🎉 All green!🧪 All tests passed 🎯 Code Coverage (details) 🔗 Commit SHA: 1af6df1 | Docs | Datadog PR Page | Give us feedback! |
Artifact Size Benchmark Reportaarch64-alpine-linux-musl
aarch64-unknown-linux-gnu
libdatadog-x64-windows
libdatadog-x86-windows
x86_64-alpine-linux-musl
x86_64-unknown-linux-gnu
|
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5434161c71
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
ajgajg1134
left a comment
There was a problem hiding this comment.
Another thing we can do here to build confidence in the decoder is take the working dd-trace-go implementation and generate a bunch of various traces and put them just in to flat files, then decode them with this implementation and the trace-agent implementation to make sure they align? (Maybe there's an easier way to check this though)
…cto-2' into anais/add-v1-decoder
| fn record(&mut self, s: T::Text) -> T::Text { | ||
| self.seen.push(s.clone()); | ||
| s | ||
| } |
There was a problem hiding this comment.
If we clone it, couldn't we take it by reference instead and not return it?
| where | ||
| T::Text: Clone, | ||
| { | ||
| let slice: &[u8] = buf.as_mut_slice(); |
There was a problem hiding this comment.
| let slice: &[u8] = buf.as_mut_slice(); | |
| let slice: &[u8] = buf.as_slice(); |
| let is_string = matches!(marker_byte, 0xa0..=0xbf | 0xd9 | 0xda | 0xdb); | ||
| let is_uint = matches!(marker_byte, 0x00..=0x7f | 0xcc | 0xcd | 0xce | 0xcf); | ||
|
|
||
| if is_string { | ||
| let s = buf.read_string()?; | ||
| Ok(table.record(s)) | ||
| } else if is_uint { | ||
| let id: u64 = decode::read_int(buf.as_mut_slice()).map_err(|_| { | ||
| DecodeError::InvalidFormat("V1 interned string reference uint read failure".to_owned()) | ||
| })?; | ||
| table.resolve(id) | ||
| } else { | ||
| Err(DecodeError::InvalidFormat(format!( | ||
| "Unexpected msgpack marker 0x{marker_byte:02x} for V1 interned string" | ||
| ))) | ||
| } |
There was a problem hiding this comment.
Why not have a proper single match expression instead of storing matches! results in boolean and then having an if-then-else?
| let len = decode::read_bin_len(buf.as_mut_slice()).map_err(|_| { | ||
| DecodeError::InvalidFormat("V1 chunk trace_id bin len read failure".to_owned()) | ||
| })?; | ||
| if len != 16 { |
There was a problem hiding this comment.
should 16 be in a constant somewhere? I think it's repeated below
| })?; | ||
| // OTEL spec: unset / unknown → Internal. | ||
| span.span_kind = match kind { | ||
| 2 => SpanKind::Server, |
There was a problem hiding this comment.
Very nit, but should the values on the left be constants as well? Like OTEL_SPAN_KIND_SERVER or something
| } | ||
|
|
||
| // Compute the encoding of a string to msgpack in a const manner | ||
| const fn msgpack_const_string_encoding<const ENCODING_LEN: usize>(s: &str) -> [u8; ENCODING_LEN] { |
There was a problem hiding this comment.
I'm sure I've seen this function and the following macro somewhere else in Paul's PR some time ago. Is this a duplicate? Should we extract it in a reusable place?
| #[cfg(any(test, feature = "test-utils"))] | ||
| // This implementation allocates, which isn't ideal for equality testing (not allocating would be | ||
| // rather tedious though), but is needed both by production code (e.g. comparing V1 tracer metadata | ||
| // before coalescing) and by tests. |
There was a problem hiding this comment.
I'm not sure where you need to equate vecmaps, but if your keys are Ord, maybe a saner implementation of Eq/PartialEq would be to sort (with a stable sort) first and then compare the underlying vec element by element.
Also, do you need the Eq instance or would a slow_compare method work as well? I think I'd like to avoid making this instance the default for everyone; I think it can be surprising for an Eq instance to allocate, but people wouldn't know anymore, they would just use it.
Yet another possibility would be to have a thin wrapper pub VecMapSlowCompare(VecMap) (but with a better name obviously 😅 ) and implement Eq on this instead. The only issue is that this doesn't support nested VecMap.
If compare_deduped/slow_compare or whatever you call a single method is sufficient here instead of making the Eq instance public, then I would suggest to go that way for this PR. If not (typically if there are VecMaps within VecMaps), then we can think about it for follow-up work.
| /// diverging V1 tracer metadata). Callers that rely on `other` being fully drained must check | ||
| /// this return value rather than assuming success. | ||
| pub fn append(&mut self, other: &mut Self) -> bool { | ||
| match self { |
There was a problem hiding this comment.
Nit: you could match on (self, other) directly, to avoid the nested ifs in each branch.
| /// Returns `true` iff every tracer-level metadata field (string fields and attributes) of `src` | ||
| /// matches `dest`. |
There was a problem hiding this comment.
Very nit, but not sure everyone knows this math-y iff
| /// Returns `true` iff every tracer-level metadata field (string fields and attributes) of `src` | |
| /// matches `dest`. | |
| /// Returns `true` if and only if every tracer-level metadata field (string fields and attributes) of `src` | |
| /// matches `dest`. |
| ("env", dest.env == src.env), | ||
| ("hostname", dest.hostname == src.hostname), | ||
| ("app_version", dest.app_version == src.app_version), | ||
| ("attributes", dest.attributes == src.attributes), |
There was a problem hiding this comment.
Is this where you need VecMap::Eq? I think we can get by with a dedup_compare method or something then, without enabling the Eq instance for everyone?
Aaalibaba42
left a comment
There was a problem hiding this comment.
I wager this is stacked with the encoder one ? I tried to not repeat myself and focus on the decoder part
| use crate::span::{v04, v05, v1, BytesData, SharedDictBytes, TraceData}; | ||
| use crate::trace_utils::convert_trace_chunks_v04_to_v05; | ||
| use crate::{msgpack_decoder, trace_utils::cmp_send_data_payloads}; | ||
| use anyhow::Ok; |
There was a problem hiding this comment.
I don't think that's a good practice. From what I can tell from the doc it can be helpful to use it for type deduction, but then I'd use it inline, shadowing std::Result::Ok feels a bit dirty
| let differing: Vec<&'static str> = [ | ||
| ("container_id", dest.container_id == src.container_id), | ||
| ("language_name", dest.language_name == src.language_name), | ||
| ( | ||
| "language_version", | ||
| dest.language_version == src.language_version, | ||
| ), | ||
| ("tracer_version", dest.tracer_version == src.tracer_version), | ||
| ("runtime_id", dest.runtime_id == src.runtime_id), | ||
| ("env", dest.env == src.env), | ||
| ("hostname", dest.hostname == src.hostname), | ||
| ("app_version", dest.app_version == src.app_version), | ||
| ("attributes", dest.attributes == src.attributes), | ||
| ] | ||
| .into_iter() | ||
| .filter_map(|(label, eq)| (!eq).then_some(label)) | ||
| .collect(); | ||
|
|
||
| if !differing.is_empty() { | ||
| warn!( | ||
| "Skipping V1 TracerPayload append: diverging metadata fields {:?}", | ||
| differing | ||
| ); | ||
| return false; | ||
| } | ||
| true |
There was a problem hiding this comment.
This materializes the Vec even if it's not required. In those cases, it's better to optimize for the happy path:
| let differing: Vec<&'static str> = [ | |
| ("container_id", dest.container_id == src.container_id), | |
| ("language_name", dest.language_name == src.language_name), | |
| ( | |
| "language_version", | |
| dest.language_version == src.language_version, | |
| ), | |
| ("tracer_version", dest.tracer_version == src.tracer_version), | |
| ("runtime_id", dest.runtime_id == src.runtime_id), | |
| ("env", dest.env == src.env), | |
| ("hostname", dest.hostname == src.hostname), | |
| ("app_version", dest.app_version == src.app_version), | |
| ("attributes", dest.attributes == src.attributes), | |
| ] | |
| .into_iter() | |
| .filter_map(|(label, eq)| (!eq).then_some(label)) | |
| .collect(); | |
| if !differing.is_empty() { | |
| warn!( | |
| "Skipping V1 TracerPayload append: diverging metadata fields {:?}", | |
| differing | |
| ); | |
| return false; | |
| } | |
| true | |
| // let fields: [ | |
| // ("container_id", dest.container_id == src.container_id), | |
| // ... | |
| // ]; | |
| if fields.iter().any(|(_, eq)| !eq) { | |
| warn!( | |
| "Skipping V1 TracerPayload append: diverging metadata fields {:?}", | |
| fields | |
| .iter() | |
| .filter_map(|(label, eq)| (!eq).then_some(*label)) | |
| .collect::<Vec<_>>() | |
| ); | |
| return false; | |
| } | |
| true |
There is 2 passes on the logging branch but it's cheaper on the happy path
| } | ||
|
|
||
| /// Generic over the deserialization mode (owned `BytesData` or borrowed `SliceData`). | ||
| #[allow(clippy::type_complexity)] |
There was a problem hiding this comment.
The types here don't look too complex, it's probably due to some kind of bad practice in the types themselves that I don't see. Or if not it's better to type alias some things instead of clippy allow imho, like type DecodedPayload<T> = (TracerPayload<T>, usize); and use it in the return type
| let count = decode::read_array_len(buf.as_mut_slice()) | ||
| .map_err(|_| DecodeError::InvalidFormat("V1 chunks array len read failure".to_owned()))?; | ||
| let mut chunks = Vec::with_capacity(count as usize); |
There was a problem hiding this comment.
How trusted can we be about the inputs ? If a malformed/malicious payload sets the count to 0xffffffff we'd have problems. Maybe min between count and buf.len().
I'm noting it here but I've seen this pattern a few times elsewhere btw.
| let v: i64 = decode::read_int(buf.as_mut_slice()).map_err(|_| { | ||
| DecodeError::InvalidFormat("V1 chunk priority read failure".to_owned()) | ||
| })?; | ||
| chunk.priority = Some(v as i32); |
There was a problem hiding this comment.
In these cases, i32::try_from is safer. If you are sure no wrapping can occur this is fine though.
There was a problem hiding this comment.
I've also seen this for sampling_mechanism and flags
@ajgajg1134 @anais-raison I think doing this locally as a one-off before merging the PR is a good idea. I would not recommended as part of this PR we have an automated CI workflow that spins up a real agent. If we can accomplish with just comparing files then we should do that. |
What does this PR do?
Adds a V1 msgpack decoder in
libdd-trace-utilsand wires the V1 variant through the trace pipeline so any V1 payload can be decoded, inspected, and re-encoded.Motivation
APMSP-2813 : Prerequisite for the sidecar V1-native path. Splitting out the
libdd-trace-utils/libdd-data-pipelineplumbing so it can land before the sidecar wiring.Additional Notes
trace_serializer.rsnow dispatches on(TraceChunks, OutputFormat); the existing v0.4 → V1 cross-encode path is preserved.trace_utils::collect_trace_chunksrenamed toconvert_trace_chunks_v04_to_v05.