/etc/logstash/conf.d/01-collect-rsyslog.conf

01-collect-rsyslog.conf

### ref: https://github.com/logstash-plugins/logstash-input-syslog/issues/15#issuecomment-355655279

input {
  tcp {
    mode    => "server"
    host    => "192.168.56.10"
    port    => "5514"
    type    => "syslog"
  }
  udp {
    mode    => "server"
    host    => "192.168.56.10"
    port    => "5514"
    type    => "syslog"
  }
}


# determine and parse type of syslog message
filter {

  if [type] == "syslog" { 
  
    # look for and, if found, decode syslog priority
    if [message] =~ "^<[0-9]{1,3}>" { 
      grok {
        match => [ "message", "^<%{NONNEGINT:priority:int}>" ]
      }
      if [priority] <= 191 {
        # check for RFC 3164 vs RFC 5424
        if [message] =~ "^<[0-9]{1,3}>[0-9]{1,2} " {
          mutate {
            add_tag => ["syslog_rfc5424"]
          }
        }
        else {
          mutate {
            add_tag =>  ["syslog_rfc3164"]
          }
        }
      }
      else {
        mutate {  
          add_tag => ["syslog_priority_invalid"]
        }
      }
    } else {
      # only RFC 3164 allows a message to specify no priority
      mutate {  
        add_tag => [ "syslog_rfc3164", "syslog_priority_missing" ]
      }
    }

    # RFC 3164 suggests adding priority if it's missing. 
    # Even if missing, syslog_pri filter adds the default priority.
    syslog_pri {
      syslog_pri_field_name => "priority"
    }

    # parse both RFC 3164 and 5424
    grok {
      patterns_dir => "/etc/logstash/pattern.d"
      match => [ "message", "%{SYSLOG}" ]
      tag_on_failure => [ "_grokparsefailure_syslog" ]
    }
    
    # Check if a timestamp source was found and work out elapsed time recieving log
    # Note, mutate filter will convert a date object to a string not in ISO8601 format, so rather use ruby filter
    ruby {
      code => "event.set('timestamp_logstash', event.get('@timestamp'))"
#      old style ruby code (<v5.0)
#      code => "event['timestamp_logstash'] = event['@timestamp']"
    }

    if [timestamp_source] {
      date {
        locale => en
        # assume timezone for cases where it isn't provided
        timezone => "Europe/Berlin"
        match => [ "timestamp_source", "MMM d H:m:s", "MMM d H:m:s", "ISO8601" ]
      }
      # add a field for delta (in seconds) between logsource and logstash
      ruby {
        code => "event.set('time_elapsed_logstash', (event.get('timestamp_logstash') - event.get('@timestamp')))"
#        old style ruby code (<v5.0)
#        code => "event['time_elapsed_logstash'] = event['timestamp_logstash'] - event['@timestamp']"
      }
    }
    else {
      mutate {
        add_tag => ["syslog_timestamp_source_missing"]
      }
    }
    
    # Check if a host source was found
    if ! [host_source] {
      mutate {
        add_tag => ["syslog_host_source_missing"]
      }
    }

    # discard redundant info
    mutate {
      remove_field => [ "priority" ] #redundant and less useful once severity and facility are decoded
      replace => { "message" => "%{message_content}" } 
      remove_field => [ "message_syslog", "message_content" ] #already in content message
    } 
    
    # normalize for logstash fields
    mutate {
      rename => { "host" => "syslog_hostname" }
      rename => { "host_source" => "received_from" }
      rename => { "program" => "syslog_program" }
    }
  }
}

output {
 if [type] == "syslog" {
   elasticsearch {
     hosts => ["localhost:9200"]
     index => "syslog-%{+YYYY.MM.dd}"
   }
 }
# use for debug - this log also to /var/log/messages 
# stdout { codec => rubydebug }
# 
# enable metadata output
# stdout { codec => rubydebug { metadata => true } }
}