logstash.conf

# A sample Logstash configuration to parse logs shipped with rsyslog.
# See also information about

input {
  file {
    path => "/var/log/rsyslog/*/*.log"
    exclude => "*.bz2"
    type => syslog
    sincedb_path => "/var/run/logstash/sincedb"
    sincedb_write_interval => 10
  }
}

output {
  elasticsearch_http {
    host => "localhost"
    flush_size => 5000
  }
}

filter {
  if [type] == "syslog" {
    mutate {
      add_field => [ "syslog_message", "%{message}" ]
      remove_field => "message"
    }
    grok {
      match => [ "syslog_message", "%{SYSLOGLINE}" ]
    }
    date {
      match => [ "timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }

    # Parse Drupal logs that are logged to syslog.
    if [program] == "drupal" {
      grok {
        match => [ "message", "https?://%{HOSTNAME:vhost}?\|%{NUMBER:d_timestamp}\|(?<d_type>[^\|]*)\|%{IP:d_ip}\|(?<d_request_uri>[^\|]*)\|(?<d_referer>[^\|]*)\|(?<d_uid>[^\|]*)\|(?<d_link>[^\|]*)\|(?<d_message>.*)" ]
      }
    }

    # Parse nginx logs that have been shipped over syslog.
    if [program] == "nginx_access" {
      mutate {
        remove_field => [ "path", "pid", "syslog_message", "timestamp" ]
      }
      grok {
        match => [ "message", "%{COMBINEDAPACHELOG} %{BASE16FLOAT:duration} %{IPORHOST:hostname} %{POSINT:port}" ]
      }
      # Put the hostname into the request, a la Varnish.
      if [port] == "443" {
        mutate {
          replace => [ "request", "https://%{hostname}%{request}" ]
        }
      }
      else {
        mutate {
          replace => [ "request", "http://%{hostname}%{request}" ]
        }
      }
      ruby {
        code => "event['duration'] = event['duration'].to_f * 1000.0"
      }
      date {
        match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
      }
      useragent {
        source => "agent"
        target => "browser"
      }
    }
  }
}