使用Fluentd tail插件监听JSON日志时出现Pattern不匹配错误

看起来你的问题出在日志解析的格式匹配上——Fluentd的tail插件默认按行读取日志，而你的日志文件里每行并不是完整的JSON对象（从错误信息里的" \"FileClass\": \"timitry\","就能看出来，这只是JSON的一个字段行，不是完整的结构）。下面分两种场景给你解决方案：

场景1：可以修改Python日志输出（最优方案） #

如果能调整Python代码的日志生成逻辑，直接输出每行一个完整的JSON对象（也就是JSON Lines格式），那你的原配置基本不用改，只需要确保Python输出的是紧凑的JSON字符串，不要格式化换行。

比如Python代码里可以这么写：

import json
import time

log_data = {
    "FileNo": 232,
    "FileClass": "timitry",
    "FileLevel": "24",
    "DataCount": 5,
    "Data": {
        "User1": "userA",
        "User2": "userB",
        "User3": "userC",
        "User4": "userD",
        "User5": "userE"
    },
    "time": time.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
}

# 输出紧凑的JSON字符串，不要用indent参数
print(json.dumps(log_data))

这样生成的日志每行都是完整的JSON，Fluentd的format json就能正常解析，不会再出现匹配错误。

场景2：无法修改Python日志（只能处理多行JSON） #

如果日志是格式化后的多行JSON（每个日志条目占多行），你需要配置Fluentd的多行解析规则，先把多行合并成一个完整的JSON对象，再解析成结构化数据。修改你的配置如下：

<source>
  @type tail
  path /home/user/Documents/logs/file_log.json
  tag first
  pos_file /var/log/fluentd/file_log.pos  # 记录读取位置，避免重启后重复采集
  <parse>
    @type multiline
    # 匹配JSON的起始行（以{开头）
    format_firstline /^\{/
    # 匹配后续所有行，直到JSON结束
    format1 /^(?<content>.*)$/
  </parse>
</source>

# 添加filter插件，把合并后的多行内容解析成JSON
<filter first*>
  @type parser
  key_name content  # 把multiline采集到的content字段解析成JSON
  reserve_data true  # 保留原始content字段（可选）
  <parse>
    @type json
    time_key time  # 指定日志中的time字段作为时间戳
    time_format %Y-%m-%dT%H:%M:%S.%N  # 匹配日志中的时间格式
  </parse>
</filter>

<match first*>
  @type elasticsearch
  hosts 192.168.111.456:9200
  user <username>
  password <password>
  index_name fluentd-%Y%m%d  # 按日期创建索引，方便后续管理（可选）
  flush_interval 30s  # 调整刷新间隔，适配你5分钟的日志更新频率（可选）
</match>

额外检查项 #

• 确保Fluentd有读取日志文件的权限：执行chmod 644 /home/user/Documents/logs/file_log.json，或者给fluentd用户添加文件所在目录的访问权限。
• 测试配置是否正确：用fluentd -c your_config.conf -v启动Fluentd，查看控制台输出，确认没有匹配错误；也可以手动添加一条完整的JSON日志到文件，看是否能正常传到Elasticsearch。

内容的提问来源于stack exchange，提问作者Anna Carolina