Logstash RabbitMQ Wrap Incoming Message

Elastic Stack Logstash
1

I'm trying to wrap the incoming message off of Rabbit and then have some extra metadata outside of the wrapped message. Can this be done?

Hoping for an end result like:
errorMessage:{ ... },
trackingID:{ ... }
headerProperties:{ ... }
message:{...}

filter {
mutate {
rename => [ "@message", "message" ]
 
    add_field => {"errorMessage" => "%{[@metadata][rabbitmq_headers][errorMessage]}"}
    add_field => {"headerProperties" => "%{[@metadata][rabbitmq_properties]}"}
    add_field => {"trackingID" => "%{[@metadata][rabbitmq_headers][trackingId]}"}
}
}
2

What you have looks reasonable. In what way is it not working? What does your configuration look like? Show sample output from a stdout { codec => rubydebug { metadata => true } } output.

3

Thanks for the quick response.

Heres a message thats saved in elastic and is the same with the standard output:

{
      "reportsource": {
        "active": true,
        "_id": "1d38ed20-39c5-11e7-a4de-8b50391cf1ee",
        "updatedDate": "2017-10-31 16:10:08.260"
       },
      "errorMessage": "NodeNotConnectedException Node not connected]",
      "description": "N",
      "active": true,
      "sourceUpdated": 1469145600000,
      "updatedDate": "2017-10-31 12:51:42.656",
      "@timestamp": "2017-10-31T19:39:22.900Z",
      "@version": "1",
      "headerProperties": "{\"delivery-mode\":2,\"content-type\":\"application/json\",\"exchange\":\"error.exchange\",\"routing-key\":\"report\",\"consumer-tag\":\"amq.ctag-Q9vmGJ0GVH9zAcPkW9eCEg\",\"priority\":0}",
      "sourceCreated": 1468339661000,
      "class": "report"
    
  }

I just can't get the message to move into the new field. It treats the entire message off of rabbit as JSON and doesn't wrap it to the new field

4

Not sure what you're trying to do here. What does your configuration look like? And what would you like the message above to look like instead?

5

@magnusbaeck apologies for not responding until now. Here is a sample configuration:

input {
  rabbitmq {
  host => "local
  queue => "Error"
  durable => true
  threads => 3
  metadata_enabled => true
  prefetch_count => 50
  type => 'rabbit-errors'
  }
}

filter {
mutate {
rename => [ "@message", "message" ]
    rename => ["log", "message"]
    add_field => {"errorMessage" => "%{[@metadata][rabbitmq_headers][errorMessage]}"}
    add_field => {"headerProperties" => "%{[@metadata][rabbitmq_properties]}"}
    add_field => {"trackingID" => "%{[@metadata][rabbitmq_headers][trackingId]}"}
}
}



output {
  elasticsearch {
   }
}

which will print out the message shown above. I would prefer the errorMessage, and headers, to live outside of the actual message from the Rabbit queue. Something like:

{
      "errorMessage": "NodeNotConnectedException Node not connected]",
      "@timestamp": "2017-10-31T19:39:22.900Z",
      "@version": "1",
      "headerProperties": "{......}",
      messgage: {
          "reportsource": {
             "active": true,
             "_id": "1d38ed20-39c5-11e7-a4de-8b50391cf1ee",
             "updatedDate": "2017-10-31 16:10:08.260"
          },
          "description": "N",
          "active": true,
          "sourceUpdated": 1469145600000,
          "updatedDate": "2017-10-31 12:51:42.656",
          "sourceCreated": 1468339661000,
          "class": "report"
      }
  }
6

Okay. Use a mutate filter to move (rename) all the fields into being subfields of message. If you can't enumerate all the fields to move (because they're dynamic in nature) you need to use a ruby filter (i.e. there's no wildcard-based moving of fields).

7

Got it! Thank you, I'll read more into the Ruby filter.

8

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.