reduce message field from elasticsearch

[uglyliu]

Kafka log content is about as follows:
{ "@metadata": { "beat": "filebeat", "type": "_doc", "version": "7.10.0" }, "message": { "upstream_response_time": "0.056", "upstream_status": "200" } }
And Finally, there will be more message fields written in elasticsearch，as follows

The vector configuration is as follows
[transforms.parse_logs]
type = "json_parser"
inputs = ["dummy_logs"]
source = '''
. = parse_json!(.message)
.message = parse_json!(.message)
'''
My purpose is to finally display on elasticsearch, there is no message field，as follows

[monwolf]

I'm doing this on filebeat itself:

processors:
  - decode_json_fields:
      fields: ['message']
      target: "json"
  - rename:
      fields:
        - from: json.@timestamp
          to: json.log_timestamp
      ignore_missing: false
      fail_on_error: true
  - move_fields:
      from: "json"
      to: ""
  - drop_fields:
      fields: ['message', 'agent', 'ecs', 'input', 'log', 'json']

This could work for you as well.

[uglyliu]

thank u

[spencergilbert]

In your remap component you can use the merge() function to combine your parsed message with the root of the event. A direct translation of @monwolf's Filebeat configuration should look something like.

transforms:
  parse_logs:
    type: remap
    inputs: ["dummy_logs"]
    source: |
      json = parse_json!(.message)
      json.log_timestamp = del(json.@timestamp)
      merge(., object!(json))
      del(.message)
      del(.agent)
      del(.ecs)
      del(.input)
      del(.log)
'''