date: 2020-05-28title: filebeat配置详解 #标题
tags: filebeat #标签
categories: elastic stack # 分类

记录下filebeat的一些总结。

input配置

基础配置模板

示例1：

用于相同配置项的日志输入。

  1. filebeat.inputs:
  2. - type: log
  3.   paths:
  4.     - /var/log/messages      # 收集messages日志
  5.     - /var/log/*.log        # 收集/var/log目录下所有.log结尾的日志

示例2：
若要针对不同的日志定义不同的配置，则需要使用此模板。

  1. filebeat.inputs:
  2. - type: log 
  3.   paths:
  4.     - /var/log/system.log
  5.     - /var/log/wifi.log
  6. - type: log 
  7.   paths:
  8.     - "/var/log/apache2/*"
  9.   fields:        # fields表示在日志消息中添加字段
  10.     apache: access

示例3

采集了配置日志切割的日志源，参考：官方文档。

filebeat.inputs常规配置

参考：

管理多行消息输入

filebeat默认收集消息是以行来读取的，类似于linux中的tailf指令，那么当需要收集java日志的时候，就需要配置管理多行消息处理，java应用日志格式一般如下：

  1. [beat-logstash-some-name-832-2015.11.28] IndexNotFoundException[no such index]
  2.     at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver$WildcardExpressionResolver.resolve(IndexNameExpressionResolver.java:566)
  3.     at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.concreteIndices(IndexNameExpressionResolver.java:133)
  4.     at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.concreteIndices(IndexNameExpressionResolver.java:77)
  5.     at org.elasticsearch.action.admin.indices.delete.TransportDeleteIndexAction.checkBlock(TransportDeleteIndexAction.java:75)

如果日志源是上面那种的，那么就需要在filebeat.inputs模块中写入以下配置：

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

上面各选项的解释如下：

  • multiline.pattern：指定要匹配的正则表达模式，需要注意的是，filebeat支持的正则表达式模式和logstash支持的模式有些不同，关于filebeat支持的正则表达式请参考官方文档。
  • multiline.negate：定义是否为否定模式，默认为false。
  • multiline.match：指定filebeat如何将匹配的行组合到事件中，设置为after或before，这些设置的行为取决于我们为negate以下内容指定的内容： | 设置negate为 | 设置match为 | 结果 | 示例：pattern:^b | | —- | —- | —- | —- | | false | after | 匹配模式的连续行将追加到不匹配的前一行 | filebeat之output详解 - 图1 | | false | before | 与该模式匹配的连续行被预置到不匹配的下一行 | filebeat之output详解 - 图2 | | true | after | 与模式不匹配的连续行被追加到匹配的前一行 | filebeat之output详解 - 图3 | | true | before | 与该模式不匹配的连续行被预置到匹配的下一行 | filebeat之output详解 - 图4 |
  • multiline.flush_pattern：指定一个正则表达式，其中当前多行将从内存中刷新，从而结束多行消息；
  • multiline.max_lines：可以合并到一个事件的最大行数。如果多行消息包含的超过 ，则丢弃任何其他行。默认值为 500。
  • multiline.timeout：指定超时后，Filebeat 会发送多行事件，即使找不到启动新事件的新模式也是如此。默认值为 5。

多行配置示例

这里将写下一下几个示例：

  • 将 Java 堆栈跟踪合并到单个事件中
  • 将 C 样式的行延续合并到单个事件中
  • 从时间戳事件组合多行

java堆栈跟踪

Java 堆栈跟踪由多行组成，每行在初始行之后以空格开头，如以下示例所示：

  1. xception in thread "main" java.lang.NullPointerException
  2.         at com.example.myproject.Book.getTitle(Book.java:16)
  3.         at com.example.myproject.Author.getBookTitles(Author.java:25)
  4.         at com.example.myproject.Bootstrap.main(Bootstrap.java:14)

要将上面的日志合并到filebeat的单个事件中，可以使用下面的多行配置：

  1. multiline.pattern: '^[[:space:]]'
  2. multiline.negate: false
  3. multiline.match: after

我这里启动了logstash来接受filebeat发来的日志消息，logstash配置文件内容如下：

  1. input {
  2.   beats {
  3.     port => 5044
  4.   }
  5.  }
  6. output {
  7.    stdout {
  8.       codec => rubydebug
  9.    }
  10. }
  11. $ logstash -f test.conf          # 启动此配置文件

filebeat配置文件如下：

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.     - /var/log/java/access.log       
  6.   multiline.pattern: '^[[:space:]]'
  7.   multiline.negate: false
  8.   multiline.match: after
  9. # 上面的multiline配置是合并以空格开头到上一行的任何行。
  11. output.logstash:
  12.   enabled: true
  13.   hosts: ["logstash:5044"]      # 输出到logstash的5044端口
  14. $ filebeat run -c filebeat.yml        # 启动filebeat

logstash输出的消息如下：

filebeat之output详解 - 图5

可以看到filebeat将多行日志整合成了一行发送到了logstash。如果不配置multiline….相关参数，那么就只会导致日志以行被拆分发送。

稍微复杂些

日志样例如下：

  1. Exception in thread "main" java.lang.IllegalStateException: A book has a null property
  2.        at com.example.myproject.Author.getBookIds(Author.java:38)
  3.        at com.example.myproject.Bootstrap.main(Bootstrap.java:14)
  4. Caused by: java.lang.NullPointerException
  5.        at com.example.myproject.Book.getId(Book.java:22)
  6.        at com.example.myproject.Author.getBookIds(Author.java:35)
  7.        ... 1 more

filebeat配置如下：

  1. cat filebeat.yml 
  2. filebeat.inputs:
  3. - type: log
  4.   enabled: true
  5.   paths:
  6.     - /var/log/nginx/access.log    
  7. # 下面匹配的是以空格开头，后面为at或...的和前一行合并为一行
  8. # 以及由Caused by产生的新行
  9.   multiline.pattern: '^[[:space:]]+(at|\.{3})[[:space:]]+\b|^Caused by:'
  10.   multiline.negate: false
  11.   multiline.match: after
  12. output.logstash:
  13.   enabled: true
  14.   hosts: ["es2:5044"]

logstash收到的消息内容如下：

filebeat之output详解 - 图6

可以看到两行内容合并为一行发送到了logstash。

如果想让上面的两段日志分两行发过去，那么需要更改如下：

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.     - /var/log/nginx/access.log  
  6.   multiline.pattern: '^[[:space:]]+(at|\.{3})'
  7.   multiline.negate: false
  8.   multiline.match: after
  9. output.logstash:
  10.   enabled: true
  11.   hosts: ["es2:5044"]

此时，logstash收到的消息如下：

filebeat之output详解 - 图7

filebeat之output详解 - 图8

线延续

有些日志是以反斜杠“\”来进行续行的，如下：

  1. printf ("%10.10ld  \t %10.10ld \t %s\
  2.   %f", w, x, y, z );

那么对应的filebeat配置文件就应该更改如下：

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.     - /var/log/nginx/access.log  
  6. # 此配置将以\字符结尾的任何行与后面的行合并。
  7.   multiline.pattern: '\\$'
  8.   multiline.negate: false
  9.   multiline.match: before
  10. output.logstash:
  11.   enabled: true
  12.   hosts: ["es2:5044"]

此时，logstash收到的消息如下：

filebeat之output详解 - 图9

时间戳开头的日志

也有些服务的日志是以下面的格式开头的，比如es或logstash，日志信息如下：

  1. [2015-08-24 11:49:14,389][INFO ][env                      ] [Letha] using [1] data paths, mounts [[/
  2. (/dev/disk1)]], net usable_space [34.5gb], net total_space [118.9gb], types [hfs]

那么如果要将上面这种类型的日志合并为一行，则需要使用一下配置：

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.     - /var/log/nginx/access.log
  6. # 此配置使用negate: true和match: after设置来指定与指定模式不匹配的任何行属于前一行。
  7.   multiline.pattern: '^\[[0-9]{4}-[0-9]{2}-[0-9]{2}'
  8.   multiline.negate: true
  9.   multiline.match: after
  10. output.logstash:
  11.   enabled: true
  12.   hosts: ["es2:5044"]

logstash收到的消息如下：

filebeat之output详解 - 图10

应用程序日志

有时应用程序日志包含事件，这些事件以自定义标记开始和结束，如下:

  1. [2015-08-24 11:49:14,389] Start new event
  2. [2015-08-24 11:49:14,395] Content of processing something
  3. [2015-08-24 11:49:14,399] End event

那么filebeat中的配置应该这么写：

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.     - /var/log/nginx/access.log 
  6.   multiline.pattern: 'Start new event'
  7.   multiline.negate: true
  8.   multiline.match: after
  9.   multiline.flush_pattern: 'End event'
  10. output.logstash:
  11.   enabled: true
  12.   hosts: ["es2:5044"]

logstash收到的消息如下：

filebeat之output详解 - 图11

filebeat中支持的配置选项

参考：官方文档。

input过滤文件或行

在filebeat收集日志时，可能日志中的某些行，如debug开头的行，或者paths指定了一个目录下的所有文件，但又不需要input某个文件，那么此时需要将其排除。

exclude_lines—匹配要filebeat排除的行

使用此选项，支持正则表达式，filebeat会删除列表中与正则表达式匹配的所有行，默认情况下，不删除任何行，空行将被忽略。

如果还指定了多行设置，则需要在用过滤隔行之前，将每条消息合并为一行，再进行过滤。

以下是将filebeat配置为删除以DBG开头的任何行

  1. filebeat.inputs:
  2. - type: log
  3.   ...
  4.   exclude_lines: ['^DBG']

有关filebeat支持的正则表达式请参考官方文档

include_lines—仅input与列表中正则表达式匹配的行

可以通过include_lines来匹配我们要输入的行，匹配的行将被input，不匹配的则忽略（默认所有行都被input）。

同样，如果还指定了多行设置，则需要在用过滤隔行之前，将每条消息合并为一行，再进行过滤。

  1. filebeat.inputs:
  2. - type: log
  3.   ...
  4.   include_lines: ['^ERR', '^WARN']

注：如果同时定义了include_lines和exclude_lines，则filebeat的include_lines首先执行，然后再执行exclude_lines。定义两个选项的顺序无关紧要，include_lines选项将始终在exclude_lines选项之前执行。

以下是导出所有包含api-stkp，但以DBG开头的行除外：

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.     - /var/log/nginx/access.log      
  6.   include_lines: ['api-stkp']
  7.   exclude_lines: ['^DBG']
  8. output.logstash:
  9.   enabled: true
  10.   hosts: ["es2:5044"]

日志文件如下：

  1. 10.252.201.136 - - [27/Apr/2020:00:00:01 +0800] "POST /api-stkp/callback HTTP/1.1" 200 221 "-" "Java/1.8.0_77"
  2. DBG--10.252.201.136 - - [27/Apr/2020:00:00:01 +0800] "POST /api-stkp/callback HTTP/1.1" 200 221 "-" "Java/1.8.0_77"
  3. 10.252.201.136 - - [27/Apr/2020:00:00:01 +0800] "POST /api-stkp/callback HTTP/1.1" 200 221 "-" "Java/1.8.0_77"

logstash收到的消息如下：

filebeat之output详解 - 图12

filebeat之output详解 - 图13

可以看到以DBG开头那行并没有输出到logstash。

exclude_files

忽略与exclude_files匹配的文件，默认情况下不排除任何文件。

有些时候，我们在定义input的paths路径时，可能直接指定了一个目录，而非一个文件，如下：

  1.   paths:
  2.     - /var/log/nginx/*          # 此时该路径下的所有文件将被采集

如果按照上面的配置，那么如果有个文件并不是日志文件呢？可能又同学会说，我可以定义paths为/var/log/nginx/*.log啊，对，没错，可以的，但是，这就需要你保证，所有要采集的日志都是以.log结尾的，而不能有其他结尾的日志文件，如.out等。那么，如果日志结尾不唯一，就需要用到了exclude_files这个选项，用于排除极个别不需要input的文件。示例如下：

  1. filebeat.inputs:
  2. - type: log
  3.   ...
  4.   exclude_files: ['\.gz$']   # 将排除以.gz结尾的文件

关于其支持的正则表达式，请参考官方文档。

output配置

output.logstash

参考：官方文档。

若要直接将收集的日志输出到logstash，则进行以下配置即可：

  1. output.logstash:
  2.   hosts: ["127.0.0.1:5044"]     # 此处指定logstash的地址及端口（端口默认5044，可以不指定，也可以指定其他的端口）

发送到Logstash的每个事件都包含以下元数据字段，我们可以在Logstash中使用这些字段进行索引和筛选:

  1. {
  2.     ...
  3.     "@metadata": { 
  4.       "beat": "filebeat", 
  5.       "version": "7.6.2" 
  6.     }
  7. }

一般会在logstash的filter中使用mutate模块将其删除掉，如下：

  1. filter {
  2. ...........
  3.      mutate {
  4.             remove_field => "beat"      
  5.             remove_field => "version"       # 7.x以下版本可能需要指定为："@version" 
  6.     }

也可以不删除其生成的元数据，直接将其作为存储到es中的索引名称（一般不会这么做）如下：

  1. input {
  2.   beats {
  3.     port => 5044    # 指定从哪个端口号接受日志，要和filebeat输出指定的端口号一致
  4.   }
  5. }
  7. output {
  8.   elasticsearch {
  9.     hosts => ["http://localhost:9200"]
  10.     index => "%{[@metadata][beat]}-%{[@metadata][version]}"    # 这里可以调用那些元数据
  11.   }
  12. }
  13. # 我在生产环境上是这么定义索引的，如下（按日期来每天生成一个索引）：
  14. index => "nginx-access-%{+YYYY.MM.dd}"

output.logstash支持的其他部分配置选项

  • enabled：用于启用或禁用输出。如果设置为false，则禁用输出，默认值为true。一般不用配置此选项。
  • hosts：指定logstash的监听地址及端口，端口若没指定，则使用默认的5044端口。如果指定了多个logstash主机，则随机选择一个主机（没有优先级），如果该主机连接失败，则再随机选择另一个主机，当然，也可以配置负载均衡。
  • loadbalance：如果指定了多个logstash节点，那么可以将此选项配置为true，表示以负载均衡的方式将日志消息发送到各个节点。
  • ttl：指定与logstash建立连接多久后重新建立连接，指定此选项的意义在于，由于和logstash之间的连接是粘性的，这样的话，在经过负载均衡后，可能由于某些配置造成logstash节点的压力差距比较大，所以需要指定ttl时间来重新建立连接。默认值为0。
  • compression_level：指定gzip的压缩等级，0表示禁用压缩，范围是1~9，数字越大，压缩越高，网络开销越小，但cpu压力会越大，默认压缩等级为3。
  • escape_html：配置字符串中的html转义，默认为false，设置为true可以启用转义，但一般不用配置此选项。
  • worker：配置主机发布事件到Logstash的工作者数。这最好在启用负载平衡模式时使用。例如:如果指定了2个hosts和3个worker，那么总共启动6个worker(每个主机3个worker)。
  • index：指定要写入事件的索引名称，默认为beat。
  • 更多选项请移步官方文档

output.kafka

在工作中的elk环境正是采用这种架构的，filebeat将收集的日志输出到kafka。

参考：官方文档。

配置示例

  1. filebeat.inputs:
  2. - type: log
  3.   enabled: true
  4.   paths:
  5.     - /var/gb/logs/site/nginx/access.log
  6.   fields:
  7.     log_topics: gt_third_ngx   # 给日志信息添加个自定义的字段
  8. output.kafka:
  9.   hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]    # 指定kafka主机列表
  10.   topics: '%{[fields.log_topics]}'      # 指定写入到哪个topic，就是上面input添加的字段命名的
  11.   partition.round_robin:
  12.     reachable_only: false
  13.   loadbalance: true   # 如果定义的hosts为多个，则可以添加此选项，以便配置为负载均衡

上面的配置支持的kafka版本为0.11到2.1.0。

关于输出到kafka，其实没有太多有用的配置，基本上面示例的配置就够用了，定义了kafka主机列表及topic名称即可，其他配置使用默认值即可。若对其他配置项有兴趣，参考官方文档吧。

output.redis

参考：官方文档。

配置示例：

  1. output.redis:
  2.   hosts: ["localhost:9679"]
  3.   password: "my_password"
  4.   key: "filebeat"
  5.   db: 0
  6.   timeout: 5

支持的配置项：

  • hosts：指定redis主机列表及端口号。
  • index：编辑添加到事件元数据以供Loqstash使用的索引名。默认的是”filebaeat”。
  • key：发布事件的Redis列表或通道的名称。如果未配置，则使用索引设置的值。
  • db：输出到redis的哪个库（redis默认有16个库）

示例：

  1. filebeat.inputs:
  2.          ............
  3.   fields:
  4.     list: ngx_access 
  5. output.redis:
  6.   hosts: ["localhost"]
  7.   key: "%{[fields.list]:fallback"   # 如果list值不存在，则回退使用索引值

更多选项请移步官方文档。