filebeat日志收集到elasticsearch

filebeat是轻量级日志收集框架，go语言开发。需要在每个日志收集的终端部署，配置日志文件路径。可以将日志收集到es，logstash，这里以收集到es为例。配置主要分为input和out两块。解压后有filebeat.yml配置文件，主要针对该文件进行配置。

- type: log
#日志文件位置
  paths:
    - /data/logs/*/*.log
output.elasticsearch:
  #es连接信息
  hosts: ["localhost:9200"]
  protocol: "http"
  username: "elastic"
  password: "888888" 

会自动创建一个 "filebeat-%{[agent.version]}-%{+yyyy.MM.dd}-%{index_num}"

例：filebeat-7.12.1-2023.05.16-000001索引文件

索引创建规则

默认使用es的索引声明周期策略

index lifecycle management (ILM) 生成索引

配置ILM

#auto false true
setup.ilm.enabled: auto
#索引别名
setup.ilm.rollover_alias: "filebeat"
#索引增加策略
setup.ilm.pattern: "{now/d}-000001"

setup.ilm.enabled默认值auto，自动使用es中filebeat生命周期策略创建索引

setup.ilm.rollover_alias默认值filebeat-%{[agent.version]} ，创建索引时指定索引别名。

setup.ilm.pattern默认值%{now/d}-000001，索引rollover增加策略。

自动生成的索引名就是使用alias+pattern。类似filebeat-7.12.1-2023.05.16-000001这种。

更多配置参考：https://www.elastic.co/guide/en/beats/filebeat/7.17/ilm.html

自定义索引文件

output.elasticsearch可以指定index，使用自定义索引第一步就是要关闭ILM,

setup.ilm.enabled: false

下一步要配置setup.template.name和setup.template.pattern

setup.template.name: "filebeat"
setup.template.pattern: "filebeat-*"
setup.template.overwrite: false

在output.elasticsearch指定index

index: "spring-%{[agent.version]}-%{+yyyy.MM.dd}"

运行就会自动生成索引spring-7.12.1-2023.05.16。index定义可以使用上下文定义变量。可以在input里自定义field

fields:
    level: system
    region: A1

自定义的fields会一并push到索引中，index中使用自定义的fields

index: "spring-%{[fields.region]}-%{[agent.version]}-%{+yyyy.MM.dd}"

会生成索引：spring-a1-7.12.1-2023.05.16。这里A1自动转成小写了。

日志多行合并

默认情况下收集日志一行一条记录，有些情况下比如格式化输出，异常栈。一条完整的日志会包含多行数据。这时候就需要配置多行匹配。配置项在filebeat.inputs里

multiline.pattern: '^\['
multiline.negate: true
multiline.match: after

multiline.pattern指定日志匹配正则，这里'^['就是匹配以 [ 开头的行。这个地方的具体格式就要合实际输出的日志格式相匹配了。

negate和match两个参数结合使用，没太看懂，理解其来感觉有点绕，自己看官方演示例子吧https://www.elastic.co/guide/en/beats/filebeat/7.17/multiline-examples.html，有个表格图例。大体意思就是遇到不匹配的是向上合并还是向下合并，归属于那一条。这里配置true和after就是不匹配的格式行归属到上一个匹配的结果行。

根据条件写入不同索引

output.elasticsearch:
  hosts: ["http://localhost:9200"]
  indices:
    - index: "warning-%{[agent.version]}-%{+yyyy.MM.dd}"
      when.contains:
        message: "WARN"
    - index: "error-%{[agent.version]}-%{+yyyy.MM.dd}"
      when.contains:
        message: "ERR"

判断message内容，是否包含某些内容。不做演示。

收集到的日志可在kibana 日志功能界面化查看检索。需要配置日志索引匹配模式，例如上面的我们就需要新增匹配日志模式spring-*。

最后filebeat.yml有效配置大概这样

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /data/logs/*/*.log

  fields:
    level: system
    region: A1

  multiline.pattern: '^#\['
  multiline.negate: true
  multiline.match: after

output.elasticsearch:
  hosts: ["localhost:9200"]
  protocol: "http"
  username: "elastic"
  password: "888888"
  index: "spring-%{[fields.region]}-%{[agent.version]}-%{+yyyy.MM.dd}"

setup.ilm.enabled: false
setup.template.name: "filebeat"
setup.template.pattern: "filebeat-*"
setup.template.overwrite: false

参考

下载地址：https://www.elastic.co/cn/downloads/past-releases/filebeat-7-17-1

guide：https://www.elastic.co/guide/en/beats/filebeat/7.17/index.html