fix(datadog_logs sink): Apply agent-json header on events from agent

[graphcareful]

Summary

When routing datadog_agent logs through Vector before sending to the datadog logs intake, users have complained that they lose contextual log level information.

The cause was determined to be a header DD-PROTOCOL: agent-json, the agent prepares HTTP requests to the logs backend with this header while Vector does not.

This header lets the logs intake take precedence of the attributes that are nestedwithin the 'message' attribute. Therefore things like log_level will depend on values arriving from the users application. Without the header the logs backend falls back on using the value applied at the root of the message, which is usually a value of info or error set by the datadog agent dependent on whether the log was emitted via stdout/stderr.

The remediation is to have Vector apply the HTTP header, however conditionally. This will occur only if the event had originated from the datadog_agent and if the user hadn't applied any transforms to remove or modify reserve attributes in a non standard way.

Vector will partition events on the two aformentioned conditions, events which do not need these conditions will still be sent to datadogs logs backend however without the DD-PROTOCOL: agent-json header applied.

Change Type

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

Is this a breaking change?

• Yes
• No

How did you test this PR?

Tested by sending data to datadog from a service I had created that wrote messages to a local file that looked like this:

{"level": "info", "message": "This is sample message: 50", "time": 1742513705.3499498}
{"level": "error", "message": "This is sample message: 72", "time": 1742513706.355413}
{"level": "warn", "message": "This is sample message: 10", "time": 1742513707.3605719}

Before the change it can be seen that all of the logs in datadog have the status of info and after the change the UI shows logs of type info, error and warn, respecting the level field.

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.

Checklist

• Please read our Vector contributor resources.
  • make check-all is a good command to run locally. This check is
    defined here. Some of these
    checks might not be relevant to your PR. For Rust changes, at the very least you should run:
    • cargo fmt --all
    • cargo clippy --workspace --all-targets -- -D warnings
    • cargo nextest run --workspace (alternatively, you can run cargo test --all)
• If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
  run dd-rust-license-tool write to regenerate the license inventory and commit the changes (if any). More details here.

References

#13291

[bruceg]

Makes sense to me.

[graphcareful]

I think this change is also missing one component. We must also verify the message originated from the datadog agent source

[pront]

Will take another look when this #22701 (comment) is resolved.

[graphcareful]

Will take another look when this #22701 (comment) is resolved.

That actually has been implemented here

[graphcareful]

Force pushed changes that modify the previous solution. This second pass allows Vector to forcefully normalize the payload to the agent format (conditionally upon detection of settings that enable this behavior). The logic will move any non reserved fields to be nested under the message key. If there are any collisions during this process they will be stored under message._collisions. In these cases !warn or !errors may be logged to alert the user that this has been performed. This should address the concerns of data loss that @pront and @brent-hronik had mentioned.

myself

[brent-hronik]

agree with the existing comments that @pront has

[pront]

Are there any Datadog docs we can link to here?

[graphcareful]

@brent-hronik are there any docs that explain specifically the agent message format?

[bruceg]

I have a question about the efficiency of the approach below. Otherwise LGTM

[bruceg]

Could this process be done the opposite way:

1. take the root object, replacing it with an empty object
2. For each key in the reserved attributes, remove it from the former root into the new root.
3. Insert the remainder into message and re-insert message into the root.

This eliminates the repeated scans over the reserved attributes and the creation of any temporaries.

[graphcareful]

I just prototyped this, ends up being more code (2 loops one for step 2 and one for step 3) and the temporaries (keys_to_move) I believe must stay. That is because you cannot call .remove on a map that you're iterating over and for step 3 (above) we would have to iterate over all remaining keys, calling .remove on the same map. The main drawback being repeated scans over the reserved attrs list.

[bruceg]

What I was thinking was something like this, which seems different than what you described:

let old_root = std::mem::take(&object_map);
for key in DD_RESERVED_SEMANTIC_ATTRS {
    if let Some((key, value)) = old_root.remove_entry(key) {
        object_map.insert(key, value); // will never be `Some`
    }
}
for (key, value) in old_root {
    if let Some(returned_entry_v) = message.insert(key, value) {
        collisions.insert(key, returned_entry_v);
    }
}
object_map.insert(MESSAGE, message);

[bruceg]

Should these be internal events that increment a metric as well? The formatting is also kinda funky.

[graphcareful]

I felt that the workflow would be for the user to observe these, then modify their processors to avoid collisions.

[vladimir-dd]

Sorry for the late feedback — I realize the PR is already closed, but I wanted to quickly share an idea for future improvements.

After this PR, the behavior looks like:

Input:

{
  "a": 1,
  "status": "info",
  "message": {
    "a": 2
  }
}

Output:

{
  "status": "info",
  "message": {
    "a": 2,
    "_collisions": {
      "a": 1
    }
  }
}

Drawbacks:

• Introduces a non-standard _collisions field that users are not expecting.

Proposed solution:

Instead of creating _collisions, move all custom attributes and the original message into a new top-level message field, leaving reserved fields at the root.
Result:

{
  "status": "info",
  "message": {
    "a": 1,
    "message": {
      "a": 2
    }
  }
}

Why:

• No new fields are invented.
• The Logs Intake already parses the message recursively and naturally handles any conflicts.
• This preserves the behavior users were familiar with before the change.

[graphcareful]

Sorry for the late feedback — I realize the PR is already closed, but I wanted to quickly share an idea for future improvements.

Thank you, feedback implemented here