in_tail: flush pending multiline records on shutdown

plugins/in_tail/tail.c

@@ -569,6 +569,19 @@ static void in_tail_pause(void *data, struct flb_config *config)
 
     /* Pause file system backend handlers */
     flb_tail_fs_pause(ctx);
+
+#ifdef FLB_HAVE_PARSER
+    if (config->is_shutting_down == FLB_TRUE && ctx->ml_ctx != NULL &&
+        ctx->multiline_shutdown_flushed == FLB_FALSE) {
+        /*
+         * Flush while processors, routes and outputs are still alive. Mark
+         * this first: appending the records can re-enter pause on a buffer
+         * limit, and the engine also pauses inputs again during teardown.
+         */
+        ctx->multiline_shutdown_flushed = FLB_TRUE;
+        flb_ml_flush_pending_now(ctx->ml_ctx);
+    }
+#endif
 }
 
 static void in_tail_resume(void *data, struct flb_config *config)
@@ -890,5 +903,5 @@ struct flb_input_plugin in_tail_plugin = {
     .cb_resume    = in_tail_resume,
     .cb_exit      = in_tail_exit,
     .config_map   = config_map,
-    .flags        = 0
+    .flags        = FLB_INPUT_SHUTDOWN_FLUSH
 };

plugins/in_tail/tail_config.h

@@ -158,6 +158,7 @@ struct flb_tail_config {
     /* Multiline core engine */
     struct flb_ml *ml_ctx;
     struct mk_list *multiline_parsers;
+    int multiline_shutdown_flushed;
 
     uint64_t files_static_count;   /* number of items in the static file list */
     struct mk_list files_static;

tests/integration/scenarios/in_tail/config/parsers_tail_it.conf

@@ -30,3 +30,24 @@
     Name   docker_multiline
     Format regex
     Regex  (?<log>^{"log":"\d{4}-\d{2}-\d{2}.*)
+
+# Used by the tail shutdown scenario: the multiline engine flushes buffered
+# groups on every timer tick, so this CRI compatible parser pins the flush
+# timeout to 60 seconds to keep the pending fragment buffered until shutdown.
+[PARSER]
+    Name        cri_test
+    Format      regex
+    Regex       ^(?<time>[^ ]+) (?<stream>stdout|stderr) (?<tag>F|P) (?<log>.*)$
+    Time_Key    time
+    Time_Format %Y-%m-%dT%H:%M:%S.%L%z
+    Skip_Empty_Values false
+
+[MULTILINE_PARSER]
+    Name          cri_test_ml
+    Type          eq
+    Match_String  F
+    Flush_Timeout 60000
+    Key_Content   log
+    Key_Pattern   tag
+    Key_Group     stream
+    Parser        cri_test

tests/integration/scenarios/in_tail/config/tail_cri_shutdown.yaml

@@ -0,0 +1,30 @@
+service:
+  flush: ${TAIL_SHUTDOWN_FLUSH}
+  grace: 3
+  log_level: debug
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+  parsers_file: ${PARSERS_FILE_TEST}
+
+pipeline:
+  inputs:
+    - name: tail
+      tag: tail.integration
+      path: ${TAIL_TEST_PATH}
+      db: ${TAIL_TEST_DB}
+      read_from_head: true
+      refresh_interval: 1
+      multiline.parser: cri_test_ml
+      mem_buf_limit: ${TAIL_SHUTDOWN_MEM_LIMIT}
+      threaded: ${TAIL_SHUTDOWN_THREADED}
+      path_key: file
+      offset_key: offset
+
+  outputs:
+    - name: http
+      match: tail.integration
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}
+      uri: /data
+      format: json
+      json_date_key: false

tests/integration/scenarios/in_tail/tests/test_in_tail_001.py

@@ -15,6 +15,73 @@
 from utils.fluent_bit_manager import fluent_bit_input_supports_config_property
 from utils.test_service import FluentBitTestService
 
+
+@pytest.mark.skipif(
+    sys.platform == "win32",
+    reason="subprocess SIGTERM uses TerminateProcess on Windows; use the in_tail runtime shutdown test",
+)
+@pytest.mark.parametrize("mode", ["normal", "prepaused", "threaded"])
+@pytest.mark.parametrize("fragment", ["partial", "empty", "complete"])
+def test_in_tail_cri_shutdown(workspace, monkeypatch, mode, fragment):
+    log_file = workspace / "shutdown.log"
+    db_path = workspace / "shutdown.db"
+    log_file.touch()
+    monkeypatch.setenv("TAIL_SHUTDOWN_FLUSH", "60" if mode == "prepaused" else "0.05")
+    monkeypatch.setenv("TAIL_SHUTDOWN_MEM_LIMIT", "1K" if mode == "prepaused" else "10M")
+    monkeypatch.setenv("TAIL_SHUTDOWN_THREADED", "true" if mode == "threaded" else "false")
+    service = Service("tail_cri_shutdown.yaml", tail_path=log_file, db_path=db_path)
+    timestamp = "2026-10-23T01:00:00.000000000Z"
+    barrier = "shutdown-barrier-" + "x" * 2048
+    if fragment == "partial":
+        content = f"{timestamp} stdout P shutdown-one..\n{timestamp} stdout P two\n"
+        expected = ["shutdown-one..two", barrier]
+    elif fragment == "empty":
+        content = f"{timestamp} stdout P \n"
+        expected = ["", barrier]
+    else:
+        content = f"{timestamp} stdout F shutdown-complete\n"
+        expected = ["shutdown-complete", barrier]
+    content += f"{timestamp} stderr F {barrier}\n"
+
+    def consumed():
+        if not db_path.exists():
+            return False
+        try:
+            with sqlite3.connect(db_path) as connection:
+                row = connection.execute("SELECT offset FROM in_tail_files").fetchone()
+            return row is not None and row[0] == len(content.encode("utf-8"))
+        except sqlite3.OperationalError:
+            return False
+
+    process = None
+    try:
+        service.start()
+        process = service.flb.process
+        write_and_sync(log_file, content)
+        service.service.wait_for_condition(
+            consumed, timeout=10, interval=0.02, description="CRI input consumed before shutdown"
+        )
+        if mode == "prepaused":
+            service.service.wait_for_condition(
+                lambda: "paused (mem buf overlimit" in Path(service.flb.log_file).read_text(
+                    encoding="utf-8", errors="replace"
+                ),
+                timeout=5, interval=0.02, description="Tail paused before shutdown",
+            )
+        if fragment == "partial":
+            assert all(
+                record.get("log") != "shutdown-one..two"
+                for record in flatten_records(data_storage["payloads"])
+            )
+    finally:
+        service.stop()
+
+    assert process is not None and process.returncode == 0
+    records = flatten_records(data_storage["payloads"])
+    assert sorted(record["log"] for record in records) == sorted(expected)
+    assert all(record["file"] == str(log_file) for record in records)
+    assert all(isinstance(record["offset"], int) for record in records)
+
 skip_on_windows = pytest.mark.skipif(
     sys.platform == "win32",
     reason="scenario relies on POSIX rotation, symlink, permissions, or inode semantics",

tests/runtime/data/common/parsers_cri_shutdown.conf

@@ -0,0 +1,24 @@
+# Test-only parsers used by the in_tail shutdown regression test.
+#
+# The multiline engine flushes buffered groups on every timer tick and the
+# built-in "cri" parser uses the default 4 second timeout. The shutdown test
+# needs the pending fragment to stay buffered until the engine is stopped, so
+# this parser mirrors the CRI semantics with a 60 second flush timeout.
+
+[PARSER]
+    Name        cri_test
+    Format      regex
+    Regex       ^(?<time>[^ ]+) (?<stream>stdout|stderr) (?<tag>F|P) (?<log>.*)$
+    Time_Key    time
+    Time_Format %Y-%m-%dT%H:%M:%S.%L%z
+    Skip_Empty_Values false
+
+[MULTILINE_PARSER]
+    Name          cri_test_ml
+    Type          eq
+    Match_String  F
+    Flush_Timeout 60000
+    Key_Content   log
+    Key_Pattern   tag
+    Key_Group     stream
+    Parser        cri_test

tests/runtime/in_tail.c

@@ -1,4 +1,4 @@
-/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
 
 /*  Fluent Bit
  *  ==========
@@ -26,6 +26,8 @@ Approach for this tests is basing on filter_kubernetes tests
 #include <fluent-bit/flb_time.h>
 #include <fluent-bit/flb_pthread.h>
 #include <fluent-bit/flb_compat.h>
+#include <fluent-bit/flb_engine.h>
+#include <fluent-bit/flb_parser.h>
 #ifdef FLB_HAVE_UNICODE_ENCODER
 #include <fluent-bit/flb_unicode.h>
 #endif
@@ -3139,8 +3141,81 @@ void flb_test_db_compare_filename()
 }
 #endif /* FLB_HAVE_SQLDB */
 
+#ifdef FLB_HAVE_PARSER
+void flb_test_cri_pending_shutdown(void)
+{
+    struct flb_lib_out_cb cb;
+    struct test_tail_ctx *ctx;
+    char *paths[] = {"cri_pending_shutdown.log"};
+    char *messages =
+        "2026-10-23T01:00:00.000000000Z stdout P pending-one..\n"
+        "2026-10-23T01:00:00.000000000Z stdout P two\n"
+        "2026-10-23T01:00:00.000000000Z stderr F shutdown-barrier";
+    int ret;
+    int num;
+
+    clear_output_num();
+    cb.cb = cb_count_msgpack;
+    cb.data = NULL;
+    ctx = test_tail_ctx_create(&cb, paths, 1, FLB_TRUE);
+    if (!TEST_CHECK(ctx != NULL)) {
+        return;
+    }
+
+    /*
+     * Load a CRI compatible multiline parser with a long flush timeout so
+     * the pending fragment is still buffered when the engine is stopped.
+     */
+    ret = flb_parser_conf_file(DPATH_COMMON "/parsers_cri_shutdown.conf",
+                               ctx->flb->config);
+    TEST_CHECK(ret == 0);
+
+    /* Allow the shutdown flush to complete even near a timer tick. */
+    ret = flb_service_set(ctx->flb, "Grace", "3", NULL);
+    TEST_CHECK(ret == 0);
+    ret = flb_input_set(ctx->flb, ctx->i_ffd,
+                        "path", paths[0],
+                        "read_from_head", "true",
+                        "multiline.parser", "cri_test_ml", NULL);
+    TEST_CHECK(ret == 0);
+    ret = flb_output_set(ctx->flb, ctx->o_ffd, "match", "*", NULL);
+    TEST_CHECK(ret == 0);
+    ret = flb_start(ctx->flb);
+    TEST_CHECK(ret == 0);
+    ret = write_msg(ctx, messages, strlen(messages));
+    TEST_CHECK(ret > 0);
+
+    /* The stderr record proves ingestion while stdout is still pending. */
+    wait_num_with_timeout(5000, &num);
+    TEST_CHECK(num == 1);
+
+    /*
+     * Signal the shutdown separately and wait for the pending group to be
+     * flushed: on some platforms the engine worker is cancelled right after
+     * the notification, which can preempt the shutdown pause callback.
+     */
+    ret = flb_engine_exit(ctx->flb->config);
+    TEST_CHECK(ret >= 0);
+    wait_expected_num_with_timeout(5000, 2, &num);
+    TEST_CHECK(num == 2);
+
+    /* Keep the file present: unlinking it can flush the stream before stop. */
+    ret = flb_stop(ctx->flb);
+    TEST_CHECK(ret == 0);
+    TEST_CHECK(get_output_num() == 2);
+    flb_destroy(ctx->flb);
+    close(ctx->fds[0]);
+    unlink(paths[0]);
+    flb_free(ctx->fds);
+    flb_free(ctx);
+}
+#endif
+
 /* Test list */
 TEST_LIST = {
+#ifdef FLB_HAVE_PARSER
+    {"cri_pending_shutdown", flb_test_cri_pending_shutdown},
+#endif
     {"issue_3943", flb_test_in_tail_issue_3943},
     /* Properties */
     {"skip_long_lines", flb_test_in_tail_skip_long_lines},