logstash-test.conf

# Where to get input
input {
  # syslog inputs
  tcp {
    port => 5000
    type => "syslog"
  }
  udp {
    port => 5000
    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
  log4j {
    codec => "json_lines"
    port  => 5200
    tags  => ["applogs","service"]
    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"
      remove_field => [ "time" ]
      prefix => "docker_"
    }
    mutate { rename => { "docker_level" => "docker_loglevel" } }
#    mutate {
#      replace => { "message" => "time=\"%{TIMESTAMP_ISO8601:docker_ts}\" level=%{LOGLEVEL:docker_loglevel} msg=\"%{GREEDYDATA:docker_msg}\"" }
#    }
    if !("_grokparsefailure" in [tags]) {
      mutate {
        replace => { "message" => "%{docker_msg}" }
        remove_field => [ "%{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" ] }
  }

  # 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] {
    elasticsearch {
    # host => ["localhost:9200"]
      host => ["ES_CONN_STR"]
      index => "parse-err-%{+YYYY.MM.dd}"
      protocol  => "http"
    }
  }

# Elasticsearch output
  elasticsearch {
  # host => ["localhost:9200"]
    host => ["ES_CONN_STR"]
    index => "logstash-%{+YYYY.MM.dd}"
    protocol  => "http"
  }
}