Parsed JSON and Grokked Strings in the same pipeline

Elastic Stack Logstash
1

The flow of data in my system is described below:

  1. Files on my filesystem, some log, some jsonlog
  2. Filebeats
    a. Human readable log lines
    b. JSON log lines
  3. Logstash
    a. grok

Something is coercing the JSON object from my jsonlog lines into a format that looks like a Ruby hash.

I would like to be able to handle the different data paths dynamically. I don't think I need multiple pipelines or filebeat configurations, but I could be wrong.

2

Something is coercing the JSON object from my jsonlog lines into a format that looks like a Ruby hash.

How so? Where are you seeing this?

3

I was seeing it in the output generated by Logstash.

I also managed to fix the problem. I don't really understand how I did it, I've just found the jumble of incantations that outputs things the way I want them to be.

I'm not really sure this actually works either, but somehow it has managed to work...

This is my filebeat configuration:

filebeat:
  inputs:
  - type: log
    recursive_glob:
      enabled: true
    paths: /var/log/**/*.log
  - type: log
    recursive_glob:
      enabled: true
    paths: /var/log/**/*.jsonlog
    processors:
    - rename: # this must be broken or incorrect, because I both don't want it to happen and it isn't happening...
        fields:
          from: "message"
          to: "message_json"

logging:
  files:
    name: filebeat.log

output:
  logstash:
    hosts:
    - # hostname

This is my logstash configuration:

input {
  beats {
    id => "input.beats.dev"
    port => 5044
    add_field => { environment => "dev" }
  }
}

filter {
  mutate {
    rename => { "source" => "source_path" }
    copy => { "message" => "unparsed" }
  }
  grok {
    match => {
      "message" => [
        # A list of grok patterns
      ]
    }
    overwrite => ["message"]
  }
  if "_grokparsefailure" in [tags] {
    json {
      source => "unparsed"
      skip_on_invalid_json => true
    }
  }
  json {
    source => "metrics_json"
    remove_field => ["metrics_json"]
  }
  date {
    match => [ "timestamp", "ISO8601", "yyyy-MM-dd HH:mm:ss" ]
    remove_field => ["timestamp"]
  }
  date {
    match => [ "timestamp", "ISO8601", "yyyy-MM-dd HH:mm:ss" ]
    remove_field => ["timestamp"]
  }
  mutate {
    uppercase => ["level"]
    remove_field => ["unparsed"]
  }
}

output {
  stdout { codec => json_lines }
}