in_tail: flush multiline before exit on EOF

[nokute78]

Fixes #8487

This patch is to flush multiline data before exit on EOF.

---

Enter [N/A] in the box, if an item is not applicable to your change.

Testing
Before we can approve your change; please submit the following in a comment:

• Example configuration file for the change
• Debug log output from testing the change

• Attached Valgrind output that shows no leaks or memory corruption was found

If this is a change to packaging of containers or native binaries then please confirm it works for all targets.

• [N/A] Run local packaging test showing all targets (including any new ones) build.
• [N/A] Set ok-package-test label to test for all targets (requires maintainer to do).

Documentation

• [N/A] Documentation required for this feature

Backporting

• [N/A] Backport to latest stable release.

Configuration

test.conf:

[SERVICE]
    Flush             1
    Daemon            Off
    Parsers_File      parsers.conf

[INPUT]
    Name              tail
    Alias             test
    Path              test.log
    Path_Key          _path
    Multiline         On
    Multiline_Flush   1
    Parser_Firstline  test-firstline
    Tag               test
    Skip_Empty_Lines  On
    Exit_On_EOF       On
    Read_From_Head    On

[FILTER]
    Name              parser
    Alias             parser:test
    Match             test
    Parser            test
    Key_Name          raw
    Preserve_Key      On
    Reserve_Data      On

[OUTPUT]
    Name              stdout
    Format            json_lines

parsers.conf:

[PARSER]
    Name         test-firstline
    Format       regex
    Regex        ^(?<raw>\[\d\d\d\d-\d\d-\d\d .+)$

[PARSER]
    Name         test
    Format       regex
    Regex        ^\[(?<time>\S+ \S+)\]\s+(?<priority>\w+) +(?<message>[^\n]*)(\n(?<trace>[\d\D]+))?$
    Time_Key     time
    Time_Format  %Y-%m-%d %H:%M:%S,%L
    Time_Keep    Off

test.log:

[2019-10-14 08:57:57,558] ERROR 
[2019-09-24 12:31:55,814] ERROR Lost connection
[2018-10-16 21:27:33,965] ERROR 
Traceback (most recent call last):
  File "/usr/bin/test", line 296, in connect
    connection.start()
ConnectFailedException
[2020-07-13 23:27:24,263] INFO Disconnected from somewhere
[2020-07-13 23:27:29,348] INFO Successfully connected to somewhere

Valgrind output

$ valgrind --leak-check=full ../bin/fluent-bit -c test.conf 
==62069== Memcheck, a memory error detector
==62069== Copyright (C) 2002-2017, and GNU GPL'd, by Julian Seward et al.
==62069== Using Valgrind-3.18.1 and LibVEX; rerun with -h for copyright info
==62069== Command: ../bin/fluent-bit -c test.conf
==62069== 
Fluent Bit v3.0.0
* Copyright (C) 2015-2024 The Fluent Bit Authors
* Fluent Bit is a CNCF sub-project under the umbrella of Fluentd
* https://fluentbit.io

___________.__                        __    __________.__  __          ________  
\_   _____/|  |  __ __   ____   _____/  |_  \______   \__|/  |_  ___  _\_____  \ 
 |    __)  |  | |  |  \_/ __ \ /    \   __\  |    |  _/  \   __\ \  \/ / _(__  < 
 |     \   |  |_|  |  /\  ___/|   |  \  |    |    |   \  ||  |    \   / /       \
 \___  /   |____/____/  \___  >___|  /__|    |______  /__||__|     \_/ /______  /
     \/                     \/     \/               \/                        \/ 

[2024/03/03 08:14:55] [ info] [fluent bit] version=3.0.0, commit=9d9ac68a2b, pid=62069
[2024/03/03 08:14:55] [ info] [storage] ver=1.2.0, type=memory, sync=normal, checksum=off, max_chunks_up=128
[2024/03/03 08:14:55] [ info] [cmetrics] version=0.7.0
[2024/03/03 08:14:55] [ info] [ctraces ] version=0.4.0
[2024/03/03 08:14:55] [ info] [input:tail:test] initializing
[2024/03/03 08:14:55] [ info] [input:tail:test] storage_strategy='memory' (memory only)
[2024/03/03 08:14:55] [ info] [output:stdout:stdout.0] worker #0 started
[2024/03/03 08:14:55] [ info] [sp] stream processor started
[2024/03/03 08:14:55] [ info] [input:tail:test] inode=4351673 file=test.log ended, stop
[2024/03/03 08:14:55] [ info] [input:tail:test] inotify_fs_add(): inode=4351673 watch_fd=1 name=test.log
[2024/03/03 08:14:56] [ warn] [engine] service will shutdown in max 5 seconds
[2024/03/03 08:14:56] [ info] [input] pausing test
{"date":1571043477.558,"priority":"ERROR","_path":"test.log","raw":"[2019-10-14 08:57:57,558] ERROR "}
{"date":1569328315.814,"priority":"ERROR","message":"Lost connection","_path":"test.log","raw":"[2019-09-24 12:31:55,814] ERROR Lost connection"}
{"date":1539725253.965,"priority":"ERROR","trace":"Traceback (most recent call last):\n  File \"/usr/bin/test\", line 296, in connect\n    connection.start()\nConnectFailedException","_path":"test.log","raw":"[2018-10-16 21:27:33,965] ERROR \nTraceback (most recent call last):\n  File \"/usr/bin/test\", line 296, in connect\n    connection.start()\nConnectFailedException"}
{"date":1594682844.263,"priority":"INFO","message":"Disconnected from somewhere","_path":"test.log","raw":"[2020-07-13 23:27:24,263] INFO Disconnected from somewhere"}
{"date":1594682849.348,"priority":"INFO","message":"Successfully connected to somewhere","_path":"test.log","raw":"[2020-07-13 23:27:29,348] INFO Successfully connected to somewhere"}
[2024/03/03 08:14:56] [ info] [input:tail:test] flushing pending multiline data...
[2024/03/03 08:14:56] [ info] [engine] service has stopped (0 pending tasks)
[2024/03/03 08:14:56] [ info] [input] pausing test
[2024/03/03 08:14:56] [ info] [input:tail:test] flushing pending multiline data...
[2024/03/03 08:14:56] [ info] [output:stdout:stdout.0] thread worker #0 stopping...
[2024/03/03 08:14:56] [ info] [output:stdout:stdout.0] thread worker #0 stopped
[2024/03/03 08:14:56] [ info] [input:tail:test] inotify_fs_remove(): inode=4351673 watch_fd=1
==62069== 
==62069== HEAP SUMMARY:
==62069==     in use at exit: 0 bytes in 0 blocks
==62069==   total heap usage: 2,557 allocs, 2,557 frees, 1,509,371 bytes allocated
==62069== 
==62069== All heap blocks were freed -- no leaks are possible
==62069== 
==62069== For lists of detected and suppressed errors, rerun with: -s
==62069== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)

---

Fluent Bit is licensed under Apache 2.0, by submitting this pull request I understand that this code will be released under the terms of that license.