feat(mezmo_datadog_agent_parser): Create MezmoDatadogAgentParser #49

biblicalph · 2026-01-26T21:55:00Z

This PR adds support for transforming Datadog agent events into logs, metrics, traces and sketches.
This allows us to deprecate VRL handling of metrics while adding support for Datadog traces and sketches.

For reference the following protos apply:

Upstream of vector, the agent payload is intercepted, decoded and normalized. The transform thus expects TracerPayload item for trace events, Sketch for sketch metrics, MetricSeries for metrics and Log for logs.

The implementation is mostly based on the existing Datadog agent source implementation; we don't use that source because it exposes an http server to collect metrics

Ref: LOG-22997

Summary

Vector configuration

How did you test this PR?

Change Type

Bug fix
New feature
Non-functional (chore, refactoring, docs)
Performance

Is this a breaking change?

Yes
No

Does this PR include user facing changes?

Yes. Please add a changelog fragment based on our guidelines.
No. A maintainer will apply the no-changelog label to this PR.

References

Notes

Please read our Vector contributor resources.
Do not hesitate to use @vectordotdev/vector to reach out to us regarding this PR.
Some CI checks run only after we manually approve them.
- We recommend adding a pre-push hook, please see this template.
- Alternatively, we recommend running the following locally before pushing to the remote branch:
  - cargo fmt --all
  - cargo clippy --workspace --all-targets -- -D warnings
  - cargo nextest run --workspace (alternatively, you can run cargo test --all)
After a review is requested, please avoid force pushes to help us review incrementally.
- Feel free to push as many commits as you want. They will be squashed into one before merging.
- For example, you can run git merge origin master and git push.
If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
run cargo vdev build licenses to regenerate the license inventory and commit the changes (if any). More details here.

The transform receives a log event and outputs a single or multiple events to one of logs, metrics, traces or _unmatched ouptuts. The Datadog agent data is not decoded by this transform; the incoming event is assumed to be already decoded. For traces, each event is assumed to be an entry of the tracerPayloads or traces (legacy) field of the agent payload. For metrics, each event is assumed to be an entry of the series field of the agent payload. The implementation of the transform is mostly based on the Datadog agent source. Ref: LOG-22997

mdeltito · 2026-01-27T15:27:57Z

src/transforms/mezmo_datadog_agent_parser/common.rs

+    let secs = value.floor();
+    let fract = value - secs;
+    let mut nanos = (fract * 1_000_000_000.0).round();
+
+    // Handle rounding overflow (e.g., 0.9999999999 rounding up to 1s)
+    let secs = if nanos >= 1_000_000_000.0 {
+        nanos -= 1_000_000_000.0;
+        secs + 1.0
+    } else {
+        secs
+    };
+
+    Some((secs as i64, nanos as u32))


for seconds you likley want to truncate rather than floor (-1.4 is -1 whole seconds, not -2)

you can also cast early to avoid working with secs as a float later, and then use the checked_* methods for overflow checking:

// truncate towards 0 AND perform a saturating cast f64 -> i64 let secs = value.trunc() as i64; let frac = value - secs as f64; let mut nanos = (frac * 1_000_000_000.0).round() as i64; let mut secs = secs; if nanos >= 1_000_000_000 { nanos -= 1_000_000_000; secs = secs.checked_add(1)?; } else if nanos < 0 { // also handle the negative offset nanos += 1_000_000_000; secs = secs.checked_sub(1)?; } Some((secs, nanos as u32))

sounds good. I avoided trunc to not handle the negative values; that assumes all dates are positive

mdeltito · 2026-01-27T15:45:40Z

src/transforms/mezmo_datadog_agent_parser/metrics.rs

+    let message_obj = {
+        let log = event
+            .maybe_as_log_mut()
+            .ok_or_else(|| "Event is not a log".to_string())?;
+        get_message_object(log)?.clone()
+    };


this (and below for sketch) will unnecessarily clone the whole message which we pass by reference below to the transform_* functions.

I'm cloning the message mostly to ensure that if any step of the transformation fails, the original message is left intact and can be routed to unmatched

as it stands, the transform functions below (and in sketch) already take a &ObjectMap reference, so the cloning isn't preventing any mutation - we are cloning and then passing the clone by ref.

this is equivalent without a clone:

let message_obj = { let log = event .maybe_as_log_mut() .ok_or_else(|| "Event is not a log".to_string())?; get_message_object(log)? // no clone }; let output_messages = match version.as_str() { // it's already a reference, remove the `&` "v1" => transform_series_v1(message_obj)?, "v2" => transform_series_v2(message_obj)?, _ => transform_series_v2(message_obj)?, };

you can also swap .maybe_as_log_mut() for .maybe_as_log()... and then you will find that the event parameter itself doesn't even need to be mutable.

mdeltito · 2026-01-27T15:51:54Z

src/transforms/mezmo_datadog_agent_parser/metrics.rs

+        if let Some(interval) = interval {
+            time_obj.insert("interval_ms".into(), Value::from(interval * 1000));
+        }
+        output.insert("time".into(), Value::Object(time_obj));


if interval is None - is inserting an empty time object correct here?

Definitely missed that one. Should be in the if block

mdeltito · 2026-01-27T16:34:10Z

src/transforms/mezmo_datadog_agent_parser/mod.rs

+                let mut new_event = event.clone();
+                insert_message(&mut new_event, payload)?;
+                Ok(new_event)


here we clone the entire event (including .message) only to overwrite .message below in insert_message.

can we do the removal above outside of insert_message to pare down the event first before adding back the message?

mdeltito · 2026-01-27T16:46:22Z

src/transforms/mezmo_datadog_agent_parser/metrics.rs

+        if !tags.is_empty() {
+            output.insert("tags".into(), Value::Object(tags.clone()));
+        }


nit: construct a Some(Value::Object(tags)) once outside of this loop, then clone that instead

let tags: Option<Value> = if !tags.is_empty() { Some(Value::Object(tags)) } else { None }; //... for point in points { //... if let Some(tags) = &tags { output.insert("tags".into(), tags.clone()); } }

same is true in the other transform functions

yes this is def a nit so feel free to ignore - but consider:

// current - clone the map and wrap it in Value each iteration let tags = ... for foo in bar { output.insert("tags".into(), Value::Object(tags.clone())); } vs // suggested - clone the `Value` directly let tags = Value::Object(...) for foo in bar { output.insert("tags".into(), tags.clone()); }

mdeltito · 2026-01-27T16:49:25Z

src/transforms/mezmo_datadog_agent_parser/mod.rs

+                        if self.reroute_unmatched {
+                            output.push(Some(UNMATCHED_OUTPUT), event);
+                        }


is there a reason we only reroute logs to _unmatched on error? for the other types we just drop the input events.

ETA: I think the issue is really the ownership of the original event for these types where 1 input event becomes N outputs. One idea would be to have all the transform_* functions (including transform_log) take ownership of the event, and return an error type that carries back the input event, e.g.:

pub struct TransformError { pub message: String, pub input: Event, }

that way we can always push the input to _unmatched on errors

mdeltito · 2026-01-27T17:11:27Z

src/transforms/mezmo_datadog_agent_parser/traces.rs

+
+fn copy_timestamp_nanos(src: &ObjectMap, src_key: &str, dst: &mut ObjectMap, dst_key: &str) {
+    if let Some(nanos) = src.get(src_key).and_then(Value::as_integer) {
+        dst.insert(dst_key.into(), Value::Timestamp(Utc.timestamp_nanos(nanos)));


timestamp_nanos will panic when the data is out of range - maybe it would make sense to reuse common::parse_timestamp here?

biblicalph · 2026-01-27T23:48:31Z

src/transforms/mezmo_datadog_agent_parser/metrics.rs


-    parser.build_events_from_messages(event, output_messages)
+        parser
+            .build_events_from_payloads(event.clone(), output_messages)


Clone is still needed here because the method mutates the event

1. Reduce cost of cloning event when emitting multiple events for a received metric or trace 2. Implement trait for transforming event 3. Move event into transform to reduce cloning. 4. On error, move the error into the event 5. Route the original event to unmatched if transforming fails. Previously this was done for logs only. Ref: LOG-22997

The vector documentation now includes the two flags for Datadog logs and metrics - likely in newer Vector versions. This matches the implementation. Ref: LOG-22997

biblicalph self-assigned this Jan 26, 2026

biblicalph marked this pull request as draft January 26, 2026 22:03

biblicalph force-pushed the biblicalph/LOG-22997 branch 2 times, most recently from 39a967b to 89057ec Compare January 26, 2026 22:59

biblicalph marked this pull request as ready for review January 26, 2026 22:59

biblicalph force-pushed the biblicalph/LOG-22997 branch from 89057ec to 36ebe26 Compare January 26, 2026 23:08

mdeltito reviewed Jan 27, 2026

View reviewed changes

biblicalph commented Jan 27, 2026

View reviewed changes

biblicalph force-pushed the biblicalph/LOG-22997 branch 2 times, most recently from 3f41866 to c71ec4c Compare January 28, 2026 03:25

feat: Add parse_log_tags and split_metric_namespace flags

1313aa2

The vector documentation now includes the two flags for Datadog logs and metrics - likely in newer Vector versions. This matches the implementation. Ref: LOG-22997

biblicalph force-pushed the biblicalph/LOG-22997 branch from c71ec4c to 1313aa2 Compare January 28, 2026 04:01

feat(mezmo_datadog_agent_parser): Create MezmoDatadogAgentParser #49

Are you sure you want to change the base?

feat(mezmo_datadog_agent_parser): Create MezmoDatadogAgentParser #49

Conversation

biblicalph commented Jan 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Vector configuration

How did you test this PR?

Change Type

Is this a breaking change?

Does this PR include user facing changes?

References

Notes

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mdeltito Jan 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

biblicalph Jan 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

biblicalph commented Jan 26, 2026 •

edited

Loading

mdeltito Jan 27, 2026 •

edited

Loading

biblicalph Jan 27, 2026 •

edited

Loading