使用filebeat收集ES集群运行日志和慢日志并写入到ES
背景
Elasticsearch集群运行过程中，运行日志和慢日志能够帮助集群使用者迅速定位出现的问题。鉴于Elasticsearch的一大应用场景是日志收集，因此我们尝试使用filebeat收集Elasticsearch集群各节点中的运行日志和慢日志，并写入到另一个公共的Elasticsearch集群中，使用Kibana进行日志检索。
为什么是filebeat
filebeat归属于Beats家族，使用go语言开发，是一个轻量的日志收集器，因为轻量所以适用于部署在需要收集日志的服务器中。相比之下，另一个可用于日志收集的logstash组件就比较笨重了，运行于JVM中，占用服务器资源比filebeat多，所以不适用直接部署在服务器中,但是logstash对已采集数据的清洗、过滤等处理能力要比filebeat强。通常的日至系统架构中，将filebeat部署在服务器中用于收集日志，然后写入到单独部署的logstash集群中，经logstash对日志内容进行统一处理之后，再写入到Elasticsearch集群中去。
实战过程
Elasticsearch集群运行日志和慢日志内容分析
首先要分析一下要收集的日志格式，确定日志每一行记录的固定pattern，以及是否要对日志内容进行解析，提取重要字段或者过滤一些无用的字段。
以下是Elasticsearch集群的运行日志和慢日志示例(5.6.4版本)：
运行日志：
[2018-10-23T20:42:37,451][INFO ][o.e.h.n.Netty4HttpServerTransport] [1540298502000001009] publish_address {10.0.64.80:16703}, bound_addresses {0.0.0.0:16703}
[2018-10-23T20:42:37,451][INFO ][o.e.n.Node ] [1540298502000001009] started
[2018-10-23T20:42:37,463][INFO ][o.e.g.GatewayService ] [1540298502000001009] recovered [0] indices into cluster_state
慢日志：
[2018-10-28T12:04:17,307][WARN ][index.indexing.slowlog.index] [1540298502000001009] [pmc/wCALr6BfRm-sr3qOQuGX
Xw] took[18.6ms], took_millis[18], type[articles], id[AWa41-J9c0s1mOPvR6F3], routing[] , source[]
从以上日志内容可以看出，运行日志是的格式为"[时间戳][日志级别][类名][节点名][日志详细信息]", 慢日志的格式为"[时间戳][日志级别][日志类别][节点名][日志详细信息]"。
需要完成的解析工作为：
- 解析出时间戳，并替换默认的@timestamp字段，并且保证时区为中国时间
- 解析出日志级别，作为一个单独的字段，便于检索
- 每一行日志中去除已经解析的时间戳和日志字段
- 解析出异常日志，与异常日志的前一行合并作为一条记录
- 为了节省存储空间，去除filebeat默认增加的beat.name,beat.version,beat.host等无关紧要的字段
定义日志解析pipeline
filebeat对收集到的日志处理能力是比较弱的，并且为了提高日志收集性能，一般不在filebeat中进行日志内容的处理，可以借助于logstash强大的日志处理能力或者Elasticsearch的ingest pipeline功能对日志内容进行处理。
Elasticsearch的ingest pipeline可查阅官方文档了解更多ingest api.
基于上一节定义的五项日志处理工作，前三项可以由ingest pipeline解决，下面定义名为es-log-pipeline的pipeline:
{
"description": "es-log-pipeline",
"processors": [{
"grok": {
"field": "message",
"patterns": [
"^\\[%{TIMESTAMP_ISO8601:timestamp}\\]\\[%{LOGLEVEL:level}%{SPACE}\\]%{GREEDYDATA:message}"
]
},
"remove": {
"field": "@timestamp"
}
}, {
"date": {
"field": "timestamp",
"formats": ["ISO8601"],
"timezone": "Asia/Shanghai",
"ignore_failure": true
},
"remove": {
"field": "timestamp"
}
}],
"on_failure": [{
"set": {
"field": "error.message",
"value": "{{ _ingest.on_failure_message }}"
}
}]
}
主要使用了grok processors进行日志解析，解析出了日志中的时间戳字段和日志级别字段，并将时间戳替换为filebeat默认提供的@timestamp字段。
调用Elasticsearch 的ingest API创建上述pipeline:
curl -X PUT _ingest/pipeline/es-log-pipeline
定义filebeat配置
日志解析工作的4和5两项可以由filebeat处理，定义filebeat.yml配置文件：
filebeat.prospectors:
- input_type: log
paths:
- {{.runlogPath}}
fields:
type: "runlog"
ip: "{{.ip}}"
fields_under_root: true
multiline.pattern: '^\[[0-9]{4}-[0-9]{2}-[0-9]{2}'
multiline.negate: true
multiline.match: after
- input_type: log
paths:
- {{.searchSlowlogPath}}
fields:
type: "slowlog"
logtype: "search"
ip: "{{.ip}}"
fields_under_root: true
multiline.pattern: '^\[[0-9]{4}-[0-9]{2}-[0-9]{2}'
multiline.negate: true
multiline.match: after
- input_type: log
paths:
- {{.indexSlowLogPath}}
fields:
type: "slowlog"
logtype: "indexing"
ip: "{{.ip}}"
fields_under_root: true
multiline.pattern: '^\[[0-9]{4}-[0-9]{2}-[0-9]{2}'
multiline.negate: true
multiline.match: after
processors:
- drop_fields:
fields: ["beat", "offset", "source"]
output.elasticsearch:
hosts: ["{{.esEndPoint}}"]
index: "es-%{[type]}-%{+yyyy.MM.dd}"
pipeline: "es-log-pipeline"
上述配置按天创建了两个索引，分别为es-runlog-%{+yyyy.MM.dd}和es-slowlog-%{+yyyy.MM.dd}，分别存储运行日志和慢日志。
经过上述配置，启动filebeat, 就可以实现收集Elasticsearch集群的运行日志和慢日志并写入到另外一个Elasticsearch集群中。
总结
- 日志格式解析的工作比较繁琐，需要详细了解grok processor的处理能力grok processor
- filebeat目录下有名为filebeat.template.json的文件，该文件定义了filebeat的默认模板，如果需要可以修改该模板配置文件或者自定义新模板。
