logstash.conf

input {
    syslog {
        port => 29124
    }
}
filter {
    grok {
        pattern_definitions => {
            "TIMESTAMP_NGX" => "%{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{HOUR}:?%{MINUTE}(?::?%{SECOND})"
        }
        match => {
          "message" =>
          "%{TIMESTAMP_NGX} \[%{LOGLEVEL}] %{NUMBER}\#%{NUMBER}: pid:%{NUMBER:pid:int}\|from:(?<from>\d{10})\|to:(?<to>\d{10})\|accounting_id:(?<accounting_id>[^|]+)\|%{WORD:entry_type}:%{NUMBER:nr_entries:int}\|bytes_in:%{NUMBER:in_bytes:int}\|bytes_out:%{NUMBER:out_bytes:int}\|latency_ms:%{NUMBER:latency_ms:int}\|upstream_latency_ms:%{NUMBER:upstream_latency_ms:int}\|%{GREEDYDATA:statuses}"
        }
        remove_field => [ "host", "severity", "facility", "priority", "severity_label", "facility_label" ]
        remove_tag => []
    }
    # date {
    #   match => [ "timestamp" , "yyyy/MM/dd HH:mm:ss" ]
    #   target => '@timestamp'
    #   # timezone => 'Asia/Shanghai'
    #   remove_field => [ "timestamp" ]
    # }
    date {
      match => [ "from" , "UNIX" ]
      target => '@from'
      remove_field => [ "from" ]
    }
    date {
      match => [ "to" , "UNIX" ]
      target => '@to'
      remove_field => [ "to" ]
    }
    kv {
        source => "statuses"
        target => "statuses_kv"
        field_split => "|"
        value_split => ":"
        remove_field => [ "statuses" ]
    }
    ruby {
        code => "s={};agg={};agg.default=0;(event.get('statuses_kv')||{}).each{|k,v|agg['%dxx'%k[0]]+=v.to_i;s[k]=v.to_i};event.set('@agg_status',agg);event.set('@nr_status',s)"
        remove_field => [ "statuses_kv" ]
    }
}
output {
  stdout { codec => rubydebug }
  elasticsearch {
    hosts => ["http://elasticsearch:9200"]
    manage_template => true
    template_overwrite => true
    template => "/etc/logstash/es_template_accounting.json"
    index => "ngx_accounting-%{+xxxx.ww}"
  }
}