企业级日志收集系统——ELKstack
1.ELKstack简介
ELKstack 是 Elasticsearch、Logstash、Kibana 三个开源软件的组合而成，形成一款强大的实时日志收集展示系统。
各组件作用如下：
Logstash：日志收集工具，可以从本地磁盘，网络服务（自己监听端口，接受用户日志），消息队列中收集各种各样的日志，然后进行过滤分析，并将日志输入到Elasticsearch中。
Elasticsearch：日志分布式存储/搜索工具，原生支持集群功能，可以将指定时间的日志生成一个索引，加快日志查询和访问。
Kibana：可视化日志web展示工具，对Elasticsearch中存储的日志进行展示，还可以生成炫丽的仪表盘。
2.使用ELKstack对运维工作的好处：
1、应用程序的日志大部分都是输出在服务器的日志文件中，这些日志大多是开发人员来看，然后开发却没有登录服务器的权限，如果开发人员需要查看日志就需要运维到服务器来拿日志，然后交给开发；试想下，一个公司有10个开发，一个开发每天找运维那一次日志，对运维来说就是一个不小的工作量，这样大大影响了运维的工作效率，部署ELKstack之后，开发人员就可以直接登录kibana中进行日志的查看，就不需要通过运维查看日志，这样就减轻了运维的工作。
2、日志种类多，且分散在不同的位置难以查找：如LAMP/LNMP网站出现访故障，这个时候可能就需要通过查询日志来进行分析故障原因，如果需要查看apache的错误日志，就需要登录到Apache服务器查看，如果查看数据库错误日志就需要登录到数据库查询，实现下如果是一个集群环境几十台主机呢？这时如果部署了ELKstack就可以登录到kibana页面进行查看日志，查看不同类型的日志只需要点点鼠标切换一下索引即可。
3.ELK实验架构图
redis消息队列作用说明：
1、防止logstash和Es无法正常通信，从而丢失日志
2、防止日志量过大导致ES无法承受大量写操作从而丢失日志
3、应用程序（php，java）在输出日志时，可以直接输出到消息队列，从而完成日志收集
补充：如果redis使用的消息队列出现扩展瓶颈，可以使用更加强大的kafka，flume来代替。
实验环境说明：
[root@node1 ~]# cat /etc/redhat-release CentOS release 6.6 (Final) [root@node1 ~]# uname -rm 2.6.32-504.el6.x86_64 x86_64
使用软件说明：
logstash-1.5.4.tar.gz
elasticsearch-1.7.2.tar.gz
kibana-4.1.2-linux-x64.tar.gz
nginx和redis均为yum安装
部署顺序：
1、Elasticsearch集群配置
2、Logstash客户端配置（直接写入数据到ES集群，写入系统messages日志）
3、Redis消息队列配置（logstash写入数据到消息队列）
4、Kibana部署
5、nginx负载均衡kibana请求
6、案例：同时收集nginx和MySQL慢查询日志
7、kibana报表功能说明
配置需要注意的事项：
1、时间必须同步
2、出了问题，检查日志
4.Elasticsearch集群安装配置
（1）配置java环境，且版本为8，如果使用7可能会出现警告信息
[root@ES1 ~]# yum -y install java-1.8.0 [root@ES1 ~]# java -version openjdk version "1.8.0_51" OpenJDK Runtime Environment (build 1.8.0_51-b16) OpenJDK 64-Bit Server VM (build 25.51-b03, mixed mode)
（2）下载elasticsearch，可以使用我直接给出的地址也可以直接去官方下载最新版，官方地址在文章结尾有给出。
wget https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-1.7.2.tar.gz
（3）安装elasticsearch，安装过程很简单，只需要把软件解压到指定目录，然后做一个软链接即可
tar xf elasticsearch-1.7.2.tar.gz mv elasticsearch-1.7.2 /usr/local/ cd /usr/local/ ln -s /usr/local/elasticsearch-1.7.2/ /usr/local/elasticsearch
（4）修改配置文件，这里的一些路径就看自己的习惯了，推荐将数据目录放到ssd硬盘加快检索速度
[root@ES1 ~]# vim /usr/local/elasticsearch/config/elasticsearch.yml 32 cluster.name: elasticsearch #组播的名称地址 40 node.name: "linux-ES1" #节点名称，不能和其他节点重复 47 node.master: true #节点能否被选举为master 51 node.data: true #节点是否存储数据 107 index.number_of_shards: 5 #索引分片的个数 111 index.number_of_replicas: 1 #分片的副本个数 145 path.conf: /usr/local/elasticsearch/config/ #配置文件的路径 149 path.data: /data/es-data #数据目录路径 159 path.work: /data/es-worker #工作目录路径 163 path.logs: /usr/local/elasticsearch/logs/ #日志文件路径 167 path.plugins: /usr/local/elasticsearch/plugins #插件路径 184 bootstrap.mlockall: true #内存不向swap交换
（5）创建相关目录
mkdir /data/es-data -p mkdir /data/es-worker -p mkdir /usr/local/elasticsearch/logs mkdir /usr/local/elasticsearch/plugins
（6）安装启动脚本
由于是使用的tar包安装的ES，默认是没有启动脚本的，如果不愿意自己手动些启动脚本，可以根据如下步骤安装
git clone https://github.com/elastic/elasticsearch-servicewrapper.git cd elasticsearch-servicewrapper/ mv service/ /usr/local/elasticsearch/bin/ /usr/local/elasticsearch/bin/service/elasticsearch install
修改服务配置文件
[root@ES1 ~]# vim /usr/local/elasticsearch/bin/service/elasticsearch.conf set.default.ES_HOME=/usr/local/elasticsearch #设置ES的安装路径，必须和安装路径保持一直 set.default.ES_HEAP_SIZE=1024 #设置分配jvm内存大小
（7）启动ES，并检查是否监听9200和9300端口是否正常监听
[root@ES1 ~]# /etc/init.d/elasticsearch start [root@ES1 ~]# netstat -lntp | grep -E "9200|9300" tcp 0 0 :::9300 :::* LISTEN 37416/java tcp 0 0 :::9200 :::* LISTEN 37416/java
访问9200端口，检查节点是否正常
5.补充：ES2节点配置
配置时只需要保证elasticsearch.yml文件中，node.name和node1节点不同即可，其他步骤都是一样的
[root@ES2 ~]# vim /usr/local/elasticsearch/config/elasticsearch.yml 40 node.name: "linux-ES2" #节点名称，不能和其他节点重复
这里我已经配置好了ES2节点
6.配置集群管理插件（head）
官方提供了一个ES集群管理插件，可以非常直观的查看ES的集群状态和索引数据信息，安装方法如下所示：
[root@ES1 ~]# /usr/local/elasticsearch/bin/plugin -i mobz/elasticsearch-head
安装完成，访问安装head插件的ES服务器，后缀如下图所示，即可查看集群信息，由于还没有向ES中写入数据，所以ES可能还看不到有索引，写入数据后就可以在此页面 看到ES存储的日志信息了
这个时候ES集群就配置完成了，下面就可以配置logstash向ES集群中写入数据了。
7.Logstash部署
（1）下载logstash
wget https://download.elastic.co/logstash/logstash/logstash-1.5.4.tar.gz
（2）部署logstash需要java环境
yum -y install java-1.8.0 tar zxf logstash-1.5.4.tar.gz mv logstash-1.5.4 /usr/local/ ln -s /usr/local/logstash-1.5.4/ /usr/local/logstash
（3）设置启动脚本，在/etc/init.d/logstash中添加如下内容即可， 脚本中给出的logstash配置文件为/etc/logstash如果有需要请手动修改。
#!/bin/sh
# Init script for logstash
# Maintained by Elasticsearch
# Generated by pleaserun.
# Implemented based on LSB Core 3.1:
# * Sections: 20.2, 20.3
#
### BEGIN INIT INFO
# Provides: logstash
# Required-Start: $remote_fs $syslog
# Required-Stop: $remote_fs $syslog
# Default-Start: 2 3 4 5
# Default-Stop: 0 1 6
# Short-Description:
# Description: Starts Logstash as a daemon.
### END INIT INFO
PATH=/sbin:/usr/sbin:/bin:/usr/bin
export PATH
if [ `id -u` -ne 0 ]; then
echo "You need root privileges to run this script"
exit 1
fi
name=logstash
pidfile="/var/run/$name.pid"
export CLASSPATH=.:$JAVA_HOME/jre/lib/rt.jar:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export PATH=$PATH:$JAVA_HOME/bin
LS_USER=logstash
LS_GROUP=logstash
LS_HOME=/usr/local/logstash
LS_HEAP_SIZE="500m"
LS_JAVA_OPTS="-Djava.io.tmpdir=${LS_HOME}"
LS_LOG_DIR=/usr/local/logstash
LS_LOG_FILE="${LS_LOG_DIR}/$name.log"
LS_CONF_FILE=/etc/logstash.conf
LS_OPEN_FILES=16384
LS_NICE=19
LS_OPTS=""
[ -r /etc/default/$name ] && . /etc/default/$name
[ -r /etc/sysconfig/$name ] && . /etc/sysconfig/$name
program=/usr/local/logstash/bin/logstash
args="agent -f ${LS_CONF_FILE} -l ${LS_LOG_FILE} ${LS_OPTS}"
start() {
JAVA_OPTS=${LS_JAVA_OPTS}
HOME=${LS_HOME}
export PATH HOME JAVA_OPTS LS_HEAP_SIZE LS_JAVA_OPTS LS_USE_GC_LOGGING
# set ulimit as (root, presumably) first, before we drop privileges
ulimit -n ${LS_OPEN_FILES}
# Run the program!
nice -n ${LS_NICE} sh -c "
cd $LS_HOME
ulimit -n ${LS_OPEN_FILES}
exec "$program" $args
" > "${LS_LOG_DIR}/$name.stdout" 2> "${LS_LOG_DIR}/$name.err" &
# Generate the pidfile from here. If we instead made the forked process
# generate it there will be a race condition between the pidfile writing
# and a process possibly asking for status.
echo $! > $pidfile
echo "$name started."
return 0
}
stop() {
# Try a few times to kill TERM the program
if status ; then
pid=`cat "$pidfile"`
echo "Killing $name (pid $pid) with SIGTERM"
kill -TERM $pid
# Wait for it to exit.
for i in 1 2 3 4 5 ; do
echo "Waiting $name (pid $pid) to die..."
status || break
sleep 1
done
if status ; then
echo "$name stop failed; still running."
else
echo "$name stopped."
fi
fi
}
status() {
if [ -f "$pidfile" ] ; then
pid=`cat "$pidfile"`
if kill -0 $pid > /dev/null 2> /dev/null ; then
# process by this pid is running.
# It may not be our pid, but that's what you get with just pidfiles.
# TODO(sissel): Check if this process seems to be the same as the one we
# expect. It'd be nice to use flock here, but flock uses fork, not exec,
# so it makes it quite awkward to use in this case.
return 0
else
return 2 # program is dead but pid file exists
fi
else
return 3 # program is not running
fi
}
force_stop() {
if status ; then
stop
status && kill -KILL `cat "$pidfile"`
fi
}
case "$1" in
start)
status
code=$?
if [ $code -eq 0 ]; then
echo "$name is already running"
else
start
code=$?
fi
exit $code
;;
stop) stop ;;
force-stop) force_stop ;;
status)
status
code=$?
if [ $code -eq 0 ] ; then
echo "$name is running"
else
echo "$name is not running"
fi
exit $code
;;
restart)
stop && start
;;
reload)
stop && start
;;
*)
echo "Usage: $SCRIPTNAME {start|stop|force-stop|status|restart}" >&2
exit 3
;;
esac
exit $?
添加为系统服务
[root@client elk-config]# chkconfig --add logstash [root@client elk-config]# chkconfig logstash on [root@client elk-config]# chkconfig --list logstash logstash 0:off 1:off 2:on 3:on 4:on 5:on 6:off
（4）Logstash向Elasticsearch写入数据
1、设置一个logstash配置文件
[root@client elk]# cat /etc/logstash.conf
input { #表示从标准输入中收集日志
stdin {}
}
output {
elasticsearch { #表示将日志输出到ES中
host => ["172.16.4.102:9200","172.16.4.103:9200"] #可以指定多台主机，也可以指定集群中的单台主机
protocol => "http"
}
}
2、手动启动logstash并写入数据
[root@client elk]# /usr/local/logstash/bin/logstash -f /etc/logstash.conf Logstash startup completed hello world #这里是自己手动写入的内容
3、写入完成之后，查看ES中已经写入数据，并且自动建立了一个索引
在基本查询选项中选择指定的索引就可以看到写入的日志内容
这个时候说明，logstash结合elasticsearch是可以正常工作的，下面就以一个实际例子还说明如何，收集系统日志。
8.Logstash收集系统日志
修改logstash配置文件为如下所示内容，并启动logstash服务就可以在head中看到messages的日志已经写入到了ES中，并且创建了索引
[root@node5 ~]# vim /etc/logstash.conf
input {
file { #表示从文件中读取日志
path => "/var/log/messages" #文件路径
start_position => "beginning" #表示从文件开始处（第一行）读取日志进行收集
}
}
output {
elasticsearch { #表示输出到ES中
host => ["172.16.4.102:9200","172.16.4.103:9200"]
protocol => "http"
index => "system-messages-%{+YYYY-MM}" #表示创建的索引名称，%{+YYYY-MM}表示安装日期创建索引，这是指定日期格式
}
}
收集成功如下图所示，自动生成了system-messages的索引
logstash结合elasticsearch 写入日志没有问题的话，下一步就是logstash向redis中写入日志，当然如果日志的写入量不是很高，也可以不需要使用消息队列，直接使用logstash向elasticsearch 写入日志。
9.Redis消息队列配置
（1）安装redis，没有特殊需求使用yum安装即可
[root@redis ~]# yum -y install redis
（2）配置redis：redis默认是监听127.0.0.1，是不允许其他主机访问的，如果允许其他主机访问就需要配置监听外部ip地址或者0.0.0.0。
[root@redis ~]# vim /etc/redis.conf bind 0.0.0.0
（3）启动redis
[root@redis ~]# /etc/init.d/redis start [root@node4 ~]# netstat -lntp | grep redis tcp 0 0 0.0.0.0:6379 0.0.0.0:* LISTEN 36660/redis-server
（4）收集系统日志写入到redis中
[root@client ~]# cat /etc/logstash.conf
input {
file {
path => "/var/log/messages"
start_position => "beginning"
}
}
output {
redis { #表示写入到redis中
host => "172.16.4.104" #指明redis服务器地址
data_type => "list" #指定数据类型为list
key => "system-messages" #指定存入的键
}
}
设置完成，重启logstash服务，登录redis查询是否写入
redis 127.0.0.1:6379> KEYS * #查看redis中的键，已经创建了一个system-messages的键
1) "system-messages"
redis 127.0.0.1:6379> TYPE system-messages #查询键的类型为list
list
redis 127.0.0.1:6379> LINDEX system-messages -1 #查询键的值
"{"message":"Oct 18 13:55:56 config-03 yum[38732]: Installed: zsh-4.3.11-4.el6.centos.x86_64","@version":"1","@timestamp":"2015-10-18T06:57:54.581Z","host":"0.0.0.0","path":"/var/log/messages","type":"system"}"
注意：由于这里还没有使用logstash从redis中取出数据，所以这里是可以看到消息队列中的数据，如果是生产环境，可能使用KEYS *看不到消息队列中有数据，说明logstash已经从消息队列中取走了数据。
（5）从redis读取日志写入到ES
说明：此logstash我部署在redis服务器上，当然也可以部署在ES上，或者其他机器上看个人需求了
[root@node4 ~]# cat /etc/logstash.conf
input {
redis { #表示从redis中取出数据
host => "172.16.4.104"
data_type => "list"
key => "system-messages" #指定去system-messages这个键中取出数据
}
}
output {
elasticsearch {
host => ["172.16.4.102:9200","172.16.4.103:9200"]
protocol => "http"
index => "system-messages-%{+YYYY-MM}"
}
}
此处配置完成之后，就已经完成了日志的收集以及存储的配置，那么就剩下WEB展示的配置了，只需要部署完成kibana就可以实现最后的日志展示功能。
10.Kibana部署
说明：我这里是在两个ES节点部署kibana并且使用nginx实现负载均衡，如果没有特殊的需求可以只部署单台节点
（1）安装kibana，每个ES节点部署一个
tar zxf kibana-4.1.2-linux-x64.tar.gz mv kibana-4.1.2-linux-x64 /usr/local/ cd /usr/local/ ln -s kibana-4.1.2-linux-x64/ kibana
（2）配置kibana，只需要指定ES地址其他配置保持默认即可。
[root@node2 ~]# vim /usr/local/kibana/config/kibana.yml elasticsearch_url: http://172.16.4.102:9200 #指定elasticsearch地址，各自ES节点的kibana指向各自节点ES
（3）添加kibana启动脚本
#!/bin/bash
#chkconfig:35 13 91
#description:this is kibana servers
. /etc/rc.d/init.d/functions
start(){
/usr/local/kibana/bin/kibana >/dev/null 2>&1 &
if [ `ps -ef | grep -v grep | grep "kibana" | wc -l` -gt 0 ];then
action "starting kibana:" /bin/true
sleep 1
else
action "starting kibana:" /bin/false
sleep 1
fi
}
stop(){
kibanapid=`ps aux | grep "kibana" | grep -v grep | awk '{print $2}'`
kill -9 $kibanapid
if [ `ps -ef | grep -v grep | grep "kibana" | wc -l` -lt 1 ];then
action "stopping kibana:" /bin/true
sleep 1
else
action "stopping kibana:" /bin/false
sleep 1
fi
}
case "$1" in
start)
start
;;
stop)
stop
;;
restart)
$0 stop;
$0 start;
;;
*)
echo $"Usage: $0 {start|stop|restart}"
;;
esac
添加到系统服务：
[root@ES1 local]# chmod +x /etc/init.d/kibana [root@ES1 local]# chkconfig --add kibana [root@ES1 local]# chkconfig kibana on [root@ES1 local]# chkconfig --list kibana kibana 0:off 1:off 2:on 3:on 4:on 5:on 6:off
（4）启动kibana并登录web页面创建系统日志索引，并查询日志
1、启动kibana，并检查是否监听5601端口
[root@ES1 local]# /etc/init.d/kibana start starting kibana: [ OK ] [root@ES1 local]# netstat -lntp | grep 5601 tcp 0 0 0.0.0.0:5601 0.0.0.0:* LISTEN 40660/node
2、登录web页面，直接访问kibana的5601端口即可
配置完成之后就可以在Discover中看到写入的日志
查询指定字段的日志，如选择侧边栏的messages点击add就可以只查询messages字段的日志了
还可以根据时间查询日志，如下中文显示是我使用网页翻译显示的
也可以根据搜索框直接查询想要的内容，下图所示为查看日志中包含yum的内容。
11.Nginx反向代理kibana使请求负载均衡
（1）安装nginx
[root@nginx ~]# yum -y install nginx
（2）配置nginx负载均衡，配置文件如下所示：
http {
upstream kibana { #定义后端主机组
server 172.16.4.102:5601 weight=1 max_fails=2 fail_timeout=2;
server 172.16.4.103:5601 weight=1 max_fails=2 fail_timeout=2;
} }
server {
listen 80;
server_name localhost;
location / { #定义反向代理，将访问自己的请求，都转发到kibana服务器
proxy_pass http://kibana/;
index index.html index.htm;
}
}
}
设置完成，访问nginx监听的ip地址即可负载均衡kibana。
由于kiaba不支持用户认证，如果需要用户认证，也可以在nginx处添加认证。
此处ELKstack + redis日志收集系统就部署完成了，需要收集日志只需要设置根据需求编写logstash配置文件即可，下面是一个配置示例
12.案例：Logstash同时收集nginx日志和mysql慢查询日志
说明：这里的日志是我复制生产日志，不是本机生成的日志。
（1）为了方便nginx日志的统计搜索，这里设置nginx访问日志格式为json
说明：如果想实现日志的报表展示，最好将业务日志直接以json格式输出，这样可以极大减轻cpu负载，也省得运维需要写负载的filter过滤正则。
log_format json '{"@timestamp":"$time_iso8601",'
'"@version":"1",'
'"client":"$remote_addr",'
'"url":"$uri",'
'"status":"$status",'
'"domain":"$host",'
'"host":"$server_addr",'
'"size":$body_bytes_sent,'
'"responsetime":$request_time,'
'"referer": "$http_referer",'
'"ua": "$http_user_agent"'
'}';
access_log /var/log/access_json.log json;
（2）收集nginx和mysql慢查询日志，保存到redis消息队列中
input {
file {
path => "/var/log/access_json.log"
type => "nginx-access" #指定日志类型，以便在一个配置文件中收集多个日志，用来区别输出
start_position => "beginning"
codec => "json"
}
file {
path => "/root/slow-mysql.log"
type => "slow-mysql" #指定日志类型，以便在一个配置文件中收集多个日志，用来区别输出
start_position => "beginning"
codec => multiline {
pattern => "^# User@Host"
negate => true
what => "previous"
}
}
}
output {
if [type] == "nginx-access" { #对input中的输入进行判断，如果日志类型为nginx-access则执行以下输出，否则不执行
redis {
host => "172.16.4.104"
data_type => "list"
key => "nginx-access"
}
}
if [type] == "slow-mysql" {
redis {
host => "172.16.4.104"
data_type => "list"
key => "slow-mysql"
}
}
}
Logstash 从redis消息队列读取日志存储到ES中，此处的logstash部署在redis消息队列上，当然也可以部署在ES上
input {
redis {
type => "nginx-access"
host => "172.16.4.104"
data_type => "list"
key => "nginx-access"
}
redis {
type => "slow-mysql"
host => "172.16.4.104"
data_type => "list"
key => "slow-mysql"
}
}
output {
if [type] == "nginx-access" {
elasticsearch {
host => ["172.16.4.102:9200","172.16.4.103:9200"]
protocol => "http"
index => "nginx-access-%{+YYYY.MM}"
}
}
if [type] == "slow-mysql" {
elasticsearch {
host => ["172.16.4.102:9200","172.16.4.103:9200"]
protocol => "http"
index => "slow-mysql-%{+YYYY.MM}"
}
}
}
（4）Kibana创建对应日志的索引，用来查询日志
1、创建nginx索引
2、这个时候就可以看到nginx日志了
（2）创建mysql慢查询索引
13.图表展示
限于文章篇幅，这里不想做过多的描述图标生产的过程，如果各位有需要可以参考给出的文档，如下是效果图：
对于kibana图表展示，只想说如下两点：
1、日志最好是json格式，否则还需要使用filter 的grok使用复杂的正则表达式进行日志内容处理，即加大的运维难度，由加大的服务器运行负担。
2、ELKStask最好用来收集业务日志，对于WEB访问日志的分析，可以使用第三方的分析工具，如百度统计，CNZZ，以及开源工具如piwik，awstats。
14.参考文档
官方文档：https://www.elastic.co/
中文指南：http://kibana.logstash.es/
篇幅问题，很多内容可能描述不是特别清楚，如果有问题，可以直接联系本人qq或者留言。
