实时日志分析作为掌握业务情况、故障分析排查的一个重要手段，目前使用最多最成熟的莫过于ELK方案，整体方案也有各种架构组合，像(Rsyslog/filebeat/...)->Elastic->Kibana、(Rsyslog/filebeat/...)->Redis
最新记录:11台ES（12核64G机械盘）,每天日志量在37.4亿条，Primary Shard大小1.7T，最高索引速度每秒8万左右。
此文为原标题“玩儿透日志分析集群搭建.调优.管理rsyslog->kafka->spark->elk”的更改版。
沟通交流请到运维开源互助社区：https://project-help.cn/forum-6.htm
实时日志分析作为掌握业务情况、故障分析排查的一个重要手段，目前使用最多最成熟的莫过于ELK方案，整体方案也有各种架构组合，像(Rsyslog/filebeat/...)->Elastic->Kibana、(Rsyslog/filebeat/...)->Redis->Logstash->Elastic->Kibana、(Rsyslog/filebeat/...)->Kafka->Logstash->Elastic->Kibana等等，复杂点的有Spark的引用，再复杂点的引入Hadoop系列。每种方案适合不同的应用场景，没有优劣之分，我目前用的是(Rsyslog/filebeat)->kafka->(Logstash/Spark)->Elastic->Kibana和Rsyslog->Rsyslog中继->Kafka->(Logstash/spark)->Elastic->Kibana方案，中间使用Spark对日志进行再次聚合。
从整体架构进行抽象总结，其实就是采集->清洗汇聚->索引->展现四个环节，再去考虑各环节中间缓存、队列的使用，然后每个环节用不同的软件工具来实现。下面介绍一下我目前方案集群的搭建和配置，希望对同行有所帮助，在ELK探索过程中多谢远川和冯超同学的奉献交流，常用的一个架构图如下：
工作中对不同场景的各种架构进行了梳理，详见博文：运维数据分析平台建设的4个段位——架构演进
一、采集（rsyslog/filebeat）
1、关于rsyslog
客户端使用rsyslog8.19.0做的收集，直接centos安装rpm包，安装详细见：
http://www.rsyslog.com/rhelcentos-rpms
将yum源配置好后：
yum install rsyslog yum install rsyslog-kafka
安装好后对应rsyslog的配置文件如下：
module(load="imfile")
module(load="omkafka")
$PreserveFQDN on
main_queue(
queue.workerthreads="10" # threads to work on the queue
queue.dequeueBatchSize="1000" # max number of messages to process at once
queue.size="50000" # max queue size
)
##########################nginx log################################
$template nginxlog,"%$myhostname%`%msg%"
if $syslogfacility-text == 'local6' then {
action(
broker=["10.13.88.190:9092","10.13.88.191:9092","10.13.88.192:9092","10.13.88.193:9092"]
type="omkafka"
topic="cms-nginx"
template="nginxlog"
partitions.auto="on"
)
stop
}
############################redis log#########################
$template redislog,"%$myhostname%`%msg%"
ruleset(name="redis7215-log") {
action(
broker=["10.13.88.190:9092","10.13.88.191:9092","10.13.88.192:9092","10.13.88.193:9092"]
type="omkafka"
topic="redis-log"
template="redislog"
partitions.auto="on"
)
}
input(type="imfile"
File="/data1/ms/log/front/redis7215.log"
Tag=""
ruleset="redis7215-log"
freshStartTail="on" #start tailf
reopenOnTruncate="on" #Truncate reopen
)
input(type="imfile"
File="/data1/ms/log/front/redis7243.log"
Tag=""
ruleset="redis7215-log"
freshStartTail="on"
reopenOnTruncate="on"
)
############################php curl log#############################
$template phpcurl-log,"%$myhostname%`%msg%"
ruleset(name="phpcurl-log") {
action(
broker=["10.13.88.190:9092","10.13.88.191:9092","10.13.88.192:9092","10.13.88.193:9092"]
type="omkafka"
topic="phpcurl-log"
template="phpcurl-log"
partitions.auto="on"
)
}
input(type="imfile"
File="/data1/ms/log/php_common/php_slow_log"
Tag=""
ruleset="phpcurl-log"
freshStartTail="on"
reopenOnTruncate="on"
)为了避免在日志发送错误时，丢在message日志里，瞬间将磁盘占满，同时配置丢弃策略
*.info;mail.none;authpriv.none;cron.none;local6.none /var/log/messages
目前收集了nginx、redis、php curl三种日志，说一下收集方案。
1)、对于nginx日志
方案1：采用nginx的rsyslog模块将日志打到local6，对应nginx的配置如下
##########elk############################# access_log syslog:local6 STAT;
然后通过如上rsyslog的配置，将日志直接入kafka队列，kafka集群是4个broker。
方案2：线上还有另一个传输方案，rsyslog设置一个中继，通过udp的方式将日志传到中继的rsyslog，由中继rsyslog入kafka，这么做的目的是方便了管理，当时还有个考虑是udp不会堵，但经过多轮测试后，nginx的rsyslog模块也是很健壮，不会堵的。
2)、对于redis、php curl的日志
通过rsyslog的imfile模块，直接对文件监听，配置见上面的rsyslog配置，在日志轮转时通过超链接的方式进行新文件的连接，对应的超连接计划任务如下，每天0点5分执行：
5 0 * * * root sh /usr/local/script/php_slow_log.sh &> /dev/null
#!/bin/bash DATE=`date +%F` ln -sf /data1/ms/log/php_common/curl-$DATE /data1/ms/log/php_common/php_slow_log
备注：
a、rsyslog向kafka推送消息时，轮询发送消息到broker上的leader partition；
b、rsyslog通过udp或tcp向外转发日志时，会默认加上时间、主机名、主机ip的属性。
c、rsyslog的其余知识博文详见：
博文1：巧用rsyslog收集多套日志并做单套日志的过滤分离
2、关于filebeat
filebeat是elk体系原生的轻量级日志采集工具，用go编写，使用优雅，对比logstash没有那么重，所以现在又说elk为elkb，加了包括filebeat的好多beat。filebeat的配置使用也写了不少博客了，这里主要说一个坑吧。
1)、filebeat文件句柄不释放深度坑
这是生产环境遇到的一个问题，表现就是磁盘满了报警，业务质量下降，但登录服务器找不到大文件，最后怀疑是有删掉没有释放的大文件，lsof查询后找到元凶：
原来收集处理速度跟不上生产速度，导致日志在轮转删除时filebeat还没处理完，就保留了句柄，排查后果断修改了配置，更改了收集策略解决收集慢问题，同时filebeat添加熔断策略如下：
queue.mem: events: 256 flush.min_events: 128 flush.timeout: 5s close_inactive: 5m close_timeout: 30m clean_inactive: 72h ignore_older: 70h
二、队列（kafka+zookeeper）
队列用的是kafka，kafka集群使用zookeeper管理，我们用了4台服务器混装了4个kafka和3个zookeeper，kafka和zookeeper的安装地址如下：
http://kafka.apache.org/downloads 注意：下载Binary downloads版本，别下错了，解压后就能用
http://zookeeper.apache.org/ 注意：安装过程很简单，按照文档来即可，不在说明
kafka管理可以用kafka-manager详见博文：kafka与zookeeper管理之kafka-manager踩坑小记
kafka原理介绍推荐：Kafka设计与原理详解
1、关于kafaka
1)、配置比较简单，基本默认即可,常调整的配置项如下：
配置文件：server.properties broker.id=190 #id num.partitions=20 #默认kafka的partion数量 log.dirs=/data1/kafka-logs #日志文件存放目录 log.retention.hours=3 #日志保留时间长短 zookeeper.connect=10.13.88.190:2181,10.13.88.191:2181,10.13.88.192:2181 #zookeeper指定 delete.topic.enable=true #topic是可以删除的
2)、安装后测试(假设kafka和zookeeper都装了)：
开两个终端，两个终端分别运行如下命令
启动：./bin/kafka-server-start.sh /usr/local/kafka/config/server.properties & 关闭：./bin/kafka-server-stop.sh 终端1：./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test 终端2：./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic test
注意两个终端的topic要一个名字，这时你在终端1输入任何数据，在终端2是同步的，证明你安装成功。
3)、kafka常用管理命令
创建topic：./bin/kafka-topics.sh --create --topic test --replication-factor 1 --partitions 32 --zookeeper localhost:2181 删除topic：./bin/kafka-topics.sh --delete --topic test --zookeeper localhost:2181 查看topic列表：./bin/kafka-topics.sh --list --zookeeper localhost:2181 查看某个topic详细：./bin/kafka-topics.sh --describe --topic test --zookeeper localhosts:2181 监控某个topic的消费：./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test 指定消费组查看消费情况：./bin/kafka-consumer-offset-checker.sh --zookeeper localhost:2181 --group test
4)、kafka集群消费吞吐量性能测试如下，每秒可以支持消费50万条信息，我们目前的QPS远远是没问题的
备注：topic下partitions的数量决定了并发消费的数量，在设置上要根据消息的QPS和硬盘情况合理配置。
2、关于zookeeper
1)、配置比较简单，大多数默认项，最好奇数个，半数以上zookeeper存活可用
配置文件：zoo.cfg dataDir=/data1/zookeeper server.1=10.13.88.190:3888:4888 server.2=10.13.88.191:3889:4888 server.3=10.13.88.192:3889:4888
注意：要在数据目录手动建立myid，myid的值是server后面的数字，数字是有范围限制的1~255
2)、zookeeper的常用管理命令
zookeeper我主要是看下它的整体状态，写了个简单脚本获取zookeeper的状态,执行结果如下：
脚本内容如下：
#!/bin/sh
#writer:gaolixu
[ -z $1 ] && echo "Please specify zoo.cfg like /usr/local/zookeeper/conf/zoo.cfg " && exit
cat $1 |grep "^server" |awk -F'[:|=]' '{print $2}' |
while read line
do
echo -ne "$line\t"
echo stat|nc -w 2 $line 2181 |egrep "^(Node|Zxid|Mode|Connections)" |tr "\n" "\t"
echo stat|nc -w 2 $line 2181 |egrep "^(Node|Zxid|Mode|Connections)" &>/dev/null || echo -n "host is done."
echo
done
使用方式：zkstat.sh /配置文件zoo.cfg的位置zookeeper是相当稳定的，基本不用管。
备注：zookeeper配置文件里不能有汉字，否则启动不起来。
三、清洗汇聚（logstash/spark）
logstash用做清洗，并且将处理好的日志推送到es里，安装过程很简单详见网址：
https://www.elastic.co/guide/en/logstash/current/installing-logstash.html#package-repositories
线上的nginx的配置文件如下：
input {
kafka {
zk_connect => "10.13.88.190:2181,10.13.88.191:2181,10.13.88.192:2181"
topic_id => "cms-nginx"
group_id => "cms-nginx"
consumer_threads => 1
reset_beginning => false
decorate_events => false
}
}
filter {
ruby {
init => "@kname = ['host-name','front','http_x_up_calling_line_id','request','http_user_agent','status','remote_addr_1','id','http_referer','request_time','body_bytes_sent','http_deviceid','http_x_forwarded_for','domain','cookie']"
code => "event.append(Hash[@kname.zip(event['message'].split('`'))]) "
remove_field => ["@version","_score","id","tags","key","message","http_deviceid","http_x_up_calling_line_id","","cookie"]
}
if [front] {
grok {
match => ["front","%{HTTPDATE:logdate}"]
}
date {
match => ["logdate", "dd/MMM/yyyy:HH:mm:ss Z"]
target => "@timestamp"
remove_field => ["front","logdate"]
}
}
if [request] {
ruby {
init => "@kname = ['method','uri','verb']"
code => "event.append(Hash[@kname.zip(event['request'].split(' '))])"
remove_field => [ "request","method","verb"]
}
}
if [remote_addr_1] {
grok {
match => [ "remote_addr_1", "%{IPV4:remote_addr}" ]
remove_field => ["remote_addr_1"]
}
}
mutate {
convert => [
"body_bytes_sent" , "integer",
"status" , "integer",
"request_time" , "float" ]
}
}
output {
elasticsearch {
hosts => ["10.39.40.94:9200","10.39.40.95:9200","10.39.40.96:9200","10.39.40.97:9200"]
workers => 1
index => "logstash-cms-nginx-%{+YYYY.MM.dd.hh}"
}
#stdout { codec => dots
#workers => 5
#} #测试性能时使用
#stdout { codec => rubydebug } #调试时使用
}
启动命令：./bin/logstash -w 4 -b 1000 -f /etc/logstash/conf.d/kafka_logstash_cms_nginx.conf &
-w 后面的worker数是根据cpu的核心数大概算一下，我这里一台服务器开三个logstash，每个起4个worker配置文件看着很长，其实阅读性很好，很易懂上手编写，无非就是定义切割点，如果大切割点下需要继续切割，就加if判断，继续切割，吐个槽里面threads和workers的数量好像不管用，我压测时去看线程数对不上,看的方法是top -H -p logstash的pid。
再就是看看哪些需要计算的变成数字型，还有个timestamp的处理，这个可以看看上面的代码，对于nginx打印的时间符合ISO8601标准，可以用他做es的时间索引，这样有个好处，如果某个环节慢索引赶不上的话，日志不会错序。时间标准详细可见：http://udn.yyuap.com/doc/logstash-best-practice-cn/filter/date.html
备注：
a、尽量去掉没用的字段，精简索引，非常重要；
b、nginx打印出来的时间是标准化的，可以用它传到es作为timestamp建索引；
c、对于响应时间、响应内容大小、状态码要转换成数字类型，方便在kibana里做计算等操作；
d、切割双引号可以使用如下配置
code => "event.append(Hash[@kname.zip(event['message'].split(34.chr))])"
e、抓包后发现，logstash向es推数据是轮训的，从zookeeper取broker的相关信息并不轮训，最终logstash从zookeeper只是拿到broker的信息，然后到kafka的broker上进行数据消费读取。
f、尽量按照官方如下写法建立多个索引向es推送，防止单个索引巨大，search时计算不出来
index => "logstash-cms-nginx-%{+YYYY.MM.dd.hh}"g、测试性能方法如下
由于没有现成工具，我们用了打点计量的方式进行压测，摘掉es后将输出变为一个点，每处理一条信息打一个点，然后将打出的点用pv命令统计出字节流量，反推出logstash的吞吐量。
cp一个配置文件，修改output如下：
output {
stdout { codec => dots
workers => 1
}
}同时为了不影响线上业务，修改group_id,这样的话测试消费和线上消费互不影响，配置文件修改如下：
kafka {
zk_connect => "10.13.88.190:2181,10.13.88.191:2181,10.13.88.192:2181"
topic_id => "nginx"
group_id => "test001"
consumer_threads => 12
reset_beginning => false
decorate_events => flase
}测试时执行命令：/opt/logstash/bin/logstash -f /tmp/kafka_test.conf |pv -abt > /dev/null
压测结果如下：
每个点是一个byte，等到数据稳定后，计算每s的吞吐量为2.93*1024=3000，也就是这一个logstash最大吞吐量为能处理3000条信息每s。
四、索引（Elastic）
ES升级调优5.2.1详见博文：ELK之ES2.4.1双实例平滑升级调优至5.2.1踩坑并supervisor管理记
ES工作原理及集群调优见博文：深入浅出剖析Elasticsearch的工作原理
线上业务最早使用的是es2.x版本，后来升级到了es5.x版本，变化是比较大的，很多配置点都不一样了，但道法自然而术变万千，不管是哪个版本，所有的调整中遵循着不变的法则，从中总结归纳，其实分为下面几个层次：
1、系统层：
HEAP、GC、文件描述符、进程数调整、关闭交换分区、进程管理最大内存、系统内存回收机制调优。
#一键优化脚本 #/bin/sh #writer:gaolixu id elasticsearch || useradd elasticsearch -s /sbin/nologin #添加用户 grep "* - nofile 512000" /etc/security/limits.conf || echo "* - nofile 512000" >> /etc/security/limits.conf #修改文件描述符数量 grep "elasticsearch - nproc unlimited" /etc/security/limits.conf || echo "elasticsearch - nproc unlimited" >> /etc/security/limits.conf #修改最大打开进程数数量 grep "fs.file-max = 1024000" /etc/sysctl.conf || echo "fs.file-max = 1024000" >> /etc/sysctl.conf #修改系统文件描述符 grep "vm.max_map_count = 262144" /etc/sysctl.conf || echo "vm.max_map_count = 262144" >> /etc/sysctl.conf #修改程序最大管理的vm grep "vm.extra_free_kbytes = 4096000" /etc/sysctl.conf || echo "vm.extra_free_kbytes = 4096000" >> /etc/sysctl.conf grep "vm.min_free_kbytes = 2097152" /etc/sysctl.conf || echo "vm.min_free_kbytes = 2097152" >> /etc/sysctl.conf grep "vm.zone_reclaim_mode = 0" /etc/sysctl.conf || echo "vm.zone_reclaim_mode = 0" >> /etc/sysctl.conf sysctl -p swapoff -a #关掉虚拟内存
2、结构层：
a、master、client节点分离；
b、冷、热数据分群，共用一个client管理。
3、业务层：
a、index调优（复制分片数、shard数、刷新时间）
b、flush调优（translog控制flush频率、同步异步）
c、merge调优（segment相关参数、触发条件）
d、空闲时间强制merge（减少segment的数量）
e、内存请求熔断调优（fielddata、request内存）
f 、数据保留7天，限制查询1天（开关索引实现）
g、跟进官方新版本，并及时升级。
4、硬调优：
a、机械硬盘更换成固态硬盘SSD。
es的安装也是比较简单详见：https://www.elastic.co/guide/en/elasticsearch/reference/current/rpm.html
es的节点类型相关说明详见：https://my.oschina.net/secisland/blog/618911
常用es的集群管理命令，当然只是看信息的可以浏览器里直接输入查看
curl http://10.39.40.94:9200/_cat/nodes?v #节点概况 curl http://10.39.40.94:9200/_cat/shards?v #查看shards的信息 curl http://10.39.40.94:9200/_cat/indices?v #查看索引信息，如果新推的日志，可以看这个确认是否索引成功 curl -X DELETE "http://10.39.40.94:9200/索引名称" #删除指定历史索引，速度很快
对于我们线上的日志，默认保存7天，每天晚上清除一次，并关索引，仅能查看一天的日志数据，清除的脚本如下：
#!/bin/bash
DATE=`date +%Y.%m.%d.%I`
DATA1=`date +%Y.%m.%d -d'-2 day'`
DATA2=`date +%Y.%m.%d -d'-7 day'`
curl -XPOST -u elastic:elastic "http://10.39.40.94:9220/logstash*${DATA1}*/_close?pretty"
curl -XDELETE -u elastic:elastic "http://10.39.40.97:9220/logstash*${DATA2}*?pretty"五、展现（kibana）
展现kibana没什么可说的，直接安装后，配置好es的地址就可以用，安装很简单有rpm包，前端可以用nginx做个代理，做限制，安装详见：https://www.elastic.co/downloads/kibana
安装后模型搭建也比较人性化，用几次就熟练了。
备注：像logstash、kafka这种加&号启动的服务（有些启动后自己fork新进程然后退出的其实不合适）可以用supervisor管理，比较方便。配置相当简单，可以在浏览器看状态,后使用supervisor monitor统一管理，截屏如下：
 |
社长"矢量比特"，曾就职中软、新浪，现任职小米，致力于DevOps运维体系的探索和运维技术的研究实践. |
