Parsing multiline python logs in fluentbit issue

Question

I want to parse some python logs (from aodh-evaluator) with fluentbit and multiline parsing doesn`t work.

This is an example of logs:

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")

Single line logs are correctly parsed, and multiline stacktraced logs are parsed line by line.

I tried with this fluentbit config:

[
    {
        "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}"]
        ]
    }
]```