отправка логов apache kafka в elasticsearch с разделением по полям

[safronowmax]

Здравствуйте, подскажите пожалуйста. Есть кластер apache kafka (версия 2.8). Необходимо логи kafka (server.log) отправлять в elasticsearch с помощью fluent bit (fluent bit пишет напрямую в elasticsearch). Логи отправляются и попадают в elasticsearch, но проблема в том, что лог не разделяется по полям, пишется все сообщение в одно поле log, хотя я настроил regexp для логов kafka. Что не так в моей конфигурации? Или проблема все-таки в некорректном regexp?
Ниже конфиг fluent-bit.conf для apache kafka:

Код: Выделить всё

[INPUT]
    Name            tail
    Path            /home/kafka/kafka/logs/server.log
    Parser          apache-kafka
    Tag             apache.kafka
    Refresh_Interval 5
    Mem_Buf_Limit   5MB
    Skip_Long_Lines On

Код: Выделить всё

[OUTPUT]
    Name            es
    Match           apache.kafka
    Host            192.168.0.15
    Port            9200
    Index           apache-kafka-%Y.%m.%d

Ниже конфиг parsers.conf для apache kafka

Код: Выделить всё

[PARSER]
   Name    apache-kafka
   Format  regex
   Regex   ^\[(?<timestamp>[^\]]+)\] (?<log_level>\S+) \[(?<thread_name>[^\]]+)\] (?<log_message>.*)$
   Time_Key     timestamp
   Time_Format  %Y-%m-%d %H:%M:%S,%L

Ниже кусок лога server.log, который и должен попадать в elasticsearch:

Код: Выделить всё

[2023-06-14 14:14:05,787] INFO [ExpirationReaper-0-AlterAcls]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2023-06-14 14:14:05,900] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Starting socket server acceptors and processors (kafka.network.SocketServer)
[2023-06-14 14:14:05,903] INFO [/config/changes-event-process-thread]: Starting (kafka.common.ZkNodeChangeNotificationListener$ChangeEventProcessThread)
[2023-06-14 14:14:05,951] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Started data-plane acceptor and processor(s) for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
[2023-06-14 14:14:05,951] INFO [SocketServer listenerType=ZK_BROKER, nodeId=0] Started socket server acceptors and processors (kafka.network.SocketServer)
[2023-06-14 14:14:05,954] INFO Kafka version: 2.8.2 (org.apache.kafka.common.utils.AppInfoParser)
[2023-06-14 14:14:05,954] INFO Kafka commitId: 3146c6ff4a24cc24 (org.apache.kafka.common.utils.AppInfoParser)
[2023-06-14 14:14:05,954] INFO Kafka startTimeMs: 1686741245951 (org.apache.kafka.common.utils.AppInfoParser)
[2023-06-14 14:14:05,954] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
[2023-06-14 14:14:06,003] INFO [broker-0-to-controller-send-thread]: Recorded new controller, from now on will use broker fluentbit-cl1:9092 (id: 0 rack: null) (kafka.server.BrokerToControllerRequestThread)

[Bizdelnick]

Сходу вижу, что часть, соответствующая \[(?<thread_name>[^\]]+)\], присутствует не во всех строках, и в ряде случаев после неё идёт двоеточие, отсутствующее в регулярке.