logstash.conf

# Where to get input
input {
  # syslog inputs
  tcp {
    port => 5000
    tags => ["system"]
    type => "syslog"
  }
  udp {
    port => 5000
    tags => ["system"]
    type => "syslog"
  }
  syslog {
    port => "6514"
    tags => ["system"]
    type => "syslog"
  }

  # CoreOS journal input
  tcp {
    codec => "line"
    port => 5004
    tags => ["coreos","docker"]
    type => "systemd"
  }

  # Logspout input
  udp {
    codec => "plain"
    port => 5006
    tags => ["docker"]
    type => "logspout"
  }

  # Log4j application input
  tcp {
    port  => 1095
    tags  => ["applogs","service","1095"]
    type  => "json"
  }
  tcp {
    port  => 4560
    tags  => ["applogs","service","4560"]
    type  => "json"
  }
  log4j {
    port  => 4561
    tags  => ["applogs","service","4561"]
    type  => "log4j"
  }
}

# Some Filtering
filter {
  # syslog/systemd filter
  if [type] == "syslog" or [type] == "systemd" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{SYSLOGPROG}: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
    }
    syslog_pri { }
    date { match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ] }

    if !("_grokparsefailure" in [tags]) {
      mutate {
        replace => { "message" => "%{syslog_message}" }
        remove_field => [  "syslog_message", "syslog_program" ]
      }
    }

    # Remove spurious fields that have names changed or been aggregated
    mutate { remove_field => [ "syslog_hostname", "syslog_timestamp" ] }
  }

  # Docker filter
  if [program] == "dockerd" {
    kv {
      source => "message"
      prefix => "docker_"
    }
    mutate {
      rename => { "docker_level" => "docker_loglevel" }
      replace => { "message" => "%{docker_msg}" }
    }
    mutate { remove_field => [ "docker_msg", "docker_time" ] }
#    mutate {
#      replace => { "message" => "time=\"%{TIMESTAMP_ISO8601:docker_ts}\" level=%{LOGLEVEL:docker_loglevel} msg=\"%{GREEDYDATA:docker_msg}\"" }
#    }
  }

  # Logspout filter
  if [type] == "logspout" {
    grok {
      match => { "message" => "%{SYSLOG5424PRI}%{NONNEGINT:ver} +(?:%{TIMESTAMP_ISO8601:ts}|-) +(?:%{HOSTNAME:containerid}|-) +(?:%{NOTSPACE:containername}|-) +(?:%{NOTSPACE:proc}|-) +(?:%{WORD:msgid}|-) +(?:%{SYSLOG5424SD:sd}|-|) +%{GREEDYDATA:msg}" }
    }
    syslog_pri { }
    date { match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ] }
    if !("_grokparsefailure" in [tags]) {
      mutate {
        replace => { "@source_host" => "%{syslog_hostname}" }
        replace => { "message" => "%{syslog_message}" }
      }
    }

    # Remove spurious fields that have names changed or been aggregated
    mutate { remove_field => [ "syslog_hostname", "syslog_message", "syslog_timestamp" ] }
  }
  
  if [type] == "log4j" {
    mutate {
      rename => { "priority" => "level" }
      }
  }
  
  if [type] == "json" {
    json {
      source => "message"
    }
  }
  
  # Add GeoIP
  geoip { source => "%{IPORHOST}" }
}

# Where to send output
output {
  # Send output to standard output device/interface
  stdout {
    codec => rubydebug
  }

  # Parse failed messages to separate index
  if "_grokparsefailure" in [tags] or "_jsonparsefailure" in [tags] {
    amazon_es {
    # host => ["localhost:9200"]
      hosts => ["ES_CONN_STR"]
      region => "ES_REGION"
      index => "parse-err-%{+YYYY.MM.dd}"
    }
  }

# Elasticsearch output
  amazon_es {
  # host => ["localhost:9200"]
    hosts => ["ES_CONN_STR"]
    region => "ES_REGION"
  }
}