tests: internal: add multiline regression coverage

Add internal multiline regression coverage for the known API-level defects that are not a good fit for the Python integration suite.

The new tests cover the first-group-only flush behavior, truncation dropping the overflow line, and the empty context crash paths using the existing tests/internal multiline harness.

Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>

Author: edsiper

tests/internal/multiline.c

@@ -5,9 +5,15 @@
 #include <fluent-bit/flb_parser.h>
 #include <fluent-bit/flb_mem.h>
 #include <fluent-bit/multiline/flb_ml.h>
+#include <fluent-bit/multiline/flb_ml_group.h>
 #include <fluent-bit/multiline/flb_ml_rule.h>
 #include <fluent-bit/multiline/flb_ml_parser.h>
 
+#ifndef FLB_SYSTEM_WINDOWS
+#include <sys/wait.h>
+#include <unistd.h>
+#endif
+
 #include "flb_tests_internal.h"
 
 struct record_check {
@@ -20,6 +26,11 @@ struct expected_result {
     struct record_check *out_records;
 };
 
+struct captured_logs {
+    int flushes;
+    char logs[8][256];
+};
+
 /* Docker */
 struct record_check docker_input[] = {
   {"{\"log\": \"aa\\n\", \"stream\": \"stdout\", \"time\": \"2021-02-01T16:45:03.01231z\"}"},
@@ -465,6 +476,139 @@ static int flush_callback(struct flb_ml_parser *parser,
     return 0;
 }
 
+static int capture_logs_callback(struct flb_ml_parser *parser,
+                                 struct flb_ml_stream *mst,
+                                 void *data, char *buf_data, size_t buf_size)
+{
+    int i;
+    int ret;
+    int len;
+    size_t off = 0;
+    msgpack_unpacked result;
+    msgpack_object *map;
+    msgpack_object key;
+    msgpack_object val;
+    struct flb_time tm;
+    struct captured_logs *captured = data;
+
+    if (!captured || captured->flushes >= 8) {
+        return 0;
+    }
+
+    msgpack_unpacked_init(&result);
+    ret = msgpack_unpack_next(&result, buf_data, buf_size, &off);
+    TEST_CHECK(ret == MSGPACK_UNPACK_SUCCESS);
+
+    flb_time_pop_from_msgpack(&tm, &result, &map);
+    TEST_CHECK(flb_time_to_nanosec(&tm) != 0L);
+
+    for (i = 0; i < map->via.map.size; i++) {
+        key = map->via.map.ptr[i].key;
+        val = map->via.map.ptr[i].val;
+
+        if (key.type != MSGPACK_OBJECT_STR || val.type != MSGPACK_OBJECT_STR) {
+            continue;
+        }
+
+        if (key.via.str.size == 3 &&
+            strncmp(key.via.str.ptr, "log", 3) == 0) {
+            len = val.via.str.size;
+            if (len > 255) {
+                len = 255;
+            }
+
+            memcpy(captured->logs[captured->flushes], val.via.str.ptr, len);
+            captured->logs[captured->flushes][len] = '\0';
+            break;
+        }
+    }
+
+    captured->flushes++;
+    msgpack_unpacked_destroy(&result);
+    return 0;
+}
+
+static int log_matches(char *actual, const char *expected)
+{
+    size_t actual_len;
+    size_t expected_len;
+
+    actual_len = strlen(actual);
+    expected_len = strlen(expected);
+
+    if (strcmp(actual, expected) == 0) {
+        return FLB_TRUE;
+    }
+
+    if (actual_len == expected_len + 1 &&
+        memcmp(actual, expected, expected_len) == 0 &&
+        actual[actual_len - 1] == '\n') {
+        return FLB_TRUE;
+    }
+
+    return FLB_FALSE;
+}
+
+static struct flb_ml_parser *create_regex_test_parser(struct flb_config *config,
+                                                      const char *name,
+                                                      const char *pattern)
+{
+    int ret;
+    struct flb_ml_parser *mlp;
+    struct flb_ml_parser_params params;
+
+    params = flb_ml_parser_params_default(name);
+    params.type = FLB_ML_REGEX;
+    params.negate = FLB_FALSE;
+    params.flush_ms = 1000;
+
+    mlp = flb_ml_parser_create_params(config, &params);
+    TEST_CHECK(mlp != NULL);
+
+    ret = flb_ml_rule_create(mlp, "start_state", (char *) pattern, "cont", NULL);
+    TEST_CHECK(ret == 0);
+    ret = flb_ml_rule_create(mlp, "cont", (char *) pattern, "cont", NULL);
+    TEST_CHECK(ret == 0);
+    ret = flb_ml_parser_init(mlp);
+    TEST_CHECK(ret == 0);
+
+    return mlp;
+}
+
+#ifndef FLB_SYSTEM_WINDOWS
+static void assert_child_crashes(void (*fn)(void))
+{
+    pid_t pid;
+    int wstatus;
+
+    pid = fork();
+    if (pid == 0) {
+        fn();
+        exit(0);
+    }
+    else if (pid > 0) {
+        if (waitpid(pid, &wstatus, 0) == -1) {
+            TEST_CHECK(0);
+            TEST_MSG("waitpid failed");
+            return;
+        }
+
+        TEST_CHECK(WIFSIGNALED(wstatus));
+        if (!WIFSIGNALED(wstatus)) {
+            TEST_MSG("expected child crash, got status %d", wstatus);
+            return;
+        }
+
+        TEST_CHECK(WTERMSIG(wstatus) == SIGSEGV);
+        TEST_MSG("expected SIGSEGV, got signal %d", WTERMSIG(wstatus));
+    }
+    else {
+        TEST_CHECK(0);
+        TEST_MSG("fork() failed");
+    }
+}
+#endif
+
 static void test_parser_docker()
 {
     int i;
@@ -1643,6 +1787,158 @@ static void test_buffer_limit_disabled()
     flb_config_exit(config);
 }
 
+static void test_known_bug_multi_group_flush_only_first_group()
+{
+    int ret;
+    int saw_a = FLB_FALSE;
+    int saw_b = FLB_FALSE;
+    int i;
+    uint64_t stream_id;
+    struct flb_config *config;
+    struct flb_ml *ml;
+    struct flb_time tm;
+    struct captured_logs captured = {0};
+
+    config = flb_config_init();
+    ml = flb_ml_create(config, "known-bug-multi-group");
+    TEST_CHECK(ml != NULL);
+
+    create_regex_test_parser(config, "bug_group_a", "/^A/");
+    create_regex_test_parser(config, "bug_group_b", "/^B/");
+
+    TEST_CHECK(flb_ml_parser_instance_create(ml, "bug_group_a") != NULL);
+    TEST_CHECK(flb_ml_group_create(ml) != NULL);
+    TEST_CHECK(flb_ml_parser_instance_create(ml, "bug_group_b") != NULL);
+
+    ret = flb_ml_stream_create(ml, "bug-stream", -1,
+                               capture_logs_callback, &captured, &stream_id);
+    TEST_CHECK(ret == 0);
+
+    flb_time_get(&tm);
+    ret = flb_ml_append_text(ml, stream_id, &tm, "A buffered", 10);
+    TEST_CHECK(ret == FLB_MULTILINE_OK);
+
+    flb_time_get(&tm);
+    ret = flb_ml_append_text(ml, stream_id, &tm, "B buffered", 10);
+    TEST_CHECK(ret == FLB_MULTILINE_OK);
+
+    flb_ml_flush_pending_now(ml);
+
+    for (i = 0; i < captured.flushes; i++) {
+        if (log_matches(captured.logs[i], "A buffered")) {
+            saw_a = FLB_TRUE;
+        }
+        if (log_matches(captured.logs[i], "B buffered")) {
+            saw_b = FLB_TRUE;
+        }
+    }
+
+    TEST_CHECK(captured.flushes == 1);
+    TEST_CHECK(saw_a == FLB_TRUE);
+    TEST_CHECK(saw_b == FLB_FALSE);
+
+    flb_ml_destroy(ml);
+    flb_config_exit(config);
+}
+
+static void test_known_bug_truncation_drops_overflow_line()
+{
+    int ret;
+    int i;
+    int saw_overflow_line = FLB_FALSE;
+    uint64_t stream_id;
+    struct flb_config *config;
+    struct flb_ml *ml;
+    struct flb_time tm;
+    struct captured_logs captured = {0};
+
+    config = flb_config_init();
+    if (config->multiline_buffer_limit) {
+        flb_free(config->multiline_buffer_limit);
+    }
+    config->multiline_buffer_limit = flb_strdup("8");
+
+    ml = flb_ml_create(config, "known-bug-truncation");
+    TEST_CHECK(ml != NULL);
+
+    create_regex_test_parser(config, "bug_truncation", "/./");
+    TEST_CHECK(flb_ml_parser_instance_create(ml, "bug_truncation") != NULL);
+
+    ret = flb_ml_stream_create(ml, "bug-stream", -1,
+                               capture_logs_callback, &captured, &stream_id);
+    TEST_CHECK(ret == 0);
+
+    flb_time_get(&tm);
+    ret = flb_ml_append_text(ml, stream_id, &tm, "AAAAA", 5);
+    TEST_CHECK(ret == FLB_MULTILINE_OK);
+
+    flb_time_get(&tm);
+    ret = flb_ml_append_text(ml, stream_id, &tm, "BBBBB", 5);
+    TEST_CHECK(ret == FLB_MULTILINE_TRUNCATED);
+
+    flb_ml_flush_pending_now(ml);
+
+    for (i = 0; i < captured.flushes; i++) {
+        if (log_matches(captured.logs[i], "BBBBB")) {
+            saw_overflow_line = FLB_TRUE;
+        }
+    }
+
+    TEST_CHECK(captured.flushes == 1);
+    TEST_CHECK(saw_overflow_line == FLB_FALSE);
+
+    flb_ml_destroy(ml);
+    flb_config_exit(config);
+}
+
+#ifndef FLB_SYSTEM_WINDOWS
+static void crash_empty_context_flush()
+{
+    struct flb_config *config;
+    struct flb_ml *ml;
+
+    config = flb_config_init();
+    ml = flb_ml_create(config, "known-bug-empty-flush");
+    flb_ml_flush_pending_now(ml);
+    flb_ml_destroy(ml);
+    flb_config_exit(config);
+}
+
+static void crash_empty_context_append()
+{
+    struct flb_config *config;
+    struct flb_ml *ml;
+    struct flb_time tm;
+
+    config = flb_config_init();
+    ml = flb_ml_create(config, "known-bug-empty-append");
+    flb_time_get(&tm);
+    flb_ml_append_text(ml, 1, &tm, "hello", 5);
+    flb_ml_destroy(ml);
+    flb_config_exit(config);
+}
+
+static void test_known_bug_empty_context_flush_crashes()
+{
+    assert_child_crashes(crash_empty_context_flush);
+}
+
+static void test_known_bug_empty_context_append_crashes()
+{
+    assert_child_crashes(crash_empty_context_append);
+}
+#else
+static void test_known_bug_empty_context_flush_crashes()
+{
+    TEST_MSG("skipped on Windows");
+}
+
+static void test_known_bug_empty_context_append_crashes()
+{
+    TEST_MSG("skipped on Windows");
+}
+#endif
+
 TEST_LIST = {
     /* Normal features tests */
     { "parser_docker",  test_parser_docker},
@@ -1657,6 +1953,14 @@ TEST_LIST = {
     { "endswith",       test_endswith},
     { "buffer_limit_truncation", test_buffer_limit_truncation},
     { "buffer_limit_disabled", test_buffer_limit_disabled},
+    { "known_bug_multi_group_flush_only_first_group",
+      test_known_bug_multi_group_flush_only_first_group},
+    { "known_bug_truncation_drops_overflow_line",
+      test_known_bug_truncation_drops_overflow_line},
+    { "known_bug_empty_context_flush_crashes",
+      test_known_bug_empty_context_flush_crashes},
+    { "known_bug_empty_context_append_crashes",
+      test_known_bug_empty_context_append_crashes},
 
     /* Issues reported on Github */
     { "issue_3817_1"  , test_issue_3817_1},