使用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": [
        "^(?m)\[%{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的默认模板，如果需要可以修改该模板配置文件或者自定义新模板。