Add a parsable logstash filter

- Convert the inner JSON encoded messages into useful/indexed entries
- Ensure the timestamp is human readable
- Remove unneeded fields and allow easier filtering

Author: Amin Jamali

examples/logstash-filters.conf

@@ -1,19 +1,51 @@
-# standard rfc5424 parsing
-grok {
-  match => [ "message", "%{SYSLOG5424LINE}" ]
-}
+filter {
+  # standard rfc5424 parsing
+  grok {
+    match => [ "message", "%{SYSLOG5424LINE}" ]
+  }
 
-syslog_pri {}
+  syslog_pri {}
 
-# extract BOSH instance metadata from structured data
-if [syslog5424_sd] =~ "\[instance@47450" {
-  grok {
-    match => [ "syslog5424_sd", "\[instance@47450 %{DATA:instance_raw}\]" ]
+  mutate {
+    rename => ["syslog5424_app", "job"]
+  }
+  mutate {
+    rename => ["syslog5424_host", "host_address"]
+  }
+  date {
+    match => [ "syslog5424_ts", "ISO8601" ]
+    target => "@timestamp"
+    remove_field => "syslog5424_ts"
   }
 
-  kv {
-    source => "instance_raw"
-    target => "syslog5424_sd_instance"
-    remove_field => "instance_raw"
+  # extract BOSH instance metadata from structured data
+  if [syslog5424_sd] =~ "\[instance@47450" {
+    grok {
+      match => [ "syslog5424_sd", "\[instance@47450 %{DATA:source_raw}\]" ]
+    }
+
+    kv {
+      source => "source_raw"
+      target => "source"
+      remove_field => "source_raw"
+    }
+
+    mutate {
+      remove_field => "syslog5424_sd"
+    }
+  }
+
+  if [syslog5424_msg] =~ /^{.*}/ {
+    json {
+      source => "syslog5424_msg"
+      target => "parsed_json"
+      remove_field => "syslog5424_msg"
+      add_tag => [ "json" ]
+    }
+
+    date {
+      match => ["[parsed_json][timestamp]", "ISO8601", "UNIX"]
+      target => ["[parsed_json][timestamp]"]
+    }
   }
 }