Fluent Bit配置与使用——基于版本V2.2.2

Fluent Bit日志采集终端

文档适用版本：V2.2

1、日志文件处理流程

数据源是一个普通文件，其中包含 JSON 内容，使用tail插件记录日志，通过parsers进行格式化匹配（图里没写），通过两个筛选器（filter）: grep初步排除某些记录，以及record_modifier更改记录内容，添加和删除特定键，最终通过输出器（output）输出。

请添加图片描述

还可以每个步骤之间添加数据验证检查点，以便了解数据结构是否正确，我们通过使用期望过滤器（expect filter）来实现这一点。

请添加图片描述

可选值：

参数值	描述
key_exists	检查日志中是否存在Key
key_not_exists	检查日志中是否不存在Key
key_val_is_null	检查Key的值是否为Null
key_val_is_not_null	检查Key的值是否不为Null
key_val_eq	检查Key的值是否等于配置值
action	不匹配时采取的操作，警告（warn）、退出（exit），警告会记录日志，错误会直接退出（状态码：255）

示例：

[FILTER]
    name        expect
    match       *
    key_exists  color   # color键是否存在
    action      exit    # 不存在直接退出

2、安装流程

RPM包安装

tar zxf fluent-bit_v2.2.2.gz
rpm -ivh libyaml-0.1.4-11.el7_0.x86_64.rpm
rpm -ivh postgresql-libs-9.2.24-9.el7_9.x86_64.rpm
rpm -ivh fluent-bit-2.2.2-1.x86_64.rpm

总配置文件路径：/etc/fluent-bit/fluent-bit.conf

执行文件路径：/opt/fluent-bit/bin/fluent-bit

docker安装

docker pull cr.fluentbit.io/fluent/fluent-bit

kubernetes安装（Helm方式）

# 添加FluentBit Helm库
helm repo add fluent https://fluent.github.io/helm-charts
# 安装默认配置的FluentBit
helm upgrade --install fluent-bit fluent/fluent-bit

3、配置详情

配置 Fluent Bit 的方法之一是使用主配置文件。Fluent Bit 允许使用一个配置文件，该配置文件可以在全局范围内工作。

主配置文件支持 4 种类型的配置段:

Service 全局配置（在这里引入解析器parsers）
Input 输入器
Filter 过滤器
Output 输出器

此外，还可以使用函数将主配置文件拆分为多个文件，以包含外部文件:

Include File

[SERVICE]
    flush        1  # 设置引擎将由输入插件进入的记录何时由输出插件输出
    daemon       Off   # Fluent Bit 是否应该作为守护(后台)进程运行
    log_file     /var/log/fluentbit.log   # 可选日志文件的绝对路径。默认情况下，所有日志都输出到标准输出(stdout)
    log_level    info  # 输出的日志级别
    parsers_file parsers.conf  # 解析器配置文件路径。配置段中可配置多个 Parsers_File 配置项
    plugins_file plugins.conf  # plugins 配置文件中可定义外部插件的路径。
    http_server  Off   # 是否启用内置 HTTP 服务
    http_listen  0.0.0.0  # HTTP 服务启用时,监听地址 
    http_port    2020  # HTTP 服务的 TCP 端口
    
[INPUT] # INPUT 配置段定义数据源(与输入插件相关联)。请注意，每个输入插件都可以添加自己的配置项:
    name cpu   # 输入插件名称（必须项）
    tag  cpu.local   # 与该插件产生的所有记录关联的标签名称
    
[FILTER]  # FILTER 配置段定义一个过滤器(与过滤插件相关联)。请注意，每个过滤插件都可以添加自己的配置项:
    Name  stdout  # 过滤插件名称
    Match *  # 与传入记录的标签匹配的模式
    
[OUTPUT]  # UTPUT 配置段指定记录标签匹配后的目的地。该配置支持以下配置项
    name  stdout  # 输出插件名称
    match *    # 匹配输入插件标签

@INCLUDE input-nginx.conf  # 调用其它配置文件，避免单文件过长（INPUT、FILTER、OUTPUT字段均可进行外部引入）

4、数据管道

4.1 Input 插入插件

Dummy # 虚拟输入插件，用于生成虚拟事件。它对于测试、调试、基准测试和开始使用 Fluent Bit 非常有用。

HTTP # HTTP 输入插件允许您将自定义记录发送到 HTTP 端点。

Health # Health插件允许您检查 TCP 服务器的健康状况。它通过每隔一定的时间间隔发出一次 TCP 连接来执行检查。

Tail # Tail输入插件允许监控一个或多个文本文件。它具有类似于 tail -f shell 命令的行为。

Dummy插件

该插件用于生成虚拟时间记录，主要用于测试、调试。

配置参数

键	描述
Dummy	模拟的 JSON 事件记录，默认为 `{"message":"dummy"}`
Rate	每秒产生的事件数，默认为 1
Interval_sec	以秒为单位的虚拟基本时间戳。默认值：0
Interval_nsec	以纳秒为单位的虚拟基本时间戳。默认值：0
Samples	如果设置有值，则事件的数量将受到限制。例如，如果 Samples=3，插件只生成三个事件并停止。
Copies	每次生成消息时要生成的消息数。默认值为 1。
Flush_on_startup	如果设置为 `true`，则在启动时生成第一个虚拟事件。默认值：`false`
Metadata	虚拟 JSON 元数据。默认：{}
Start_time_sec	以秒为单位的虚拟基本时间戳。默认值：0
Start_time_nsec	以纳秒为单位的虚拟基本时间戳。默认值：0

示例

[INPUT]
    Name   dummy
    Dummy {"message": "custom dummy"}

[OUTPUT]
    Name   stdout
    Match  *

HTTP插件

HTTP 输入插件允许您将自定义记录发送到 HTTP 端点。

配置参数

键	描述	默认值
listen	监听地址	0.0.0.0
port	监听端口	9880
tag_key	指定要覆盖标签的Key值。如果设置，则标记将被键的值覆盖	4M
buffer_max_size	指定接收 JSON 消息的最大缓冲区大小（以 KB 为单位）。	512K
buffer_chunk_size	设置传入传入 JSON 消息的区块大小。
successful_response_code	成功响应代码	201
success_header	添加HTTP头

示例

# 模拟http请求
curl -d '{"key1":"value1","key2":"value2"}' -XPOST -H "content-type: application/json" http://localhost:8888/app.log
# 配置文件
[INPUT]
    name http
    listen 0.0.0.0
    port 8888

[OUTPUT]
    name stdout
    match app.log

Health插件

该插件允许您检查 TCP 服务器的健康状况。它通过每隔一定的时间间隔发出一次 TCP 连接来执行检查。

配置参数

键	描述
Host	要检查的目标主机或 IP 地址的名称。
Port	要检查的目标主机或 IP 地址的端口。
Interval_Sec	服务检查间隔（以秒为单位）。默认值为 1。
Internal_Nsec	服务检查指定纳秒间隔，与Interval_Sec配置键结合使用。默认值为 0。
Alert	如果启用，则仅当探活失败时，它才会生成消息。默认情况下，此选项处于禁用状态。
Add_Host	将主机名追加到每条记录。默认值为 false。
Add_Port	将端口号追加到每条记录。默认值为 false。

示例

[INPUT]
    Name          health
    Host          127.0.0.1
    Port          80
    Interval_Sec  1
    Interval_NSec 0

[OUTPUT]
    Name   stdout
    Match  *

Tail插件

该插件读取 Path 路径下的每个匹配文件，对于找到的每个新行（用换行符（\n）分隔），它会生成

一条新记录。或者，可以使用数据库文件（使用Sqlite记录），以便插件可以具有跟踪文件的历史记录

和偏移状态，用于重启后恢复进度。

配置参数

键	描述	默认值
Buffer_Chunk_Size	设置初始缓冲区大小以读取文件数据。此值用于增加缓冲区大小。	32k
Buffer_Max_Size	设置每个监控文件的缓冲区大小。如果超过此限制，则将从监控文件列表中删除该文件。	32k
Path	通过使用通配符指定一个或多个日志文件的
Path_Key	如果启用，它将附加监控文件的名称作为记录的一部分。指定的值成为映射中的键
Exclude_Path	文件排除，使用逗号分隔
Read_from_Head	从文件的头部读取内容，而不是从文件尾部读取内容。	False
Refresh_Interval	刷新监控文件列表的时间间隔(秒)
Rotate_Wait	指定监控文件的额外时间，以防止日志文件滚动丢失某些数据
Skip_Long_Lines	当监控的文件由于行(Buffer_Max_Size)很长而达到缓冲区容量时，默认行为是停止监视该文件。	Off
Skip_Empty_Lines	跳过日志中的空行	Off
DB	指定跟踪监控文件的偏移量的数据库文件
DB.Sync	设置默认sqlite同步方法。	FULL
DB.journal_mode	设置数据库的日志模式（WAL）。	WAL
Mem_Buf_Limit	指定tail插件使用的最大内存，如果达到限制，插件会停止采集，刷新数据后会恢复
Parser	指定解析器
Key	当消息是非结构化数据时(未应用解析器)，消息将以字符串形式作为 log 键的值。此选项允许为该键指定名称	log
Tag	Tag 标识数据源，用于后续处理流程Filter,output时选择数据
Tag_Regex	使用通配符方法进行数据源标识

示例

[INPUT]
    # 使用 tail 插件
    Name              tail
    # Tag 标识数据源，用于后续处理流程Filter,output时选择数据
    Tag               nginx.*
    Path              /var/log/nginx/access.log
    # 指定解析器
    Parser            json
    Mem_Buf_Limit     15MB
    # 初始buffer size
    Buffer_Chunk_Size 32k
    # 每个文件的最大buffer size
    Buffer_Max_Size   64k
    # 跳过长度大于 Buffer_Max_Size 的行，Skip_Long_Lines 若设为Off遇到超过长度的行会停止采集
    Skip_Long_Lines   On
    # 监控日志文件 refresh 间隔
    Refresh_Interval  10

4.2 Parsers 解析插件

默认情况

使用该插件，将非结构化日志条目格式化为一个具体、统一的结构，使其更容易处理和进一步过滤。

格式化类型：

Json映射

正则匹配 Fluent Bit 使用基于 Ruby 的正则表达式

配置参数

键	描述
Name	解析器名称（唯一）
Format	指定解析器格式，选项：JSON、正则、logfmt
Regex	正则表达式
Time_Key
Time_Format	指定时间字段的格式。
Time_Offset	为本地日期指定固定的 UTC 时间偏移量（例如 -0600、+0200 等）。
Time_Keep	启用此选项将使解析器在日志条目中保留原始时间字段及其值。
Skip_Empty_Values	指定一个布尔值，用于确定解析器是否应跳过空值。默认值为 `true`。

示例

# 原始日志
23-Jan-2024 18:42:06.178 信息 [Periodic background build discarder thread] hudson.model.AsyncPeriodicWork.lambda$doRun$0 Finished Periodic background build discarder. 1 ms
# 解析器
[PARSER]
    Name    tomcat
    Format  regex
    Regex   ^(?<time>\d{2}-\w{3}-\d{4} (.*).(\d{3})) (?<level>[\u4e00-\u9fa5]+) (?<class>\[.*\]) (?<msg>.*)
# 输出日志
## 将原始日志拆为了
### time字段："23-Jan-2024 18:42:06.178"；
### level字段："信息"；
### class字段："[Periodic background build discarder thread]"；
### msg字段："hudson.model.AsyncPeriodicWork.lambda$doRun$0 Finished Periodic background build discarder. 1 ms"
{
    "_time_":"2024-01-23T10:42:06.178810Z"
    "time":"23-Jan-2024 18:42:06.178"
    "level":"信息"
    "class":"[Periodic background build discarder thread]"
    "msg":"hudson.model.AsyncPeriodicWork.lambda$doRun$0 Finished Periodic background build discarder. 1 ms"
}

多行情况

多行解析器（Multiline Parser）默认内置了以下几种：

解析器	描述
docker	处理 Docker 容器引擎生成的日志条目。此解析器支持由 Docker 拆分的日志条目的串联。
cri	处理 CRI-O 容器引擎生成的日志条目。与 docker 解析器类似，用于解析containerd的日志
go	处理基于 Go 的语言应用程序生成的日志条目，并在检测到多行消息时执行串联。
python	处理基于 Python 的语言应用程序生成的日志条目，并在检测到多行消息时执行串联。
java	处理由 Java 语言应用程序生成的日志条目，并在检测到多行消息时执行串联。

可自定义的多行匹配规则

键	描述	默认值
name	为多行解析器指定名称
type	设置多行解析模式，一般为：`regex`正则
parser	事先定义解析器。注意：当解析器应用于原始文本时，正则表达式将使用 `key_content` 配置属性应用于结构化消息的特定键（见下文）。
key_content	对于传入的结构化消息，请指定包含应由正则表达式处理并可能连接的数据的键。
flush_timeout	超时刷新时间	5s
rule	定义多行匹配规则，可定义多个规则，但存在一定使用规定。

Rule规则的定义

一条规则由四部分组成

# 规则名 |  状态名称      |  正则规则       | 下一步状态
rules   |   state name   | regex pattern  | next state
# 第一步必须为“start_state”，正则与多行文本的首行匹配
rule      "start_state"   "/([a-zA-Z]+ \d+ \d+\:\d+\:\d+)(.*)/"   "cont"

4.3 Filters 过滤插件

官方文档：Expect - Fluent Bit: Official Manual

多个Filier过滤器顺序执行

CheckList插件

该插件会查找指定列表中的值是否存在，然后允许添加记录以告知是否找到。

配置参数

键	描述
file	关键字匹配文件
lookup_key	日志文件中需要查找的字段
record	当在`lookup_key`字段中查到对应值时要添加的值。value：键值
mode	设置检查模式。支持精准`exact`和模糊`partial`。默认值：`exact`。
ignore_case	是否忽略大小写。默认：false

示例：

[INPUT]
    name           tail
    tag            test1
    path           test1.log
    read_from_head true
    parser         json

[FILTER]
    name       checklist
    match      test1
    # 在日志中查找含有文件中包含关键字的部分
    file       ip_list.txt
    # 关键字所在字段
    lookup_key $remote_addr
    # 添加/替换（这里为：将`ioc`字段的值替换为abc）
    record     ioc    abc

[OUTPUT]
    name       stdout
    match      test1

Grep插件

该插件允许您根据值或嵌套值的正则表达式模式匹配或排除特定记录。

配置参数

键	格式	描述
Regex	key regex	仅保留与正则表达式匹配的字段的记录
Exclude	key regex	排除与正则表达式匹配的字段的记录
Logical_Op	Operation	Specify which logical operator to use. `AND` , `OR` and `legacy` are allowed as an Operation. Default is `legacy` for backward compatibility. In `legacy` mode the behaviour is either AND or OR depending whether the `grep` is including (uses AND) or excluding (uses OR). Only available from 2.1+.

示例

[INPUT]
    name   tail
    path   lines.txt
    parser json

[FILTER]
    name   grep
    match  *
    # 仅保留log字段值为aa的日志
    regex  log aa

[OUTPUT]
    name   stdout
    match  *

Lua插件

用于应用lua脚本

Lua - Fluent Bit: Official Manual

Parser插件（常用）

该插件用于解析事件记录中的字段。

配置参数

键	描述	默认值
Key_Name	指定要解析的日志中的具体字段名称。
Parser	指定解析器名称，允许多个解析器条目（每行一个）。
Preserve_Key	是否保留原始字段（Key_Name里写的）	False
Reserve_Data	是否保留其它原始字段	False

示例

# 原始日志
{"data":"100 0.5 true This is example"}
# 解析器
[PARSER]
    Name dummy_test
    Format regex
    Regex ^(?<INT>[^ ]+) (?<FLOAT>[^ ]+) (?<BOOL>[^ ]+) (?<STRING>.+)$
# 过滤器
[FILTER]
    Name parser
    Match dummy.*
    Key_Name data
    Parser dummy_test
# 输出
[0] dummy.data: [1499347993.001371317, {"INT"=>"100", "FLOAT"=>"0.5", "BOOL"=>"true", "STRING"=>"This is example"}]

Record Modifier插件

Record Modifier插件允许附加字段或排除特定字段。

配置参数

键	描述
Record	追加字段。值为键值对
Remove_key	如果键匹配，则删除。（黑名单）值为：mem.total
Allowlist_key	如果键不匹配，则删除。（白名单）值为：mem.total
Uuid_key	如果设置，插件会将 uuid 附加到每条记录。分配的值将成为映射中的键。

示例

# 原始记录
{"Mem.total"=>1016024, "Mem.used"=>716672, "Mem.free"=>299352, "Swap.total"=>2064380, "Swap.used"=>32656, "Swap.free"=>2031724}
# 追加
[FILTER]
    Name record_modifier
    Match *
    Record hostname ${HOSTNAME}
    Record product Awesome_Tool
# 删除Swap.used
[FILTER]
    Name record_modifier
    Match *
    Remove_key Swap.used
# 仅保留Swap.used
[FILTER]
    Name record_modifier
    Match *
    Allowlist_key Swap.used

Modify插件（常用）

详情：修改 - Fluent Bit： Official Manual

Multiline插件（常用）

该插件主要用于上下文拆分为多个日志行的消息进行合并。

默认支持格式：

GO

Python

Ruby

Java（Google Cloud Platform Java stacktrace format）

配置参数

键	描述
multiline.parser	指定一个或多个多行解析器，通过逗号分隔
multiline.key_content	用于保存处理后内容键的名称。
mode
buffer	启用缓冲模式。在缓冲模式下，筛选器可以从逐个引入记录的输入中连接多行（例如：转发），而不是以块的形式连接
flush_ms	挂起的多行记录的刷新时间。默认值为 2000。
emitter_name
emitter_storage.type
emitter_mem_buf_limit

示例

test.log通过tail插件输入，接着进入parsers_multiline.conf进行多行解析，然后调用multiline进行多行过滤，最终输出。

[SERVICE]
    flush                 1
    log_level             info
    parsers_file          parsers_multiline.conf

[INPUT]
    name                  tail
    path                  test.log
    read_from_head        true

[FILTER]
    name                  multiline
    match                 *
    multiline.key_content log
    multiline.parser      go, multiline-regex-test

[OUTPUT]
    name                  stdout
    match                 *

Nest插件

Nest 插件允许您对嵌套数据进行操作或处理嵌套数据。

配置参数

键	值格式	描述
Operation	nest / lift	选择操作：嵌套（nest）或拆解（lift）
Wildcard	通配符	记录哪个字段需要进行嵌套（nest）
Nest_under	字符串	记录新嵌套字段的键名称
Nested_under	字符串	取消嵌套（lift）的嵌套字段的键名称
Add_prefix	字符串	使用此字符串作为受影响的键的前缀（在Key前加文本）
Remove_prefix	字符串	如果前缀与此字符串匹配，请从受影响的键中删除前缀

示例

# 原始文本：{"Mem.total"=>3865308, "Mem.used"=>3648492, "Mem.free"=>216816, "Swap.total"=>4194300, "Swap.used"=>1065856, "Swap.free"=>3128444}
[INPUT]
    Name mem
    Tag  mem.local

[OUTPUT]
    Name  stdout
    Match *

# 将所有Mem.*嵌套在LAYER1下
# 输出为：{"Swap.total"=>4194300, "Swap.used"=>1065888, "Swap.free"=>3128412, "LAYER1"=>{"Mem.total"=>3865308, "Mem.used"=>3643692, "Mem.free"=>221616}
[FILTER]
    Name nest
    Match *
    Operation nest
    Wildcard Mem.*
    Nest_under LAYER1
[FILTER]
    Name nest
    Match *
    Operation nest
    Wildcard LAYER1*
    Nest_under LAYER2
# 取消LAYER2字段嵌套，并在下一级LAYER1前新增Lifted2_
# 输出为：{"Swap.total"=>4194300, "Swap.used"=>1065680, "Swap.free"=>3128620, "Lifted2_LAYER1"=>{"Mem.total"=>3865308, "Mem.used"=>3688680, "Mem.free"=>176628}}
[FILTER]
    Name nest
    Match *
    Operation lift
    Nested_under LAYER2
    Add_prefix Lifted2_

# 新增LAYER3嵌套，并删除上一级字段Lifted2_前缀
# 输出为：{"Swap.total"=>4194300, "Swap.used"=>1066244, "Swap.free"=>3128056, "LAYER3"=>{"LAYER1"=>{"Mem.total"=>3865308, "Mem.used"=>3660424, "Mem.free"=>204884}}}
[FILTER]
    Name nest
    Match *
    Operation nest
    Wildcard Lifted2_*
    Nest_under LAYER3
    Remove_prefix Lifted2_

Kubernetes插件

kubernetes 过滤器插件允许使用元数据丰富您的日志文件。

当 Fluent Bit 作为 DaemonSet 部署在 Kubernetes 并配置为从容器(使用 tail 或 systemd 输入插件)读取日志时，此过滤器可以执行以下操作:

分析 Tag 标签并提取以下元数据:

Pod Name

Namespace

Container Name

Container ID

查询 Kubernetes API Server 以获取有关 Pod 的额外元数据:

Pod ID

Labels

Annotations

数据将会缓存在本地内存中，并附加到每个日志记录上。

详情：Kubernetes - Fluent Bit：官方手册

示例

[FILTER]
    # 使用kubernetes过滤器
    Name                kubernetes
    # 匹配ingress.*这个Tag对应的 INPUT
    Match               ingress.*
    # kubernetes API Server 地址
    Kube_URL            https://kubernetes.default.svc:443
    # kubernetes 上serviceAccount的CA证书路径
    Kube_CA_File        /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
    # kubernetes 上serviceAccount的token路径
    Kube_Token_File     /var/run/secrets/kubernetes.io/serviceaccount/token
    # 当源日志来自tail插件，这个配置用于指定tail插件使用的前缀值
    Kube_Tag_Prefix     ingress.var.log.containers.
    # Merge_Log=On 解析log字段的json内容，提取到根层级, 附加到Merge_Log_Key指定的字段上
    Merge_Log           Off
    # 合并log字段后是否保持原始log字段
    Keep_Log            Off
    # 允许Kubernetes Pod 建议预定义的解析器
    K8S-Logging.Parser  Off
    # 允许Kubernetes Pod 从日志处理器中排除其日志
    K8S-Logging.Exclude Off
    # 是否在额外的元数据中包含 Kubernetes 资源标签信息
    Labels              Off
    # 是否在额外的元数据中包括 Kubernetes 资源信息
    Annotations         Off

4.4 Output 输出插件

Amazon S3

详情：Amazon S3 – Fluent Bit：官方手册

Elasticsearch

详情：Elasticsearch - Fluent Bit：官方手册

输出至文件

配置参数

键	描述	默认值
Path	用于存储文件的目录路径。
File	设置用于存储记录的文件名。
Format	文件内容的格式。（out_file、JSON、CSV、LTSV、Template）	out_file
Mkdir	是否递归创建输出目录。
Workers	专用线程数量	1

HTTP

详情：HTTP - Fluent Bit: Official Manual

kafka

详情：Kafka - Fluent Bit: Official Manual

示例

[OUTPUT]
    # 使用kafka插件
    Name              kafka
    # 匹配Nginx access日志
    Match             tomcat.*
    # 指定Kafka Brokers地址
    Brokers           192.168.142.12:9092
    # 指定Kafka topic，如果需要推送到多个topic，多个topic通过','分隔
    Topics            fluentBit-tomcat-topic
    # 将Timestamp_Key设置为_time_，原默认值为@timestamp
    Timestamp_Key     _time_
    # 指定时间戳转成成的时间字符串格式
    Timestamp_Format  iso8601
    # 设置为false表示不限制重试次数
    Retry_Limit       false
    # 当kafka结束空闲连接时，隐藏"Receive failed: Disconnected"报错
    rdkafka.log.connection.close false
    # Kafka生产者队列中总消息容量最大值，此处设置为10MB，producer buffer is not included in http://fluentbit.io/documentation/0.12/configuration/memory_usage.html#estimating
    rdkafka.queue.buffering.max.kbytes 10240
    # Kafka生产者在leader已成功收到的数据并得到确认后才发送下一条message。
    rdkafka.request.required.acks 1

5、配置实例

Nginx

输入日志："192.168.142.1 - - [22/Jan/2024:08:25:24 +0000] "GET / HTTP/1.1" 304 0 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36" "-" "

[PARSER]
    Name   nginx
    Format regex
    Regex ^(?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")
    Time_Key time
    Time_Format %d/%b/%Y:%H:%M:%S %z
[INPUT]
    Name              tail
    Tag               nginx.*
    Path              /var/log/nginx/access.log
    Parser            json
    Mem_Buf_Limit     15MB
    Buffer_Chunk_Size 32k
    Buffer_Max_Size   64k
    Skip_Long_Lines   On
    Refresh_Interval  10
[OUTPUT]
    Name              kafka
    Match             nginx.*
    Brokers           192.168.142.12:9092
    Topics            fluentBit-nginx-topic
    Timestamp_Key     _time_
    Timestamp_Format  iso8601
    Retry_Limit       false
    rdkafka.log.connection.close false
    rdkafka.queue.buffering.max.kbytes 10240
    rdkafka.request.required.acks 1

输出：

请添加图片描述

Tomcat

输入日志：[INFO ] 2024-01-22 01:42:32,748 --http-nio-8080-exec-7-- [com.hylh.mobile.biz.mq.sender.impl.QueueSenderServiceImpl] 测试发送结束

![Tomcat](F:\Work\Fluent Bit\Tomcat.png)![Tomcat](F:\Work\Fluent Bit\Tomcat.png)[MULTILINE_PARSER]
    Name    multiline-tomcat
    Type  regex
    Flush_timeout 1000
    rule      "start_state"    "/^(?<level>\[ERROR\]) (?<time>\d{4}-\d{2}-\d{2} (.*),(\d{3})) (?<thread>--(.*)--) (?<class>\[\w(.*).\w\]) (?<message>(.*))/"  "cont"
    rule      "cont"   "/(^com|org|java).\w(.*).\w(.*): \w(.*)/"  "cont"
    rule      "cont"          "/\s+at.*/"  "cont"
    
[INPUT]
    Name              tail
    Tag               tomcat.*
    Path              /opt/apache-tomcat-8.5.88/logs/catalina.out
    Parser            tomcat
    Mem_Buf_Limit     15MB
    Buffer_Chunk_Size 32k
    Buffer_Max_Size   64k
    Skip_Long_Lines   On
    Refresh_Interval  5
    
[FILTER]
    name                  multiline
    match                 tomcat.*
    multiline.key_content log
    multiline.parser      multiline-tomcat
    
[OUTPUT]
    Name              kafka
    Match             tomcat.*
    Brokers           192.168.142.12:9092
    Topics            fluentBit-tomcat-topic
    Timestamp_Key     _time_
    Timestamp_Format  iso8601
    Retry_Limit       false
    rdkafka.log.connection.close false
    rdkafka.queue.buffering.max.kbytes 10240
    rdkafka.request.required.acks 1

Docker-Tomcat

输入日志：{"log":"[ERROR] 2024-01-22 08:53:22,090 --schedule-pool-6-- [com.rx.common.utils.Threads] Invalid bound statement (not found): com.test.system.mapper.store.sytem.SysStoreOperLogMapper.insertOperlog \r\n","stream":"stdout","time":"2024-01-22T00:53:22.092252133Z"}

[MULTILINE_PARSER]
    Name    multiline-tomcat
    Type  regex
    Flush_timeout 1000
    rule      "start_state"    "/^(?<level>\[ERROR\]) (?<time>\d{4}-\d{2}-\d{2} (.*),(\d{3})) (?<thread>--(.*)--) (?<class>\[\w(.*).\w\]) (?<message>(.*))/"  "cont"
    rule      "cont"   "/(^com|org|java).\w(.*).\w(.*): \w(.*)/"  "cont"
    rule      "cont"          "/\s+at.*/"  "cont"
[PARSER]
    Name    get_level
    Format  regex
    Regex   (?<level>\[\w{4,5} *\])

[INPUT]
    Name              tail
    Tag               dockerTomcat.*
    Path              /opt/test-json.log
    Read_from_head    true
    Mem_Buf_Limit     15MB
    # 指定多行匹配，先按照docker模式进行分割
    multiline.parser docker, cri
    Buffer_Chunk_Size 32k
    Buffer_Max_Size   64k
    Skip_Long_Lines   On
    Refresh_Interval  5

[FILTER]
    name                  multiline
    match                 dockerTomcat.*
    multiline.key_content log
    multiline.parser      multiline-tomcat
[FILTER]
    Name            parser
    match           dockerTomcat.*
    Key_Name        log
    Parser          get_level
    Reserve_Data    True
    Preserve_Key    True

[OUTPUT]
    Name              kafka
    Match             dockerTomcat.*
    Brokers           192.168.142.12:9092
    Topics            fluentBit-docker-tomcat-topic
    Timestamp_Key     _time_
    Timestamp_Format  iso8601
    Retry_Limit       false
    rdkafka.log.connection.close false
    rdkafka.queue.buffering.max.kbytes 10240
    rdkafka.request.required.acks 1