Using the HTTP source or VRL is there any way to store the source message on a separate field?

[leandrojmp]

Hello,

I'm using vector to build something like an ingestion layer to make it easier to onboard new applications on our SIEM, for this I'm using the http source and the kafka sink.

Any request made the http source in the path /logs/someName will create a topic in kafka named some_prefix-someName.

I'm using json in both the source and the sink and the main issue I'm having right now is that the http source will save the fields from the source message in the root of the event, and the timestamp field that the http source adds can conflict with a timestamp field from the source message.

I know that if the source message has a timestamp field it will override the timestamp added by vector, but with this I lose the information of the vector timestamp.

Is there any way to store the source message received by the http source in a different field? Or maybe tell vector to use a different field name for its timestamp field?

My config is something like this:

# sources

sources:
  logs:
    type: http_server
    address: 0.0.0.0:30000
    encoding: json
    host_key: ""
    method: POST
    path: /logs
    path_key: vector_request_path
    response_code: 200
    strict_path: false

# transform

transforms:
  topic:
    inputs: [ "logs" ]
    type: remap
    drop_on_error: true
    source: |
      . |= parse_groks!(
                .vector_request_path,
                patterns: [
                    "^/logs/%{DATA:log_topic}"
                ]
            )
      del(.source_type)
      del(.vector_request_path)

  event:
    inputs: ["*"]
    type: filter
    condition:
      type: "vrl"
      source: ".log_topic != null"

# sinks

sinks:
  kafka:
    type: kafka
    inputs: ["event"]
    bootstrap_servers: 127.0.0.1:29092
    topic: "some_prefix-{{log_topic}}"
    encoding:
      codec: json

The sample document I'm testing is this one:

[ {"field": "value1", "anotherField": "anotherValue"},
{"field": "value2", "extraField": ["value2.1", "value2.2"]},
{"field": "value3", "timestamp": "override-vector-timestamp"}
]

And this how it looks like in Kafka

I'm looking for to have something like this in Kafka:

{
	"message": {
        "anotherField": "anotherValue",
	    "field": "value1"
    },
	"log_topic": "sample",
	"timestamp": "2024-10-24T04:00:58.691623361Z"
}

{
	"message": { 
        "extraField": [
		    "value2.1",
		    "value2.2"
	    ],
	    "field": "value2"
    },
	"log_topic": "sample",
	"timestamp": "2024-10-24T04:00:58.691623361Z"
}

{
	"message": { 
        "field": "value3",
        "timestamp": "override-vector-timestamp"
    },
    "log_topic": "sample",
	"timestamp": "2024-10-24T04:00:58.691623361Z"
}

[leandrojmp]

After some searchs I found this guide and used this in my configuration:

log_schema:
  timestamp_key: "vector_timestamp"

With this I'm now at least able to change the name of the timestamp field that vector generates, but I would still like to have the source message in a separate field.

I tried to add the setting message_key: "vector_message" to the log_schema, but it has no effect.