logstash config - extract data from ceph logs

logstash.conf

input {
  gelf {
    port => 12222
    type => gelf
    codec => json { }
  }
}

filter {
  if [type] == "gelf" {
    mutate {
      add_field => { "full_message" => "%{message}" }
      add_field => { "parsed" => true }
    }
    if [subsys_name] == "osd" {
      grok {
        match => [
             "message", "osd.%{INT:osd_whoami} %{INT:osd_epoch} %{GREEDYDATA:message}",
             "message", "osd.%{INT:osd_whoami} %{GREEDYDATA:message}",
             "message", "%{GREEDYDATA:message}"
        ]
        overwrite => [ "message" ]
      }
    } else if [subsys_name] == "filestore" {
     grok {
        match => [
             "message", "(?<filestore_backend>xfs|btrfs|generic|zfs)filestorebackend\(%{UNIXPATH:filestore_path}\) %{GREEDYDATA:message}",
             "message", "(?<store>mem|file)store\(%{UNIXPATH:filestore_path}\) %{GREEDYDATA:message}",
             "message", "filestore \((?<init>init)\)%{GREEDYDATA:message}",
             "message", "filestore %{GREEDYDATA:message}"
        ]
        overwrite => [ "message" ]
      }
    } else if [subsys_name] == "mds" {
     grok {
        match => [
             "message", "mds.%{INT:mds_nodeid}.%{DATA:mds_module} %{GREEDYDATA:message}",
             "message", "mds.beacon.%{DATA:mds_name} %{GREEDYDATA:message}",
             "message", "MDS(IO|Internal)ContextBase::complete: %{GREEDYDATA:mds_context_name}"
        ]
        overwrite => [ "message" ]
     }
    } else if [subsys_name] == "paxos" {
     grok {
        match => [
             "message", "mon.%{DATA:mon_name}@%{INT:mon_rank}\(%{DATA:mon_state}\).paxosservice\(%{DATA:mon_service_name} %{INT:mon_fc}..%{INT:mon_lc}\) %{GREEDYDATA:message}",
             "message", "mon.%{DATA:mon_name}@%{INT:mon_rank}\(%{DATA:mon_state}\).paxos\(%{DATA:mon_paxos_name} %{DATA:mon_paxos_state} c %{INT:mon_paxos_first_commited}..%{INT:mon_paxos_last_commited}\) %{GREEDYDATA:message}"
        ]
        overwrite => [ "message" ]
     }
    } else if [subsys_name] == "mon" {
     grok {
        match => [
             "message", "mon.%{DATA:mon_name}@%{INT:mon_rank}\(%{DATA:mon_state}\).%{DATA:mon_service_name} %{GREEDYDATA:message}",
             "message", "mon.%{DATA:mon_name}@%{INT:mon_rank}\(%{DATA:mon_state}\).%{DATA:mon_service_name} v%{INT:mon_version} %{GREEDYDATA:message}",
             "message", "mon.%{DATA:mon_name}@%{INT:mon_rank}\(%{DATA:mon_state}\) %{GREEDYDATA:message}",
             "message", "%{GREEDYDATA:message}"

        ]
        overwrite => [ "message" ]
     }
    } else if [subsys_name] == "ms" {
     grok {
        match => [
             "message", "-- %{HOSTPORT:msg_src} (?<msg_direction>\-\-\>|\<\=\=) %{HOSTPORT:msg_dest} -- %{GREEDYDATA:message}",
             "message", "%{GREEDYDATA:message}"
        ]
        overwrite => [ "message" ]
     }
    } else if [subsys_name] == "objecter" {
     grok {
        match => [
             "message", "osd.%{INT:osd_whoami}.objecter %{GREEDYDATA:message}"
        ]
        overwrite => [ "message" ]
     }
    } else if [subsys_name] == "monc" {
     grok {
        match => [
             "message", "log_channel\(%{WORD:monc_log_channel}\) log %{DATA:monc_log_prio} : %{GREEDYDATA:message}",
             "message", "monclient\((?<monc_hunting>hunting)\): %{GREEDYDATA:message}",
             "message", "monclient: %{GREEDYDATA:message}"
        ]
        overwrite => [ "message" ]
     }
    } else if [subsys_name] == "journal" {
     grok {
        match => [
             "message", "journal %{GREEDYDATA:message}"
        ]  
        overwrite => [ "message" ]
     }
    } else if [subsys_name] == "auth" {
    } else {
      mutate {
         add_field => { "parsed" => false}
      }  
    }
  }
}

output {
  if "_grokparsefailure" in [tags] {
     stdout { codec => rubydebug }
  }
  if ! [parsed] {
     stdout { codec => rubydebug }
  }
  gelf {
    host => "127.0.0.1"
  }
#  elasticsearch {
#  }
}