filebeat生产环境配置

配置文件属性

生产配置

filebeat.inputs: 
- type: log
  enabled: true
  paths: 
  - /tmp/logs/*.log
  - /var/log/system.log
  - /var/log/wifi.log
  symlinks: true
  json.keys_under_root: true
  json.message_key: xxx
  json.add_error_key: true
  # 如果想卡部分日志，比如用时间作为过滤条件，加上下面两个选项，一个是时间筛选，一个是句柄是否关闭
  ignore_older: 168h
  close_inactive: 24h
  
  multiline.pattern: '^\s+|^"'
  multiline.match: after
  fields:
    type: fb-urmp-xx-test
  fields_under_root: true
output.console:
  pretty: true

参数释义

paths

待收集日志的路径列表

symlinks

使用 symlinks 属性的主要场景包括：

日志轮转：有些日志轮转工具会创建符号链接指向最新的日志文件。启用 symlinks 后，Filebeat 可以确保收集到最新的日志内容。
一致性路径：在一些系统中，符号链接可能用于统一路径，而实际文件位置可能会变化。通过设置 symlinks，Filebeat 可以稳定地读取这些符号链接指向的日志文件。

假设你有一个应用程序，它生成的日志文件在 /var/log/myapp/ 目录下，并且有一个日志轮转机制。轮转机制会将旧日志重命名并创建一个符号链接指向最新的日志文件，如下所示：

/var/log/myapp/
├── myapp.log -> myapp.log.20240628
├── myapp.log.20240626
├── myapp.log.20240627
└── myapp.log.20240628

在上面的结构中，myapp.log 是一个符号链接，指向 myapp.log.20240628，这是最新的日志文件。

不使用 symlinks

如果你在 Filebeat 配置中没有启用 symlinks：

这种情况下，Filebeat 只会读取实际的日志文件，即 myapp.log.20240626、myapp.log.20240627 和 myapp.log.20240628，不会读取 myapp.log（符号链接）。如果日志轮转机制仅更新符号链接而不生成新文件，Filebeat 可能会漏掉最新的日志数据。

如果你在 Filebeat 配置中启用了 symlinks：

这种情况下，Filebeat 不仅会读取实际的日志文件 myapp.log.20240626、myapp.log.20240627 和 myapp.log.20240628，还会读取 myapp.log 所指向的最新日志文件 myapp.log.20240628。这样，即使日志轮转机制只更新符号链接，Filebeat 也能获取到最新的日志内容。

json_xxx

json.keys_under_root

这个属性用于指定 JSON 键是否应该被解析为顶级字段。当设置为 true 时，意味着 JSON 对象中的所有键都将被解析为顶级字段，而不是嵌套在 message 字段下，并且没有message字段了

比如，我现在日志目录里面生成一条json日志：

{"timestamp":"2024-06-28T12:00:00Z","loglevel":"INFO444","logs":"User logged in"}

使用之后：

{
  "@timestamp": "2024-06-28T09:44:06.150Z",
  ...,
  ...,
  "timestamp": "2024-06-28T12:00:00Z",
  "loglevel": "INFO444",
  "logs": "User logged in",
  "input": {
    "type": "log"
  },
  ...,
}

json.add_error_key

当设置为 true 时，如果 JSON 解析过程中出现错误，Filebeat 会在输出中添加一个 json_error 键，并将错误信息存储在这个键下。这有助于调试和识别解析错误。

比如，我现在日志目录里面生成一条json日志：(注意看，不符合json规范)

{"timestamp":"2024-06-28T12:00:00Z","loglevel":"INFO444","logs":"User logged in

设置json.add_error_key: true

效果：

{
  "@timestamp": "2024-06-28T09:45:29.164Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.16.3"
  },
  ...,
  //看这里，解析错误会多生成这个
  "error": {
    "message": "Error decoding JSON: invalid character '\\n' in string literal",
    "type": "json"
  },
  "message": "{\"timestamp\":\"2024-06-28T12:00:00Z\",\"loglevel\":\"INFO444\",\"logs\":\"User logged in"
}

json.message_key

建议搭配json.add_error_key: true来使用

使用这个属性的话，采集的日志没有message字段

这个值用来在解析json错误的时候，将日志的源值添加到这个字段中

json.message_key 设置一个日志原值已经存在的字段，如果json解析正常的话，那么该属性继续显示日志原值的值

如果设置一个日志原值已经不存在的字段，json解析正常的话，那么该属性的值为空串

如果解析失败的话，那么就会将日志的原值，添加到这个字段中

比如日志原值为：
{"timestamp":"2024-06-28T12:00:00Z","loglevel":"INFO444","logs":"User logged in"}
设置json.message_key:logs的话，json解析正常的话，没有用，但是filbeat采集的日志没有message字段
设置json.message_key:test的话,json解析正常的话，fileat采集的日志多出一个test字段，但是test为空串，没有message字段

比如我设置为 json.message_key: logs,json.add_error_key: true

那么我日志里面生成一条数据

{"timestamp":"2024-06-28T12:00:00Z","loglevel":"INFO444","logs":"User logged in

filebeat采集的信息为：

{
  "@timestamp": "2024-06-28T09:57:15.426Z",
  ...,
  //看这里，logs存储的是日志的原值
  "logs": "{\"timestamp\":\"2024-06-28T12:00:00Z\",\"loglevel\":\"INFO444\",\"logs\":\"User logged in",
  "log": {
    "offset": 7704,
    "file": {
      "path": "/tmp/logs/urmp-manager.log"
    }
  },
  ...,
}

如果日志json正确解析的话，filebeat采集的日志为：

{
  "@timestamp": "2024-06-28T10:10:19.555Z",
  ...,
  "loglevel": "INFO444",
  ...,
  "logs": "User logged in",
  "timestamp": "2024-06-28T12:00:00Z",
  "input": {
    "type": "log"
  }
...
}

multiline_xx

multiline.pattern

指定用于匹配多行的正则表达式

multiline.match

multiline.pattern: ‘^\s’
multiline.match: after

指定Filebeat如何把多行合并成一个事件。可选的值是 after 或者 before。

如果match=after，则以空格开头的和前面一行将合并成一条完整日志；如果 match=before，则以空格开头的和后面一行将合并成一条完整日志。

multiline.negate

默认为false，表示匹配pattern的行合并到上一行；true表示不匹配pattern的行合并到上一行；

假如

multiline.pattern: ‘^\s’
multiline.match: after

源日志的内容为：

{"timestamp": "2024-07-09 16:45:23.650","traceId": "[SW_CTX:[test::urmp-manager,urmp-manager-test-75564f678c-swvmg,d529c220b9a045a98e5eb9ea233ac90d.165.17205147234650001,d529c220b9a045a98e5eb9ea233ac90d.165.17205147234650000,0]]","level": "ERROR","service": "urmp-manager", "thread": "http-nio-8080-exec-4", "class": "com.hero.commons.core.handlers.MyExceptionHandler", "msg": "error", "exception":"com.hero.commons.core.exception.MyCustomException: 系统错误
	at com.hero.commons.core.handlers.MyExceptionHandler.handleException(MyExceptionHandler.java:160)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:568)
	at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:207)
	at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:152)
	at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:118)
	at org.springframework.web.servlet.mvc.method.annotation.ExceptionHandlerExceptionResolver.doResolveHandlerMethodException(ExceptionHandlerExceptionResolver.java:413)
	at org.springframework.web.servlet.handler.AbstractHandlerMethodExceptionResolver.doResolveException(AbstractHandlerMethodExceptionResolver.java:74)
" }

如果multiline.negate:true

那么采集的日志内容为：

{
  ...
  "error": {
    "message": "Error decoding JSON: invalid character 'a' looking for beginning of value",
    "type": "json"
  },
  "ori": "\tat org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:118)",
  "fb_collect_app": "urmp-manager-test"
}
{
 ...
  "ori": "\tat org.springframework.web.servlet.mvc.method.annotation.ExceptionHandlerExceptionResolver.doResolveHandlerMethodException(ExceptionHandlerExceptionResolver.java:413)",
  "system_env": "test",
}
{
  ...
  "ori": "\tat org.springframework.web.servlet.handler.AbstractHandlerMethodExceptionResolver.doResolveException(AbstractHandlerMethodExceptionResolver.java:74)\n\" }",
}

注意看
源日志最后一行" }规则不符合pattern的格式，所以他合并到上一行了，其他符合的并没有合并到上一行

如果multiline.negate:false

采集的日志为

{
 ...,
  "ori": "{\"timestamp\": \"2024-07-09 17:01:48.143\",\"traceId\": \"[SW_CTX:[test::urmp-manager,urmp-manager-test-6db7b7b7bf-5rmvc,77065c0d2a864a28b9a2802143f5b690.166.17205157080370001,77065c0d2a864a28b9a2802143f5b690.166.17205157080370000,0]]\",\"level\": \"ERROR\",\"service\": \"urmp-manager\", \"thread\": \"http-nio-8080-exec-5\", \"class\": \"com.hero.commons.core.handlers.MyExceptionHandler\", \"msg\": \"error\", \"exception\":\"com.hero.commons.core.exception.MyCustomException: 系统错误\n\tat com.hero.commons.core.handlers.MyExceptionHandler.handleException(MyExceptionHandler.java:160)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)\n\tat java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:568)\n\tat org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:207)\n\tat org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:152)\n\tat org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:118)\n\tat org.springframework.web.servlet.mvc.method.annotation.ExceptionHandlerExceptionResolver.doResolveHandlerMethodException(ExceptionHandlerExceptionResolver.java:413)\n\tat org.springframework.web.servlet.handler.AbstractHandlerMethodExceptionResolver.doResolveException(AbstractHandlerMethodExceptionResolver.java:74)",
 ...
}

multiline.pattern

指定一个正则表达式，多行将从内存刷新到磁盘

multiline.max_lines

可以合并成一个事件的最大行数。如果一个多行消息包含的行数超过max_lines，则超过的行被丢弃。默认是500。

fields

topic 对应的消息字段或自定义增加的字段。

fields_under_root

如果值为ture，那么fields存储在输出文档的顶级位置

document_type

跟fields字段意思差不多，fields是用来替换他的，他是早期的产物

exclude_lines

exclude_lines: [‘^DBG’] #不包含匹配正则的行

include_lines

include_lines: [‘^ERR’, ‘^WARN’] #包含匹配正则的行

close_inactive

启动选项时，如果在制定时间没有被读取，将关闭文件句柄

ingore_older

#默认为0，表示禁用，可以配置2h，2m等，注意ignore_older必须大于close_inactive的值。

表示忽略超过设置值未更新的文件或者文件从来没有被harvester收集

filebeat工作原理

filebeat结构：由两个组件构成，分别是inputs（输入）和harvesters（收集器），这些组件一起工作来跟踪文件并将事件数据发送到您指定的输出，harvester负责读取单个文件的内容。harvester逐行读取每个文件，并将内容发送到输出。为每个文件启动一个harvester。harvester负责打开和关闭文件，这意味着文件描述符在harvester运行时保持打开状态。如果在收集文件时删除或重命名文件，Filebeat将继续读取该文件。这样做的副作用是，磁盘上的空间一直保留到harvester关闭。默认情况下，Filebeat保持文件打开，直到达到close_inactive

关闭harvester可以会产生的结果：

文件处理程序关闭，如果harvester仍在读取文件时被删除，则释放底层资源。
只有在scan_frequency结束之后，才会再次启动文件的收集。
如果该文件在harvester关闭时被移动或删除，该文件的收集将不会继续

一个input负责管理harvesters和寻找所有来源读取。如果input类型是log，则input将查找驱动器上与定义的路径匹配的所有文件，并为每个文件启动一个harvester。每个input在它自己的Go进程中运行，Filebeat当前支持多种输入类型。每个输入类型可以定义多次。日志输入检查每个文件，以查看是否需要启动harvester、是否已经在运行harvester或是否可以忽略该文件

Filebeat目前支持两种prospector类型：log和stdin。

Filebeat如何保持文件的状态?

Filebeat保留每个文件的状态，并经常将状态刷新到磁盘中的注册表文件中。该状态用于记住harvester读取的最后一个偏移量，并确保发送所有日志行。如果无法访问输出（如Elasticsearch或Logstash），Filebeat将跟踪最后发送的行，并在输出再次可用时继续读取文件。当Filebeat运行时，每个输入的状态信息也保存在内存中。当Filebeat重新启动时，来自注册表文件的数据用于重建状态，Filebeat在最后一个已知位置继续每个harvester。对于每个输入，Filebeat都会保留它找到的每个文件的状态。由于文件可以重命名或移动，文件名和路径不足以标识文件。对于每个文件，Filebeat存储唯一的标识符，以检测文件是否以前被捕获。

filebeat何如保证至少一次数据消费

Filebeat保证事件将至少传递到配置的输出一次，并且不会丢失数据。是因为它将每个事件的传递状态存储在注册表文件中。在已定义的输出被阻止且未确认所有事件的情况下，Filebeat将继续尝试发送事件，直到输出确认已接收到事件为止。如果Filebeat在发送事件的过程中关闭，它不会等待输出确认所有事件后再关闭。当Filebeat重新启动时，将再次将Filebeat关闭前未确认的所有事件发送到输出。这样可以确保每个事件至少发送一次，但最终可能会有重复的事件发送到输出。通过设置shutdown_timeout选项，可以将Filebeat配置为在关机前等待特定时间