Multiline parser does not works for multiple inputs

[seriy379]

Bug Report

Describe the bug
I have a cluster of Kubernetes with 2 pods and I want to compile logs from each module separately.
But the multiline parser only works for the first INPUT and does not work for the second INPUT

To Reproduce
My config

  fluent-bit.conf: |
    [SERVICE]
        Flush                     60
        Log_Level                 info
        Daemon                    off
        Parsers_File              parsers.conf
        HTTP_Server               Off
        storage.path              /var/fluent-bit/state/flb-storage/
        storage.sync              normal
        storage.checksum          off
        storage.backlog.mem_limit 5M

    [FILTER]
        Name                kubernetes
        Match               application.*
        Kube_URL            https://kubernetes.default.svc:443
        Kube_Tag_Prefix     application.var.log.containers.
        Merge_Log           On
        K8S-Logging.Parser  On
        K8S-Logging.Exclude Off
        Labels              Off
        Annotations         Off

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

    @INCLUDE account-log.conf
    @INCLUDE api-log.conf

  account-log.conf: |
    [INPUT]
        Name                tail
        Tag                 application.account.*
        Path                /var/log/containers/account*.log
        Mem_Buf_Limit       5MB
        Multiline.parser    cri
        Refresh_Interval    30
        Skip_Long_Lines     On
        Read_from_Head      On

    [OUTPUT]
        Name                cloudwatch_logs
        Match               application.account.*
        region              ${AWS_REGION}
        log_group_name      ${CLUSTER_NAME}/account
        log_stream_prefix   ${VERSION}-${HOST_NAME}-
        auto_create_group   true
        extra_user_agent    container-insights

  api-log.conf: |
    [INPUT]
        Name                tail
        Tag                 application.api.*
        Path                /var/log/containers/api*.log
        Mem_Buf_Limit       5MB
        Multiline.parser    cri
        Refresh_Interval    30
        Skip_Long_Lines     On
        Read_from_Head      On

    [OUTPUT]
        Name                cloudwatch_logs
        Match               application.api.*
        region              ${AWS_REGION}
        log_group_name      ${CLUSTER_NAME}/api
        log_stream_prefix   ${VERSION}-${HOST_NAME}-
        auto_create_group   true
        extra_user_agent    container-insights

  parsers.conf: |
    [MULTILINE_PARSER]
        name          multiline-regex
        type          regex
        flush_timeout 1000
        # rules | state name    | regex pattern               | next state
        # ------|---------------|-----------------------------------------
        rule      "start_state"   "/.*Error.*/"                 "cont"
        rule      "cont"          "/^\s+.*|^}$/"                "cont"

Expected behavior
All INPUTs and OUTPUTs correctly parse multiline logs and merge them to one log

Screenshots
First pod: multiline work

Second pod: multiline does not work

Your Environment
KubernetesVersion: 1.20.10
Image: amazon/aws-for-fluent-bit:2.23.2

[robert-heinzmann-logmein]

I see a similar behaviour with 1.9.3 on k8s with a single input but 2 containers/pods on the same node. Both pods logs java exceptions, but only the first pods if properly logging a single java exception, the second is not. After a restart of the fluentbit and if I change order then the other one is properly sending a single exception.

Reproduce:

Version1:

• start fluent bit
• log multiline java exception in pod1
• log multiline java exception in pod2

Result: pod1 exception == single record, pod 2 exception == one record per line

Version2:

• start fluent bit
• log multiline java exception in pod2
• log multiline java exception in pod1

Result: pod2 exception == single record, pod 1 exception == one record per line

Config:

    [INPUT]
        Name            tail
        Parser            docker
        Tag                kube.*
        Alias              container_logs
        Path              /var/log/containers/*.log
        DB                /var/lib/fluent-bit/container-logs.db
        multiline.parser  docker, cri
        Mem_Buf_Limit     5MB
        
    [FILTER]
        name                  multiline
        match                 kube.*
        multiline.key_content log
        multiline.parser      java, go
        

[robert-heinzmann-logmein]

Actually I created a simple reproducer for this that works locally and uses the java example from the manual.

Reproducer

shell# tree -f /root/reproducer/
/root/reproducer
├── /root/reproducer/etc
│   └── /root/reproducer/etc/fluent-bit.conf
└── /root/reproducer/logs
    ├── /root/reproducer/logs/pod1.log
    └── /root/reproducer/logs/pod2.log

2 directories, 3 files

# Very slim config file
shell# cat /root/reproducer/etc/fluent-bit.conf
[SYSTEM]
    Flush 1

[INPUT]
    Name              tail
    Tag               *
    Alias             container_logs
    Path              /var/log/containers/*.log
    DB                /var/lib/fluent-bit/container-logs.db
    multiline.parser  docker, cri
    Mem_Buf_Limit     5MB
    Read_from_Head    true

[FILTER]
    name                  multiline
    match                 *
    multiline.key_content log
    multiline.parser      java, go
    #debug_flush           True

[OUTPUT]
    Name             stdout
    Match            *
    Format           json_lines
    json_date_key    time
    json_date_format iso8601

# Java exception example from the documentation as logged by pod1 (docker) using "json-file"
shell# cat /root/reproducer/logs/pod1.log
{"log":"Dec 14 06:41:08 Exception in thread \\\"main\\\" java.lang.RuntimeException: Something has gone wrong, aborting!\n","stream":"stdout","time":"2022-05-22T15:02:37.349634795Z"}
{"log":"    at com.myproject.module.MyProject.badMethod(MyProject.java:22)\n","stream":"stdout","time":"2022-05-22T15:02:37.349663935Z"}
{"log":"    at com.myproject.module.MyProject.oneMoreMethod(MyProject.java:18)\n","stream":"stdout","time":"2022-05-22T15:02:37.349668234Z"}
{"log":"    at com.myproject.module.MyProject.anotherMethod(MyProject.java:14)\n","stream":"stdout","time":"2022-05-22T15:02:37.349671048Z"}
{"log":"    at com.myproject.module.MyProject.someMethod(MyProject.java:10)\n","stream":"stdout","time":"2022-05-22T15:02:37.349673835Z"}
{"log":"    at com.myproject.module.MyProject.main(MyProject.java:6)\n","stream":"stdout","time":"2022-05-22T15:02:37.349676423Z"}
{"log":"\n","stream":"stdout","time":"2022-05-22T15:02:37.349679166Z"}

# Java exception example from the documentation as logged by pod2 (docker) using "json-file"
shell# cat /root/reproducer/logs/pod2.log
{"log":"Dec 14 06:41:08 Exception in thread \\\"main\\\" java.lang.RuntimeException: Something has gone wrong, aborting!\n","stream":"stdout","time":"2022-05-22T15:04:02.399655026Z"}
{"log":"    at com.myproject.module.MyProject.badMethod(MyProject.java:22)\n","stream":"stdout","time":"2022-05-22T15:04:02.399683818Z"}
{"log":"    at com.myproject.module.MyProject.oneMoreMethod(MyProject.java:18)\n","stream":"stdout","time":"2022-05-22T15:04:02.399687928Z"}
{"log":"    at com.myproject.module.MyProject.anotherMethod(MyProject.java:14)\n","stream":"stdout","time":"2022-05-22T15:04:02.399690723Z"}
{"log":"    at com.myproject.module.MyProject.someMethod(MyProject.java:10)\n","stream":"stdout","time":"2022-05-22T15:04:02.399693759Z"}
{"log":"    at com.myproject.module.MyProject.main(MyProject.java:6)\n","stream":"stdout","time":"2022-05-22T15:04:02.39969641Z"}
{"log":"\n","stream":"stdout","time":"2022-05-22T15:04:02.399702217Z"}

Output

# Now let's run the reproducer
shell# docker run --rm --name reproducer \
  --mount type=tmpfs,destination=/var/lib/fluent-bit \
  -v /root/reproducer/etc:/fluent-bit/etc \
  -v /root/reproducer/logs:/var/log/containers \
  fluent/fluent-bit:1.9.3-debug \
  /fluent-bit/bin/fluent-bit -c /fluent-bit/etc/fluent-bit.conf  -v

Fluent Bit v1.9.3
* Copyright (C) 2015-2022 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

[2022/05/22 15:58:46] [ info] Configuration:
[2022/05/22 15:58:46] [ info]  flush time     | 1.000000 seconds
[2022/05/22 15:58:46] [ info]  grace          | 5 seconds
[2022/05/22 15:58:46] [ info]  daemon         | 0
[2022/05/22 15:58:46] [ info] ___________
[2022/05/22 15:58:46] [ info]  inputs:
[2022/05/22 15:58:46] [ info]      tail
[2022/05/22 15:58:46] [ info] ___________
[2022/05/22 15:58:46] [ info]  filters:
[2022/05/22 15:58:46] [ info]      multiline.0
[2022/05/22 15:58:46] [ info] ___________
[2022/05/22 15:58:46] [ info]  outputs:
[2022/05/22 15:58:46] [ info]      stdout.0
[2022/05/22 15:58:46] [ info] ___________
[2022/05/22 15:58:46] [ info]  collectors:
[2022/05/22 15:58:46] [ info] [fluent bit] version=1.9.3, commit=9eb4996b7d, pid=1
[2022/05/22 15:58:46] [debug] [engine] coroutine stack size: 24576 bytes (24.0K)
[2022/05/22 15:58:46] [ info] [storage] version=1.2.0, type=memory-only, sync=normal, checksum=disabled, max_chunks_up=128
[2022/05/22 15:58:46] [ info] [cmetrics] version=0.3.1
[2022/05/22 15:58:46] [debug] [tail:container_logs] created event channels: read=21 write=22
[2022/05/22 15:58:46] [ info] [input:tail:container_logs] multiline core started
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] flb_tail_fs_inotify_init() initializing inotify tail input
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] inotify watch fd=29
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] scanning path /var/log/containers/*.log
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] inode=158339697 with offset=0 appended as /var/log/containers/pod1.log
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] scan_glob add(): /var/log/containers/pod1.log, inode 158339697
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] inode=158339698 with offset=0 appended as /var/log/containers/pod2.log
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] scan_glob add(): /var/log/containers/pod2.log, inode 158339698
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] 2 new files found on path '/var/log/containers/*.log'
[2022/05/22 15:58:46] [ info] [filter:multiline:multiline.0] created emitter: emitter_for_multiline.0
[2022/05/22 15:58:46] [debug] [emitter:emitter_for_multiline.0] created event channels: read=34 write=35
[2022/05/22 15:58:46] [debug] [stdout:stdout.0] created event channels: read=37 write=38
[2022/05/22 15:58:46] [debug] [router] match rule tail.0:stdout.0
[2022/05/22 15:58:46] [debug] [router] match rule emitter.1:stdout.0
[2022/05/22 15:58:46] [ info] [sp] stream processor started
[2022/05/22 15:58:46] [ info] [output:stdout:stdout.0] worker #0 started
[2022/05/22 15:58:46] [ info] [filter:multiline:multiline.0] created new multiline stream for tail.0_var.log.containers.pod1.log
[2022/05/22 15:58:46] [debug] [filter:multiline:multiline.0] Created new ML stream for tail.0_var.log.containers.pod1.log
[2022/05/22 15:58:46] [debug] [input chunk] update output instances with new chunk size diff=516
[2022/05/22 15:58:46] [debug] [input chunk] update output instances with new chunk size diff=68
[2022/05/22 15:58:46] [ info] [filter:multiline:multiline.0] created new multiline stream for tail.0_var.log.containers.pod2.log
[2022/05/22 15:58:46] [debug] [filter:multiline:multiline.0] Created new ML stream for tail.0_var.log.containers.pod2.log
[2022/05/22 15:58:46] [debug] [input chunk] update output instances with new chunk size diff=177
[2022/05/22 15:58:46] [debug] [input chunk] update output instances with new chunk size diff=135
[2022/05/22 15:58:46] [debug] [input chunk] update output instances with new chunk size diff=139
[2022/05/22 15:58:46] [debug] [input chunk] update output instances with new chunk size diff=139
[2022/05/22 15:58:46] [debug] [input chunk] update output instances with new chunk size diff=136
[2022/05/22 15:58:46] [debug] [input chunk] update output instances with new chunk size diff=128
[2022/05/22 15:58:46] [debug] [input chunk] update output instances with new chunk size diff=68
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] [static files] processed 1.8K
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] inode=158339697 file=/var/log/containers/pod1.log promote to TAIL_EVENT
[2022/05/22 15:58:46] [ info] [input:tail:container_logs] inotify_fs_add(): inode=158339697 watch_fd=1 name=/var/log/containers/pod1.log
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] inode=158339698 file=/var/log/containers/pod2.log promote to TAIL_EVENT
[2022/05/22 15:58:46] [ info] [input:tail:container_logs] inotify_fs_add(): inode=158339698 watch_fd=2 name=/var/log/containers/pod2.log
[2022/05/22 15:58:46] [debug] [input:tail:container_logs] [static files] processed 0b, done
[2022/05/22 15:58:47] [debug] [task] created task=0x7f9b6483d620 id=0 OK
[2022/05/22 15:58:47] [debug] [task] created task=0x7f9b6483d690 id=1 OK
[2022/05/22 15:58:47] [debug] [output:stdout:stdout.0] task_id=0 assigned to thread #0
[2022/05/22 15:58:47] [debug] [output:stdout:stdout.0] task_id=1 assigned to thread #0
{"log":"Dec 14 06:41:08 Exception in thread \\\"main\\\" java.lang.RuntimeException: Something has gone wrong, aborting!\n    at com.myproject.module.MyProject.badMethod(MyProject.java:22)\n    at com.myproject.module.MyProject.oneMoreMethod(MyProject.java:18)\n    at com.myproject.module.MyProject.anotherMethod(MyProject.java:14)\n    at com.myproject.module.MyProject.someMethod(MyProject.java:10)\n    at com.myproject.module.MyProject.main(MyProject.java:6)\n","stream":"stdout","time":"2022-05-22T15:02:37.349634795Z"}
{"log":"\n","stream":"stdout","time":"2022-05-22T15:02:37.349679166Z"}

----- MULTILINE FLUSH (stream_id=6284141986831512414) -----
[0] [1653231757.349634795, {"log"=>"Dec 14 06:41:08 Exception in thread \"main\" java.lang.RuntimeException: Something has gone wrong, aborting!
    at com.myproject.module.MyProject.badMethod(MyProject.java:22)
    at com.myproject.module.MyProject.oneMoreMethod(MyProject.java:18)
    at com.myproject.module.MyProject.anotherMethod(MyProject.java:14)
    at com.myproject.module.MyProject.someMethod(MyProject.java:10)
    at com.myproject.module.MyProject.main(MyProject.java:6)
", "stream"=>"stdout", "time"=>"2022-05-22T15:02:37.349634795Z"}]
----------- EOF -----------

----- MULTILINE FLUSH (stream_id=6284141986831512414) -----
[0] [1653231757.349679166, {"log"=>"
", "stream"=>"stdout", "time"=>"2022-05-22T15:02:37.349679166Z"}]
----------- EOF -----------

----- MULTILINE FLUSH (stream_id=2148379754670323614) -----
[0] [1653231842.399655026, {"log"=>"Dec 14 06:41:08 Exception in thread \"main\" java.lang.RuntimeException: Something has gone wrong, aborting!
", "stream"=>"stdout", "time"=>"2022-05-22T15:04:02.399655026Z"}]
----------- EOF -----------

----- MULTILINE FLUSH (stream_id=2148379754670323614) -----
[0] [1653231842.399683818, {"log"=>"    at com.myproject.module.MyProject.badMethod(MyProject.java:22)
", "stream"=>"stdout", "time"=>"2022-05-22T15:04:02.399683818Z"}]
----------- EOF -----------

----- MULTILINE FLUSH (stream_id=2148379754670323614) -----
[0] [1653231842.399687928, {"log"=>"    at com.myproject.module.MyProject.oneMoreMethod(MyProject.java:18)
", "stream"=>"stdout", "time"=>"2022-05-22T15:04:02.399687928Z"}]
----------- EOF -----------

----- MULTILINE FLUSH (stream_id=2148379754670323614) -----
[0] [1653231842.399690723, {"log"=>"    at com.myproject.module.MyProject.anotherMethod(MyProject.java:14)
", "stream"=>"stdout", "time"=>"2022-05-22T15:04:02.399690723Z"}]
----------- EOF -----------

----- MULTILINE FLUSH (stream_id=2148379754670323614) -----
[0] [1653231842.399693759, {"log"=>"    at com.myproject.module.MyProject.someMethod(MyProject.java:10)
", "stream"=>"stdout", "time"=>"2022-05-22T15:04:02.399693759Z"}]
----------- EOF -----------

----- MULTILINE FLUSH (stream_id=2148379754670323614) -----
[0] [1653231842.399696410, {"log"=>"    at com.myproject.module.MyProject.main(MyProject.java:6)
", "stream"=>"stdout", "time"=>"2022-05-22T15:04:02.39969641Z"}]
----------- EOF -----------

----- MULTILINE FLUSH (stream_id=2148379754670323614) -----
[0] [1653231842.399702217, {"log"=>"
", "stream"=>"stdout", "time"=>"2022-05-22T15:04:02.399702217Z"}]
----------- EOF -----------
[2022/05/22 15:58:47] [debug] [out flush] cb_destroy coro_id=0
[2022/05/22 15:58:47] [debug] [out flush] cb_destroy coro_id=1
{"log":"Dec 14 06:41:08 Exception in thread \\\"main\\\" java.lang.RuntimeException: Something has gone wrong, aborting!\n","stream":"stdout","time":"2022-05-22T15:04:02.399655026Z"}
{"log":"    at com.myproject.module.MyProject.badMethod(MyProject.java:22)\n","stream":"stdout","time":"2022-05-22T15:04:02.399683818Z"}
{"log":"    at com.myproject.module.MyProject.oneMoreMethod(MyProject.java:18)\n","stream":"stdout","time":"2022-05-22T15:04:02.399687928Z"}
{"log":"    at com.myproject.module.MyProject.anotherMethod(MyProject.java:14)\n","stream":"stdout","time":"2022-05-22T15:04:02.399690723Z"}
{"log":"    at com.myproject.module.MyProject.someMethod(MyProject.java:10)\n","stream":"stdout","time":"2022-05-22T15:04:02.399693759Z"}
{"log":"    at com.myproject.module.MyProject.main(MyProject.java:6)\n","stream":"stdout","time":"2022-05-22T15:04:02.39969641Z"}
{"log":"\n","stream":"stdout","time":"2022-05-22T15:04:02.399702217Z"}
[2022/05/22 15:58:47] [debug] [task] destroy task=0x7f9b6483d620 (task_id=0)
[2022/05/22 15:58:47] [debug] [task] destroy task=0x7f9b6483d690 (task_id=1)
^C[2022/05/22 15:58:48] [engine] caught signal (SIGINT)
[2022/05/22 15:58:48] [ info] [input] pausing container_logs
[2022/05/22 15:58:48] [ warn] [engine] service will shutdown in max 5 seconds
[2022/05/22 15:58:49] [ info] [engine] service has stopped (0 pending tasks)
[2022/05/22 15:58:49] [debug] [input:tail:container_logs] inode=158339697 removing file name /var/log/containers/pod1.log
[2022/05/22 15:58:49] [ info] [input:tail:container_logs] inotify_fs_remove(): inode=158339697 watch_fd=1
[2022/05/22 15:58:49] [debug] [input:tail:container_logs] inode=158339698 removing file name /var/log/containers/pod2.log
[2022/05/22 15:58:49] [ info] [input:tail:container_logs] inotify_fs_remove(): inode=158339698 watch_fd=2
[2022/05/22 15:58:49] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2022/05/22 15:58:49] [ info] [output:stdout:stdout.0] thread worker #0 stopped

As we can the first log os parsed correctly

{"log":"Dec 14 06:41:08 Exception in thread \\\"main\\\" java.lang.RuntimeException: Something has gone wrong, aborting!\n    at com.myproject.module.MyProject.badMethod(MyProject.java:22)\n    at com.myproject.module.MyProject.oneMoreMethod(MyProject.java:18)\n    at com.myproject.module.MyProject.anotherMethod(MyProject.java:14)\n    at com.myproject.module.MyProject.someMethod(MyProject.java:10)\n    at com.myproject.module.MyProject.main(MyProject.java:6)\n","stream":"stdout","time":"2022-05-22T15:02:37.349634795Z"}

The second is not

{"log":"Dec 14 06:41:08 Exception in thread \\\"main\\\" java.lang.RuntimeException: Something has gone wrong, aborting!\n","stream":"stdout","time":"2022-05-22T15:04:02.399655026Z"}
{"log":"    at com.myproject.module.MyProject.badMethod(MyProject.java:22)\n","stream":"stdout","time":"2022-05-22T15:04:02.399683818Z"}
{"log":"    at com.myproject.module.MyProject.oneMoreMethod(MyProject.java:18)\n","stream":"stdout","time":"2022-05-22T15:04:02.399687928Z"}
{"log":"    at com.myproject.module.MyProject.anotherMethod(MyProject.java:14)\n","stream":"stdout","time":"2022-05-22T15:04:02.399690723Z"}
{"log":"    at com.myproject.module.MyProject.someMethod(MyProject.java:10)\n","stream":"stdout","time":"2022-05-22T15:04:02.399693759Z"}
{"log":"    at com.myproject.module.MyProject.main(MyProject.java:6)\n","stream":"stdout","time":"2022-05-22T15:04:02.39969641Z"}

[chenlingmin]

I had the same problem as you
#5245

[robert-heinzmann-logmein]

Hello there, wanted to follow up. I re-ran the reproducer with the 1.9.6 version of fluent bit and I do see this works now

## ... pod1 logs ...
----- MULTILINE FLUSH (stream_id=6284141986831512414) -----
[0] [1653231757.349634795, {"log"=>"Dec 14 06:41:08 Exception in thread \"main\" java.lang.RuntimeException: Something has gone wrong, aborting!
    at com.myproject.module.MyProject.badMethod(MyProject.java:22)
    at com.myproject.module.MyProject.oneMoreMethod(MyProject.java:18)
    at com.myproject.module.MyProject.anotherMethod(MyProject.java:14)
    at com.myproject.module.MyProject.someMethod(MyProject.java:10)
    at com.myproject.module.MyProject.main(MyProject.java:6)
", "stream"=>"stdout", "time"=>"2022-05-22T15:02:37.349634795Z"}]
----------- EOF -----------
## ... pod2 logs ...
----- MULTILINE FLUSH (stream_id=2148379754670323614) -----
[0] [1653231842.399655026, {"log"=>"Dec 14 06:41:08 Exception in thread \"main\" java.lang.RuntimeException: Something has gone wrong, aborting!
    at com.myproject.module.MyProject.badMethod(MyProject.java:22)
    at com.myproject.module.MyProject.oneMoreMethod(MyProject.java:18)
    at com.myproject.module.MyProject.anotherMethod(MyProject.java:14)
    at com.myproject.module.MyProject.someMethod(MyProject.java:10)
    at com.myproject.module.MyProject.main(MyProject.java:6)
", "stream"=>"stdout", "time"=>"2022-05-22T15:04:02.399655026Z"}]
----------- EOF -----------

🎉

[github-actions]

This issue is stale because it has been open 90 days with no activity. Remove stale label or comment or this will be closed in 5 days. Maintainers can add the exempt-stale label.

[github-actions]

This issue was closed because it has been stalled for 5 days with no activity.