ELK

环境:
Vagrant 1.8.1
CentOS 7.2 192.168.0.228
Elasticsearch 2.3.2
logstash 2.2.4
Kibana 4.4.2
filebeat 1.2.2
topbeat 1.2.2

搭建ELK日志分析平台。此处为其核心配置文件。具体搭建过程请参考ELK环境搭建.docx文档

Screenshots

elasticsearch索引列表

Nginx日志分析

Syslog系统日志分析

Tomcat日志分析

系统日志分析

Topbeat Dashboard

logstash命令

logstash命令帮助

$ /opt/logstash/bin/logstash -h
Usage:
    /bin/logstash agent [OPTIONS]

Options:
    -f, --config CONFIG_PATH      Load the logstash config from a specific file
                                  or directory.  If a directory is given, all
                                  files in that directory will be concatenated
                                  in lexicographical order and then parsed as a
                                  single config file. You can also specify
                                  wildcards (globs) and any matched files will
                                  be loaded in the order described above.
    -e CONFIG_STRING              Use the given string as the configuration
                                  data. Same syntax as the config file. If no
                                  input is specified, then the following is
                                  used as the default input:
                                  "input { stdin { type => stdin } }"
                                  and if no output is specified, then the
                                  following is used as the default output:
                                  "output { stdout { codec => rubydebug } }"
                                  If you wish to use both defaults, please use
                                  the empty string for the '-e' flag.
                                   (default: "")
    -w, --pipeline-workers COUNT  Sets the number of pipeline workers to run.
                                   (default: 1)
    -b, --pipeline-batch-size SIZE Size of batches the pipeline is to work in.
                                   (default: 125)
    -u, --pipeline-batch-delay DELAY_IN_MS When creating pipeline batches, how long to wait while polling
                                  for the next event.
                                   (default: 5)
    --filterworkers COUNT         DEPRECATED. Now an alias for --pipeline-workers and -w
    -l, --log FILE                Write logstash internal logs to the given
                                  file. Without this flag, logstash will emit
                                  logs to standard output.
    -v                            Increase verbosity of logstash internal logs.
                                  Specifying once will show 'informational'
                                  logs. Specifying twice will show 'debug'
                                  logs. This flag is deprecated. You should use
                                  --verbose or --debug instead.
    --quiet                       Quieter logstash logging. This causes only 
                                  errors to be emitted.
    --verbose                     More verbose logging. This causes 'info' 
                                  level logs to be emitted.
    --debug                       Most verbose logging. This causes 'debug'
                                  level logs to be emitted.
    --debug-config                translation missing: en.logstash.runner.flag.debug_config (default: false)
    -V, --version                 Emit the version of logstash and its friends,
                                  then exit.
    -p, --pluginpath PATH         A path of where to find plugins. This flag
                                  can be given multiple times to include
                                  multiple paths. Plugins are expected to be
                                  in a specific directory hierarchy:
                                  'PATH/logstash/TYPE/NAME.rb' where TYPE is
                                  'inputs' 'filters', 'outputs' or 'codecs'
                                  and NAME is the name of the plugin.
    -t, --configtest              Check configuration for valid syntax and then exit.
    --[no-]allow-unsafe-shutdown  Force logstash to exit during shutdown even
                                  if there are still inflight events in memory.
                                  By default, logstash will refuse to quit until all
                                  received events have been pushed to the outputs.
                                   (default: false)
    -h, --help                    print help

检查指定logstash配置文件

$ /opt/logstash/bin/logstash agent -f /etc/logstash/conf.d/14-log4j_to_es.conf -t

收集指定配置文件日志

$ /opt/logstash/bin/logstash agent -f /etc/logstash/conf.d/14-log4j_to_es.conf

查看logstash服务状态

$ sudo service logstash status 
logstash is running

$ sudo service logstash start|stop|restart

kibana

修改/opt/kibana/config

server.port: 5601
server.host: "192.168.0.228"
elasticsearch.url: "http://192.168.0.228:9200"
kibana.index: ".kibana"

ELK使用场景配置示例

syslog日志

logstash filter配置

filter {
  if [type] == "syslog" {
    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
      add_field => [ "received_at", "%{@timestamp}" ]
      add_field => [ "received_from", "%{host}" ]
    }
    syslog_pri { }
    date {
      match => [ "syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
    }
  }
}

java日志收集

http://kibana.logstash.es/content/logstash/examples/java.html

log4j SocketAppender

logstash配置

input {
  # log4j SocketAppender
  log4j {
    mode => "server"
    host => "192.168.0.228"
    port => 4560
    type => "log4j"
  }
}

filter {
}

output {
  if [type] == "log4j" {
    elasticsearch {
      action => "index"
      hosts  => "192.168.0.228:9200"
      index  => "log4j-access-%{+yyyy.MM.dd}"
    }
    redis {
      host => "192.168.0.46"
      port => 6379
      data_type => "list"
      key => "logstash:log4j"
    }
  } 
}

log4j.properties

log4j.rootLogger=logstash

###SocketAppender###
log4j.appender.logstash=org.apache.log4j.net.SocketAppender
# logstash中log4j input中的端口号
log4j.appender.logstash.Port=4560
# logstash所在机器IP
log4j.appender.logstash.RemoteHost=192.168.0.228
log4j.appender.logstash.ReconnectionDelay=60000
log4j.appender.logstash.LocationInfo=true
log4j.appender.logstash.Application=elk-log4j-simple

log4j-jsonevent-layout

logstash配置

input {
  # log4j-jsonevent-layout
  file {
    codec => json
    path => "/home/vagrant/tomcat-7.0.69/bin/target/*.log"
    type => "log4j"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}

filter {
}

output {
  if [type] == "log4j" {
    elasticsearch {
      action => "index"
      hosts  => "192.168.0.228:9200"
      index  => "log4j-access-%{+yyyy.MM.dd}"
    }
    redis {
      host => "192.168.0.46"
      port => 6379
      data_type => "list"
      key => "logstash:log4j"
    }
  } 
}

<!-- 将lo4j日志输出为json -->
<dependency>
    <groupId>net.logstash.log4j</groupId>
    <artifactId>jsonevent-layout</artifactId>
    <version>1.7</version>
</dependency>

注:output中elasticsearch项中index为当前被索引文档在elasticsearch中索引名称。使用kibana搜索的时候需要事先根据该值创建一个index pattern

tomcat日志

logstash pattern配置

JAVACLASS (?:[a-zA-Z0-9-]+\.)+[A-Za-z0-9$]+

JAVALOGMESSAGE (.*)

# MMM dd, yyyy HH:mm:ss eg: Jan 9, 2014 7:13:13 AM
CATALINA_DATESTAMP %{MONTH} %{MONTHDAY}, 20%{YEAR} %{HOUR}:?%{MINUTE}(?::?%{SECOND}) (?:AM|PM)

# yyyy-MM-dd HH:mm:ss,SSS ZZZ eg: 2014-01-09 17:32:25,527 -0800
TOMCAT_DATESTAMP 20%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{HOUR}:?%{MINUTE}(?::?%{SECOND}) %{ISO8601_TIMEZONE}

CATALINALOG %{CATALINA_DATESTAMP:timestamp} %{JAVACLASS:class} %{JAVALOGMESSAGE:logmessage}

# 2014-01-09 20:03:28,269 -0800 | ERROR | com.example.service.ExampleService - something compeletely unexpected happened...
TOMCATLOG %{TOMCAT_DATESTAMP:timestamp} \| %{LOGLEVEL:level} \| %{JAVACLASS:class} - %{JAVALOGMESSAGE:logmessage}

logstash filter配置

filter {
  if [type] == "tomcat_access" {
    grok {
      match => [ "message", "%{TOMCATLOG}", "message", "%{CATALINALOG}" ]
    }
    date {
      match => [ "timestamp", "yyyy-MM-dd HH:mm:ss,SSS Z", "MMM dd, yyyy HH:mm:ss a" ]
    }
  }
}

apache日志

logstash filter配置

filter {
  if [type] == "apache-access" {
    grok {
      match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
  }
}

nginx访问日志

logstash pattern配置

NGUSERNAME [a-zA-Z\.\@\-\+_%]+
NGUSER %{NGUSERNAME}
NGINXACCESS %{IPORHOST:clientip} %{NGUSER:ident} %{NGUSER:auth} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response} (?:%{NUMBER:bytes}|-) (?:"(?:%{URI:referrer}|-)"|%{QS:referrer}) %{QS:agent}

logstash filter配置

filter {
  if [type] == "nginx-access" {
    grok {
      match => { "message" => "%{NGINXACCESS}" }
    }
    geoip {
      source => "clientip"
      target => "geoip"
      database => "/etc/logstash/GeoLiteCity.dat"
      add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
      add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}"  ]
    }
    mutate {
      convert => [ "[geoip][coordinates]", "float"]
    }
  }
}

nginx access日志转化为json格式

http://kibana.logstash.es/content/logstash/examples/nginx-access.html

http://kibana.logstash.es/content/logstash/plugins/codec/json.html

nginx.conf

log_format json '{"@timestamp":"$time_iso8601",'
                 '"host":"$server_addr",'
                 '"clientip":"$remote_addr",'
                 '"size":$body_bytes_sent,'
                 '"responsetime":$request_time,'
                 '"upstreamtime":"$upstream_response_time",'
                 '"upstreamhost":"$upstream_addr",'
                 '"http_host":"$host",'
                 '"url":"$uri",'
                 '"xff":"$http_x_forwarded_for",'
                 '"referer":"$http_referer",'
                 '"agent":"$http_user_agent",'
                 '"status":"$status"}';
access_log  /var/log/nginx/access.log  json;

logstash配置

input {
  file {             #从nginx日志读入
    type => "nginx-access"
    path => "/var/log/nginx/access.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
    codec => "json"  #这里指定codec格式为json
  }
}

filter {
    mutate {
        split => [ "upstreamtime", "," ]
    }
    mutate {
        convert => [ "upstreamtime", "float" ]
    }
}

output {
  if [type] == "nginx-access" {
    elasticsearch {
      hosts => ["192.168.0.228:9200"]
      index => "nginx-access-%{+yyyy.MM.dd}"
    }
  }
}

mysql慢日志

http://kibana.logstash.es/content/logstash/examples/mysql-slow.html

logstash配置

input {
  file {
   type => "mysql-slow"
   path => "/var/log/mysql/mysql-slow.log"
   start_position => "beginning"
   sincedb_path => "/dev/null"
   codec => multiline {         #这里用到了logstash的插件功能，将本来属于一行的多行日志条目整合在一起，让他属于一条   
     pattern => "^# User@Host"  #用到了正则去匹配
     negate => true
     what => "previous"
   }
  }
}

filter {
  # drop sleep events
  grok {
    match => { "message" => "SELECT SLEEP" }
    add_tag => [ "sleep_drop" ]
    tag_on_failure => [] # prevent default _grokparsefailure tag on real records
  }
  if "sleep_drop" in [tags] {
    drop {}
  }
  grok {
    match => [ "message", "(?m)^# User@Host: %{USER:user}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\s*# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*(?:use %{DATA:database};\s*)?SET timestamp=%{NUMBER:timestamp};\s*(?<query>(?<action>\w+)\s+.*)\n# Time:.*$" ]
  }
  date {
    match => [ "timestamp", "UNIX" ]
    remove_field => [ "timestamp" ]
  }
}

output {
  if [type] == "mysql-slow" {
     elasticsearch {
        action => "index"
        hosts  => "192.168.0.228:9200"
        index  => "mysql-slow-%{+yyyy.MM.dd}"
     }
  }
}