enhancement(amqp source): expose message headers and properties in events (#23368)

[KimJS0328]

Summary

Exposes AMQP 0.9.1 message headers and BasicProperties on events emitted by the amqp source, and fixes the message timestamp to be interpreted as Unix seconds (per the AMQP spec) instead of being misinterpreted.

• Adds two new configuration options: headers_key (default: headers) and properties_key (default: properties). Both can be disabled by setting them to an empty value, following the same pattern as the existing routing_key_field / exchange_key / offset_key.
• headers is populated from the user-defined FieldTable via a recursive AMQPValue → Value conversion, preserving nested tables/arrays and supported scalar types.
• properties is populated with the well-known AMQP BasicProperties scalar fields (content_type, content_encoding, delivery_mode, priority, correlation_id, reply_to, expiration, message_id, timestamp, type, user_id, app_id, cluster_id). headers is intentionally omitted from properties to avoid duplication with the top-level headers key.
• AMQP timestamps (both the message-level BasicProperties::timestamp and any AMQPValue::Timestamp values inside headers/properties) are now interpreted as Unix seconds and converted into proper chrono::DateTime<Utc> values. The schema definitions for both Vector and Legacy log namespaces are updated accordingly.

Vector configuration

sources:
  rabbitmq_events:
    type: amqp
    connection_string: "amqp://guest:guest@127.0.0.1:5672/%2f"
    queue: vector.rabbitmq.events
    consumer: vector-rabbitmq-event-consumer

    # new options (default values)
    headers_key: headers
    properties_key: properties

sinks:
  stdout:
    type: console
    inputs: [rabbitmq_events]
    encoding:
      codec: json
      json:
        pretty: true

Example event emitted (Legacy namespace):

{
  "exchange": "amq.rabbitmq.event",
  "headers": {
    "peer_addr": "12.345.67.89",
    "peer_port": 45842,
    "recv_bytes": 844,
    "route": "{}",
    "send_bytes": 150830,
    "sock_addr": "12.123.45.67",
    "sock_port": 25672,
    "timestamp_in_ms": 1779688712659
  },
  "message": "",
  "offset": 555,
  "properties": {
    "delivery_mode": 2,
    "timestamp": "2026-05-25T05:58:32Z"
  },
  "routing": "node.node.stats",
  "source_type": "amqp",
  "timestamp": "2026-05-25T05:58:32Z"
}

How did you test this PR?

• Unit tests added in src/sources/amqp.rs:
  • output_schema_definition_vector_namespace / output_schema_definition_legacy_namespace updated to cover the new headers and properties metadata fields.
  • amqp_field_table_to_value_preserves_supported_types covers the recursive FieldTable → Value conversion including nested tables, arrays, strings/bytes, integers, booleans, and timestamps.
  • basic_properties_to_value_includes_scalar_properties covers the scalar property extraction.
  • amqp_timestamp_to_datetime_uses_unix_seconds covers the Unix-seconds timestamp parsing.
• Integration test amqp_source_consume_event_with_headers_and_properties added in the existing amqp-integration-tests module — publishes a message with custom headers, content type, priority, and timestamp via RabbitMQ and asserts the resulting event shape.
• Manually ran Vector locally via cargo run against a real RabbitMQ broker, consumed live events from a queue with the configuration shown above, and verified that headers, properties, and the corrected timestamp appear on the emitted events as expected (sample output included in the "Vector configuration" section).
• Verified locally:
  • make fmt
  • make check-fmt
  • make check-clippy
  • make check-markdown
  • make check-generated-docs
  • ./scripts/check_changelog_fragments.sh

Change Type

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

Is this a breaking change?

• Yes
• No

The new headers_key and properties_key options default to headers and properties respectively, which adds new top-level fields to emitted events. This is consistent with how the existing routing_key_field / exchange_key / offset_key already behave and follows the same opt-out pattern (setting the key to empty disables the field).

The timestamp fix changes the AMQP message timestamp from a previously incorrect value into a correctly interpreted Unix-seconds DateTime. Users relying on the prior (incorrect) behavior would see corrected timestamp values.

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.

Changelog fragments included:

• changelog.d/23368_amqp_source_headers_properties.feature.md
• changelog.d/23368_amqp_source_timestamp.fix.md

References

• Closes: Expose AMQP message properties (headers) #23368

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:
    • make fmt
    • make check-clippy (if there are failures it's possible some of them can be fixed with make clippy-fix)
    • make test
• 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 make build-licenses to regenerate the license inventory and commit the changes (if any). More details on the dd-rust-license-tool.

[github-actions]

All contributors have signed the CLA ✍️ ✅
_{Posted by the CLA Assistant Lite bot.}

[KimJS0328]

I have read the CLA Document and I hereby sign the CLA