multiline: add built-in json parser for JSON multiline objects

include/fluent-bit/multiline/flb_ml_parser.h

@@ -89,5 +89,6 @@ struct flb_ml_parser *flb_ml_parser_java(struct flb_config *config, char *key);
 struct flb_ml_parser *flb_ml_parser_go(struct flb_config *config, char *key);
 struct flb_ml_parser *flb_ml_parser_ruby(struct flb_config *config, char *key);
 struct flb_ml_parser *flb_ml_parser_python(struct flb_config *config, char *key);
+struct flb_ml_parser *flb_ml_parser_json(struct flb_config *config, char *key);
 
 #endif

plugins/filter_multiline/ml.c

@@ -1023,7 +1023,7 @@ static struct flb_config_map config_map[] = {
     {
      FLB_CONFIG_MAP_CLIST, "multiline.parser", NULL,
      FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct ml_ctx, multiline_parsers),
-     "specify one or multiple multiline parsers: docker, cri, go, java, etc."
+     "specify one or multiple multiline parsers: docker, cri, go, java, json, etc."
     },
 
     {

plugins/in_tail/tail.c

@@ -826,7 +826,7 @@ static struct flb_config_map config_map[] = {
     {
      FLB_CONFIG_MAP_CLIST, "multiline.parser", NULL,
      FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct flb_tail_config, multiline_parsers),
-     "specify one or multiple multiline parsers: docker, cri, go, java, etc."
+     "specify one or multiple multiline parsers: docker, cri, go, java, json, etc."
     },
 #endif
 

src/multiline/CMakeLists.txt

@@ -6,6 +6,7 @@ set(src_multiline
   multiline/flb_ml_parser_java.c
   multiline/flb_ml_parser_go.c
   multiline/flb_ml_parser_ruby.c
+  multiline/flb_ml_parser_json.c
   # core
   multiline/flb_ml_stream.c
   multiline/flb_ml_parser.c

src/multiline/flb_ml_parser.c

@@ -188,6 +188,13 @@ int flb_ml_parser_builtin_create(struct flb_config *config)
         goto error;
     }
 
+    /* JSON */
+    mlp = flb_ml_parser_json(config, NULL);
+    if (!mlp) {
+        flb_error("[multiline] could not init 'json' built-in parser");
+        goto error;
+    }
+
     ret = 0;
     return ret;
 

src/multiline/flb_ml_parser_json.c

@@ -0,0 +1,95 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2026 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/multiline/flb_ml.h>
+#include <fluent-bit/multiline/flb_ml_rule.h>
+#include <fluent-bit/multiline/flb_ml_parser.h>
+
+#define rule flb_ml_rule_create
+
+static void rule_error(struct flb_ml_parser *ml_parser)
+{
+    int id;
+
+    id = mk_list_size(&ml_parser->regex_rules);
+    flb_error("[multiline: json] rule #%i could not be created", id);
+    flb_ml_parser_destroy(ml_parser);
+}
+
+/*
+ * Built-in multiline mode for pretty-printed JSON objects.
+ *
+ * Tail reads line-by-line, so a formatted JSON object is not valid JSON per
+ * line. This parser groups lines from an opening '{' until the next object
+ * starts or flush_timeout expires.
+ *
+ * Do not attach a JSON parser here: per-line JSON parsing fails on partial
+ * lines (see ml_append_try_parser_type_text). Use filter_parser on the
+ * assembled 'log' field after grouping.
+ */
+struct flb_ml_parser *flb_ml_parser_json(struct flb_config *config, char *key)
+{
+    int ret;
+    struct flb_ml_parser *mlp;
+
+    mlp = flb_ml_parser_create(config,               /* Fluent Bit context */
+                               "json",               /* name      */
+                               FLB_ML_REGEX,         /* type      */
+                               NULL,                 /* match_str */
+                               FLB_FALSE,            /* negate    */
+                               FLB_ML_FLUSH_TIMEOUT, /* flush_ms  */
+                               key,                  /* key_content */
+                               NULL,                 /* key_group   */
+                               NULL,                 /* key_pattern */
+                               NULL,                 /* parser ctx  */
+                               NULL);                /* parser name */
+
+    if (!mlp) {
+        flb_error("[multiline] could not create 'json mode'");
+        return NULL;
+    }
+
+    ret = rule(mlp,
+               "start_state",
+               "/^\\{.*/",
+               "cont", NULL);
+    if (ret != 0) {
+        rule_error(mlp);
+        return NULL;
+    }
+
+    ret = rule(mlp,
+               "cont",
+               "/^([^\\S\\r\\n].*|})$/",
+               "cont", NULL);
+    if (ret != 0) {
+        rule_error(mlp);
+        return NULL;
+    }
+
+    ret = flb_ml_parser_init(mlp);
+    if (ret != 0) {
+        flb_error("[multiline: json] error on mapping rules");
+        flb_ml_parser_destroy(mlp);
+        return NULL;
+    }
+
+    return mlp;
+}

tests/internal/multiline.c

@@ -364,6 +364,60 @@ struct record_check go_output[] = {
     {"one more line, no multiline\n"}
 };
 
+/* JSON (pretty-printed and single-line objects) */
+struct record_check json_input[] = {
+    {"{\"id\":101,\"level\":\"info\",\"msg\":\"single-line record A\"}"},
+    {"{"},
+    {"  \"id\": 102,"},
+    {"  \"level\": \"warn\","},
+    {"  \"msg\": \"multiline record B\""},
+    {"}"},
+    {"{\"id\":103,\"level\":\"info\",\"msg\":\"single-line record C\"}"},
+    {"{"},
+    {"  \"id\": 104,"},
+    {"  \"level\": \"error\","},
+    {"  \"msg\": \"multiline record D\""},
+    {"}"},
+    /* invalid boundary: unindented line after opening brace */
+    {"{"},
+    {"\"bad\": true"},
+    /* invalid boundary: closing brace with trailing content */
+    {"{"},
+    {"  \"ok\": 1"},
+    {"} trailing"},
+    /* valid single-line object after invalid boundaries */
+    {"{\"id\":105,\"level\":\"info\",\"msg\":\"after boundaries\"}"},
+};
+
+struct record_check json_output[] = {
+    {"{\"id\":101,\"level\":\"info\",\"msg\":\"single-line record A\"}\n"},
+    {
+        "{\n"
+        "  \"id\": 102,\n"
+        "  \"level\": \"warn\",\n"
+        "  \"msg\": \"multiline record B\"\n"
+        "}\n"
+    },
+    {"{\"id\":103,\"level\":\"info\",\"msg\":\"single-line record C\"}\n"},
+    {
+        "{\n"
+        "  \"id\": 104,\n"
+        "  \"level\": \"error\",\n"
+        "  \"msg\": \"multiline record D\"\n"
+        "}\n"
+    },
+    /* unindented continuation must not merge into buffered '{' */
+    {"{\n"},
+    {"\"bad\": true\n"},
+    /* trailing content on '}' line must not merge into partial object */
+    {
+        "{\n"
+        "  \"ok\": 1\n"
+    },
+    {"} trailing\n"},
+    {"{\"id\":105,\"level\":\"info\",\"msg\":\"after boundaries\"}\n"},
+};
+
 /*
  * Issue 3817 (case: 1)
  * --------------------
@@ -1202,6 +1256,55 @@ static void test_parser_go()
     flb_config_exit(config);
 }
 
+static void test_parser_json()
+{
+    int i;
+    int len;
+    int ret;
+    int entries;
+    uint64_t stream_id = 0;
+    struct record_check *r;
+    struct flb_config *config;
+    struct flb_time tm;
+    struct flb_ml *ml;
+    struct flb_ml_parser_ins *mlp_i;
+    struct expected_result res = {0};
+
+    res.key = "log";
+    res.out_records = json_output;
+
+    config = flb_config_init();
+
+    ml = flb_ml_create(config, "json-test");
+    TEST_CHECK(ml != NULL);
+
+    mlp_i = flb_ml_parser_instance_create(ml, "json");
+    TEST_CHECK(mlp_i != NULL);
+
+    ret = flb_ml_stream_create(ml, "json", -1, flush_callback, (void *) &res,
+                               &stream_id);
+    TEST_CHECK(ret == 0);
+
+    entries = sizeof(json_input) / sizeof(struct record_check);
+    for (i = 0; i < entries; i++) {
+        r = &json_input[i];
+        len = strlen(r->buf);
+
+        flb_time_get(&tm);
+        flb_ml_append_text(ml, stream_id, &tm, r->buf, len);
+    }
+
+    flb_ml_flush_pending_now(ml);
+
+    if (ml) {
+        flb_ml_destroy(ml);
+    }
+
+    TEST_CHECK(res.current_record == (sizeof(json_output) / sizeof(struct record_check)));
+
+    flb_config_exit(config);
+}
+
 static int flush_callback_to_buf(struct flb_ml_parser *parser,
                                  struct flb_ml_stream *mst,
                                  void *data, char *buf_data, size_t buf_size)
@@ -2129,6 +2232,7 @@ TEST_LIST = {
     { "parser_ruby",    test_parser_ruby},
     { "parser_elastic", test_parser_elastic},
     { "parser_go",      test_parser_go},
+    { "parser_json",    test_parser_json},
     { "container_mix",  test_container_mix},
     { "endswith",       test_endswith},
     { "buffer_limit_truncation", test_buffer_limit_truncation},