input {
  tcp {
    port => {{ logstash_syslog_port }}
    type => syslog
  }
  udp {
    port => {{ logstash_syslog_port }}
    type => syslog
  }
}

filter {
  if [type] == "syslog" {
    grok {
      match => { "message" => "<%{POSINT:syslog_pri}>%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{@source_host}" ]
    }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
    # Try and parse the tracking log json
    # 142 is syslog facility 17 (local1) and Informational.
    # This is used to reduce the number of errors in json parsing as
    # tracking uses that facility and priority by default.
    if "142" in [syslog_pri] {
      json {
        source => "syslog_message"
        target => "tracking"
      }
    }
    if !("_grokparsefailure" in [tags]) {
      mutate {
        replace => [ "@source_host", "%{syslog_hostname}" ]
        replace => [ "@message", "%{syslog_message}" ]
      }
    }
    mutate {
      remove_field => [ "syslog_hostname", "syslog_message", "syslog_timestamp" ]
    }
  }
}

output {
  # Example just to output to elasticsearch
  elasticsearch { }
  # And gzip for each host and program
  file {
       path => '{{ logstash_data_dir }}/%{@source_host}/all.%{+yyyyMMdd}.gz'
       gzip => true
  }
  # Should add option for S3 as well.
}
