Logstash
Logstash is the open source version of splunk, using ElasticSearch as its search engine.
Installation
For detailed information, consult logstash's tutorial (http://logstash.net/docs/). Prior to logstash 1.4.0, the logstash package comes as a monolithic .jar file. To get started, install java and run the jar file.
If you are using logstash >= 1.4.0, it's probably easier to just install logstash and ElasticSearch from their repository:
[logstash-1.4]
name=logstash repository for 1.4.x packages
baseurl=http://packages.elasticsearch.org/logstash/1.4/centos
gpgcheck=1
gpgkey=http://packages.elasticsearch.org/GPG-KEY-elasticsearch
enabled=1
[elasticsearch-1.3]
name=Elasticsearch repository for 1.3.x packages
baseurl=http://packages.elasticsearch.org/elasticsearch/1.3/centos
gpgcheck=1
gpgkey=http://packages.elasticsearch.org/GPG-KEY-elasticsearch
enabled=1
Then run yum install logstash elasticsearch
Configuration
The configuration files for logstash are located at:
/etc/sysconfig/logstash/etc/logstash/conf.d/
The files in the conf.d directory will be treated as a single configuration file in sequence by their filenames by logstash.
You may want to change the DATA_DIR path in the logstash configuration. The actual input/parse/output configurations will be placed in the conf.d directory. More on this below.
ElasticSearch's files are at:
/etc/elasticsearch/elasticsearch.yml
You will need to configure ElasticSearch based on how you want to set up your search. For replication/sharding, you should ideally have more than one server. If you do have more than one server, make sure you have the node names set and have the autodiscoverer configured.
Configuration
The configuration you provide logstash defines how logstash deals with incoming messages. There are three main parts to the configuration:
- Input
- Filter
- Output
Input defines what ports logstash listens on and how to tag the incoming messages. Filter defines what logstash needs to do on the incoming messages, based on the tags defined from the input step. Output defines how these messages are stored.
For example, my current configuration is:
input {
# Import syslog messages
tcp {
type => syslog_import
port => 4401
}
# Accept syslog messages from hosts
syslog {
type => syslog
port => 5544
}
}
filter {
if [type] == "syslog" {
# Does the syslog parsing.
syslog_pri { }
mutate {
replace => [ "@source", "%{logsource}" ]
replace => [ "@message", "%{message}" ]
replace => [ "@program", "%{program}" ]
replace => [ "@type", "syslog" ]
}
# Date is parsed and placed into @timstamp.
date {
match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss", "ISO8601" ]
}
# Clean up the extra syslog_ fields generated above from grok.
mutate {
remove_field => [ "syslog_hostname", "syslog_message", "syslog_program", "syslog_timestamp", "type", "message", "logsource", "program"]
}
}
# For imported syslog messages...
if [type] == "syslog_import" {
if [message] =~ /last message repeated.*/ {
drop {
}
}
if [message] == "" {
drop {
}
}
# Parse with grok
grok {
# Use the custom SYSLOGYEARTIMESTAMP pattern from the patterns
# directory. We need this to define year.
patterns_dir => "./patterns"
# The pattern to match.
# This is the standard syslog pattern.
match => { "message" => "%{SYSLOGYEARTIMESTAMP:syslog_timestamp} (%{USER:syslog_user}\@)?%{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
# Add a few intermediate fields
add_field => [ "received_at", "%{@timestamp}" ]
add_field => [ "received_from", "%{host}" ]
}
# When the above grok parsing fails, a '_grokparsefailure' tag gets
# added to the message. In that case, we attempt to update some fields.
# Why? Beats me.
if !("_grokparsefailure" in [tags]) {
mutate {
replace => [ "@source", "%{syslog_hostname}" ]
replace => [ "@message", "%{syslog_message}" ]
replace => [ "@program", "%{syslog_program}" ]
replace => [ "@type", "syslog imported" ]
}
}
# Parse the date. This puts it into the @timestamp field on a successful
# parse.
date {
match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss", "YYYY MMM d HH:mm:ss", "YYYY MMM dd HH:mm:ss" ]
}
# Clean up the extra syslog_ fields generated above from grok.
mutate {
remove_field => [ "syslog_hostname", "syslog_message", "syslog_program", "syslog_timestamp", "type", "message", "host" ]
}
}
}
output {
# Debugging
# stdout {
# codec => json
# }
elasticsearch {
# Define our own... hosted on my computer
# bind_host => "leo-linux"
# bind_port => 9200
host => "127.0.0.1"
port => 9300
cluster => "logstash_es"
node_name => "logstash_0"
# Index defaults to 'logstash-%{+YYYY.MM.dd}'
# The templates being used can be defined using:
template => "/etc/logstash/template/logstash.json"
}
}
I do some processing on the messages coming in from port 4401, tagged as syslog_import. In the filter step, any messages tagged as syslog_import will be processed by grok, the parser. Let's take a closer look at the grok configuration:
grok {
# Use the custom SYSLOGYEARTIMESTAMP pattern from the patterns
# directory. We need this to define year.
patterns_dir => "./patterns"
# The pattern to match.
# This is the standard syslog pattern.
match => { "message" => "%{SYSLOGYEARTIMESTAMP:syslog_timestamp} (%{USER:syslog_user}\@)?%{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
# Add a few intermediate fields
add_field => [ "received_at", "%{@timestamp}" ]
add_field => [ "received_from", "%{host}" ]
}
This grok instance attempts to match the incoming message to the defined pattern. The syntax defining matched strings is %{PATTERN_NAME:variable_name} where PATTERN_NAME is a grok-pattern defined in /patterns/* in the .jar file and also in the directory defined in the patterns_dir directory, and variable_name is the name that can be used to referenced the matched value later on in the grok instance.
The SYSLOGYEARTIMESTAMP pattern is a custom pattern defined in my ./patterns directory.
cat patterns/extra
SYSLOGYEARTIMESTAMP %{YEAR} %{MONTH} +%{MONTHDAY} %{TIME}
In the case above, syslog messages being imported whose date field matches the format given in SYSLOGYEARTIMESTAMP will be placed in the variable syslog_timestamp.
</syntaxhighlight >