FluentBit解析Python多行日志失效问题求助

我在用FluentBit解析aodh-evaluator的Python日志，单行日志能正常解析，但带堆栈信息的多行日志会被拆成单行处理，无法合并成一条完整日志。

日志示例 #

2024-01-03 13:49:59.217 1742204 WARNING oslo_db.sqlalchemy.utils [-] Unique keys not in sort_keys. The sorting order may be unstable.
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator [-] alarm evaluation cycle failed: sqlalchemy.exc.ProgrammingError: (pymysql.err.ProgrammingError) (1146, "Table 'aodh.alarm' doesn't exist")
[SQL: SELECT alarm.alarm_id AS alarm_alarm_id, alarm.enabled AS alarm_enabled, alarm.name AS alarm_name, alarm.type AS alarm_type, alarm.severity AS alarm_severity, alarm.description AS alarm_description, alarm.timestamp AS alarm_timestamp, alarm.user_id AS alarm_user_id, alarm.project_id AS alarm_project_id, alarm.state AS alarm_state, alarm.state_reason AS alarm_state_reason, alarm.state_timestamp AS alarm_state_timestamp, alarm.ok_actions AS alarm_ok_actions, alarm.alarm_actions AS alarm_alarm_actions, alarm.insufficient_data_actions AS alarm_insufficient_data_actions, alarm.repeat_actions AS alarm_repeat_actions, alarm.rule AS alarm_rule, alarm.time_constraints AS alarm_time_constraints, alarm.evaluate_timestamp AS alarm_evaluate_timestamp
FROM alarm
WHERE alarm.type != %(type_1)s AND alarm.evaluate_timestamp < %(evaluate_timestamp_1)s AND alarm.enabled = true ORDER BY alarm.timestamp DESC]
[parameters: {'type_1': 'event', 'evaluate_timestamp_1': datetime.datetime(2024, 1, 3, 13, 49, 29, 216630)}]
(Background on this error at: https://sqlalche.me/e/14/f405)
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator Traceback (most recent call last):
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator   File "/usr/lib/python3/dist-packages/sqlalchemy/engine/base.py", line 1802, in _execute_context
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator     self.dialect.do_execute(
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator   File "/usr/lib/python3/dist-packages/sqlalchemy/engine/default.py", line 732, in do_execute
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator     cursor.execute(statement, parameters)
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator   File "/usr/lib/python3/dist-packages/pymysql/cursors.py", line 148, in execute
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator     result = self._query(query)
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator   File "/usr/lib/python3/dist-packages/pymysql/cursors.py", line 310, in _query
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator     conn.query(q)
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator   File "/usr/lib/python3/dist-packages/pymysql/connections.py", line 548, in query
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator     self._affected_rows = self._read_query_result(unbuffered=unbuffered)
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator   File "/usr/lib/python3/dist-packages/pymysql/connections.py", line 775, in _read_query_result
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator     result.read()
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator   File "/usr/lib/python3/dist-packages/pymysql/connections.py", line 1156, in read
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator     first_packet = self.connection._read_packet()
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator   File "/usr/lib/python3/dist-packages/pymysql/connections.py", line 725, in _read_packet
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator     packet.raise_for_error()
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator   File "/usr/lib/python3/dist-packages/pymysql/protocol.py", line 221, in raise_for_error
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator     err.raise_mysql_exception(self._data)
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator   File "/usr/lib/python3/dist-packages/pymysql/err.py", line 143, in raise_mysql_exception
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator     raise errorclass(errno, errval)
2024-01-03 13:49:59.228 1742204 ERROR aodh.evaluator pymysql.err.ProgrammingError: (1146, "Table 'aodh.alarm' doesn't exist")

当前FluentBit配置 #

[
    {
        "output": [
            ["name", "kafka"],
            ["match", "*"],
            ["brokers", "10.110.210.196:9092"],
            ["topics", "aodh"],
            ["Timestamp_Format", "iso8601"],
            ["format", "json"],
            ["match", "aodh_log"]
        ]
    },
    {
        "multiline_parser": [
            ["name", "aodh-parser"],
            ["type", "regex"],
            ["flush_timeout", "1000"],
            ["rule", "\"start_state\"", "\"^(\\d+-\\d+-\\d+ \\d+:\\d+:\\d+\\.\\d+) (\\d+) (ERROR|WARNING|INFO) ([a-zA-Z.]+) \\[([^\\]]+)\\] (.*)\""],
            ["rule", "\"cont\"", "\"^(?!\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}\\.\\d+).*\""],
            ["merge_previous", "true"]
        ]
    },
    {
        "input": [
            ["name", "tail"],
            ["path", "/var/log/aodh/aodh-evaluator.log, /var/log/aodh/aodh-expirer.log, /var/log/aodh/aodh-listener.log, /var/log/aodh/aodh-notifier.log"],
            ["path_key", "logpath"],
            ["tag", "aodh_log"],
            ["parser", "aodh-parser"]
        ]
    },
    {
        "filter": [
            ["Name", "record_modifier"],
            ["Match", "*"],
            ["Record", "hostname", "${HOSTNAME}"]
        ]
    }
]

问题原因及修复方案 #

问题点分析 #

1. Multiline Parser规则格式错误：规则里的引号嵌套和转义有问题，导致正则无法正确匹配；同时缺少状态流转配置，匹配起始行后没有切换到续行状态。
2. Tail插件参数错误：使用了parser参数处理多行日志，而FluentBit中parser仅用于单行解析，多行解析需要用multiline_parser参数。

修正后的配置 #

[
    {
        "output": [
            ["name", "kafka"],
            ["match", "*"],
            ["brokers", "10.110.210.196:9092"],
            ["topics", "aodh"],
            ["Timestamp_Format", "iso8601"],
            ["format", "json"],
            ["match", "aodh_log"]
        ]
    },
    {
        "multiline_parser": [
            ["name", "aodh-parser"],
            ["type", "regex"],
            ["flush_timeout", "1000"],
            // 匹配日志起始行，切换到cont状态
            ["rule", "start_state", "^(\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}\\.\\d+) (\\d+) (ERROR|WARNING|INFO) ([a-zA-Z.]+) \\[([^\\]]+)\\] (.*)", "cont"],
            // 匹配所有非起始行的内容，保持cont状态
            ["rule", "cont", "^(?!\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}\\.\\d+).*", "cont"],
            ["merge_previous", "true"]
        ]
    },
    {
        "input": [
            ["name", "tail"],
            ["path", "/var/log/aodh/aodh-evaluator.log, /var/log/aodh/aodh-expirer.log, /var/log/aodh/aodh-listener.log, /var/log/aodh/aodh-notifier.log"],
            ["path_key", "logpath"],
            ["tag", "aodh_log"],
            // 替换为multiline_parser处理多行日志
            ["multiline_parser", "aodh-parser"]
        ]
    },
    {
        "filter": [
            ["Name", "record_modifier"],
            ["Match", "*"],
            ["Record", "hostname", "${HOSTNAME}"]
        ]
    }
]

关键修改说明 #

• Multiline规则修正：去掉规则中多余的转义引号，给起始行规则添加next_state为cont，确保匹配到起始行后，后续行进入续行判断逻辑；续行规则同样指定next_state为cont，持续合并非起始行内容。
• Tail插件参数替换：将parser改为multiline_parser，让插件使用多行解析器来处理日志，确保带堆栈的多行内容被合并成一条日志。

内容的提问来源于stack exchange，提问作者Alexandru Dan