From 537bd8d8a0e45a6a21fddb53fb9256ac016955db Mon Sep 17 00:00:00 2001
From: Jorge Niedbalski <jorge.niedbalski@chronosphere.io>
Date: Mon, 2 Dec 2024 14:07:33 +0100
Subject: [PATCH] filter_parser: fix reserve data and preserve key handling

Handle properly the 4 combinations for Reserve_Data and Preserve_Key:
- Off/Off: keep only parsed fields
- Off/On: parsed fields + original key
- On/Off: all fields except target key
- On/On: keep all fields

Signed-off-by: Jorge Niedbalski <jnr@metaklass.org>
---
 plugins/filter_parser/filter_parser.c | 163 ++++-----
 tests/runtime/filter_parser.c         | 462 ++++++++++++++++++++++----
 2 files changed, 477 insertions(+), 148 deletions(-)

diff --git a/plugins/filter_parser/filter_parser.c b/plugins/filter_parser/filter_parser.c
index 11bf71f7e53..64a504ef905 100644
--- a/plugins/filter_parser/filter_parser.c
+++ b/plugins/filter_parser/filter_parser.c
@@ -142,10 +142,6 @@ static int cb_parser_init(struct flb_filter_instance *f_ins,
                           struct flb_config *config,
                           void *data)
 {
-    (void) f_ins;
-    (void) config;
-    (void) data;
-
     struct filter_parser_ctx *ctx = NULL;
 
     /* Create context */
@@ -156,13 +152,12 @@ static int cb_parser_init(struct flb_filter_instance *f_ins,
     }
     ctx->ins = f_ins;
 
-    if ( configure(ctx, f_ins, config) < 0 ){
+    if (configure(ctx, f_ins, config) < 0) {
         flb_free(ctx);
         return -1;
     }
 
     flb_filter_set_context(f_ins, ctx);
-
     return 0;
 }
 
@@ -174,11 +169,9 @@ static int cb_parser_filter(const void *data, size_t bytes,
                             void *context,
                             struct flb_config *config)
 {
-    int continue_parsing;
     struct filter_parser_ctx *ctx = context;
     struct flb_time tm;
     msgpack_object *obj;
-
     msgpack_object_kv *kv;
     int i;
     int ret = FLB_FILTER_NOTOUCH;
@@ -191,10 +184,8 @@ static int cb_parser_filter(const void *data, size_t bytes,
     char *out_buf;
     size_t out_size;
     struct flb_time parsed_time;
-
     msgpack_object_kv **append_arr = NULL;
     size_t append_arr_len = 0;
-    int append_arr_i;
     struct mk_list *head;
     struct filter_parser *fp;
     struct flb_log_event_encoder log_encoder;
@@ -209,9 +200,7 @@ static int cb_parser_filter(const void *data, size_t bytes,
     ret = flb_log_event_decoder_init(&log_decoder, (char *) data, bytes);
 
     if (ret != FLB_EVENT_DECODER_SUCCESS) {
-        flb_plg_error(ctx->ins,
-                      "Log event decoder initialization error : %d", ret);
-
+        flb_plg_error(ctx->ins, "Log event decoder initialization error : %d", ret);
         return FLB_FILTER_NOTOUCH;
     }
 
@@ -219,11 +208,8 @@ static int cb_parser_filter(const void *data, size_t bytes,
                                      FLB_LOG_EVENT_FORMAT_DEFAULT);
 
     if (ret != FLB_EVENT_ENCODER_SUCCESS) {
-        flb_plg_error(ctx->ins,
-                      "Log event encoder initialization error : %d", ret);
-
+        flb_plg_error(ctx->ins, "Log event encoder initialization error : %d", ret);
         flb_log_event_decoder_destroy(&log_decoder);
-
         return FLB_FILTER_NOTOUCH;
     }
 
@@ -231,77 +217,69 @@ static int cb_parser_filter(const void *data, size_t bytes,
                     &log_decoder,
                     &log_event)) == FLB_EVENT_DECODER_SUCCESS) {
         out_buf = NULL;
-        append_arr_i = 0;
 
         flb_time_copy(&tm, &log_event.timestamp);
         obj = log_event.body;
 
         if (obj->type == MSGPACK_OBJECT_MAP) {
             map_num = obj->via.map.size;
-            if (ctx->reserve_data) {
-                append_arr_len = obj->via.map.size;
-                append_arr = flb_calloc(append_arr_len, sizeof(msgpack_object_kv *));
+            /* Calculate initial array size based on configuration */
+            append_arr_len = (ctx->reserve_data ? map_num : 0);
+            if (ctx->preserve_key && !ctx->reserve_data) {
+                append_arr_len = 1; /* Space for preserved key */
+            }
 
+            if (append_arr_len > 0) {
+                append_arr = flb_calloc(append_arr_len, sizeof(msgpack_object_kv *));
                 if (append_arr == NULL) {
                     flb_errno();
-
                     flb_log_event_decoder_destroy(&log_decoder);
                     flb_log_event_encoder_destroy(&log_encoder);
-
                     return FLB_FILTER_NOTOUCH;
                 }
-            }
 
-            continue_parsing = FLB_TRUE;
-            for (i = 0; i < map_num && continue_parsing; i++) {
-                kv = &obj->via.map.ptr[i];
+                /* Initialize array */
                 if (ctx->reserve_data) {
-                    append_arr[append_arr_i] = kv;
-                    append_arr_i++;
+                    for (i = 0; i < map_num; i++) {
+                        append_arr[i] = &obj->via.map.ptr[i];
+                    }
                 }
-                if ( msgpackobj2char(&kv->key, &key_str, &key_len) < 0 ) {
-                    /* key is not string */
+            }
+
+            /* Process the target key */
+            for (i = 0; i < map_num; i++) {
+                kv = &obj->via.map.ptr[i];
+                if (msgpackobj2char(&kv->key, &key_str, &key_len) < 0) {
                     continue;
                 }
+
                 if (key_len == ctx->key_name_len &&
                     !strncmp(key_str, ctx->key_name, key_len)) {
-                    if ( msgpackobj2char(&kv->val, &val_str, &val_len) < 0 ) {
-                        /* val is not string */
+                    if (msgpackobj2char(&kv->val, &val_str, &val_len) < 0) {
                         continue;
                     }
 
                     /* Lookup parser */
                     mk_list_foreach(head, &ctx->parsers) {
                         fp = mk_list_entry(head, struct filter_parser, _head);
-
-                        /* Reset time */
                         flb_time_zero(&parsed_time);
 
                         parse_ret = flb_parser_do(fp->parser, val_str, val_len,
                                                   (void **) &out_buf, &out_size,
                                                   &parsed_time);
                         if (parse_ret >= 0) {
-                            /*
-                             * If the parser succeeded we need to check the
-                             * status of the parsed time. If the time was
-                             * parsed successfully 'parsed_time' will be
-                             * different than zero, if so, override the time
-                             * holder with the new value, otherwise keep the
-                             * original.
-                             */
                             if (flb_time_to_nanosec(&parsed_time) != 0L) {
                                 flb_time_copy(&tm, &parsed_time);
                             }
 
-                            if (ctx->reserve_data) {
+                            if (append_arr != NULL) {
                                 if (!ctx->preserve_key) {
-                                    append_arr_i--;
-                                    append_arr_len--;
-                                    append_arr[append_arr_i] = NULL;
+                                    append_arr[i] = NULL;
+                                }
+                                else if (!ctx->reserve_data) {
+                                    /* Store only the key being preserved */
+                                    append_arr[0] = kv;
                                 }
-                            }
-                            else {
-                                continue_parsing = FLB_FALSE;
                             }
                             break;
                         }
@@ -322,27 +300,58 @@ static int cb_parser_filter(const void *data, size_t bytes,
                         &log_encoder, log_event.metadata);
             }
 
-            if (out_buf != NULL) {
-                if (ctx->reserve_data) {
+            if (out_buf != NULL && parse_ret >= 0) {
+                if (append_arr != NULL && append_arr_len > 0) {
                     char *new_buf = NULL;
-                    int  new_size;
-                    int ret;
-                    ret = flb_msgpack_expand_map(out_buf, out_size,
-                                                 append_arr, append_arr_len,
-                                                 &new_buf, &new_size);
-                    if (ret == -1) {
-                        flb_plg_error(ctx->ins, "cannot expand map");
-
-                        flb_log_event_decoder_destroy(&log_decoder);
-                        flb_log_event_encoder_destroy(&log_encoder);
-                        flb_free(append_arr);
-
-                        return FLB_FILTER_NOTOUCH;
+                    int new_size;
+                    size_t valid_kv_count = 0;
+                    msgpack_object_kv **valid_kv = NULL;
+
+                    /* Count valid entries */
+                    for (i = 0; i < append_arr_len; i++) {
+                        if (append_arr[i] != NULL) {
+                            valid_kv_count++;
+                        }
                     }
 
-                    flb_free(out_buf);
-                    out_buf = new_buf;
-                    out_size = new_size;
+                    if (valid_kv_count > 0) {
+                        valid_kv = flb_calloc(valid_kv_count, sizeof(msgpack_object_kv *));
+                        if (!valid_kv) {
+                            flb_errno();
+                            flb_log_event_decoder_destroy(&log_decoder);
+                            flb_log_event_encoder_destroy(&log_encoder);
+                            flb_free(append_arr);
+                            flb_free(out_buf);
+                            return FLB_FILTER_NOTOUCH;
+                        }
+
+                        /* Fill valid entries */
+                        valid_kv_count = 0;
+                        for (i = 0; i < append_arr_len; i++) {
+                            if (append_arr[i] != NULL) {
+                                valid_kv[valid_kv_count++] = append_arr[i];
+                            }
+                        }
+
+                        ret = flb_msgpack_expand_map(out_buf, out_size,
+                                                   valid_kv, valid_kv_count,
+                                                   &new_buf, &new_size);
+
+                        flb_free(valid_kv);
+
+                        if (ret == -1) {
+                            flb_plg_error(ctx->ins, "cannot expand map");
+                            flb_log_event_decoder_destroy(&log_decoder);
+                            flb_log_event_encoder_destroy(&log_encoder);
+                            flb_free(append_arr);
+                            flb_free(out_buf);
+                            return FLB_FILTER_NOTOUCH;
+                        }
+
+                        flb_free(out_buf);
+                        out_buf = new_buf;
+                        out_size = new_size;
+                    }
                 }
 
                 if (encoder_result == FLB_EVENT_ENCODER_SUCCESS) {
@@ -355,7 +364,6 @@ static int cb_parser_filter(const void *data, size_t bytes,
                 ret = FLB_FILTER_MODIFIED;
             }
             else {
-                /* re-use original data*/
                 if (encoder_result == FLB_EVENT_ENCODER_SUCCESS) {
                     encoder_result = \
                         flb_log_event_encoder_set_body_from_msgpack_object(
@@ -371,26 +379,22 @@ static int cb_parser_filter(const void *data, size_t bytes,
                 flb_plg_error(ctx->ins, "log event encoder error : %d", encoder_result);
             }
 
-            flb_free(append_arr);
-            append_arr = NULL;
-        }
-        else {
-            continue;
+            if (append_arr != NULL) {
+                flb_free(append_arr);
+                append_arr = NULL;
+            }
         }
     }
 
     if (log_encoder.output_length > 0) {
-        *ret_buf   = log_encoder.output_buffer;
+        *ret_buf = log_encoder.output_buffer;
         *ret_bytes = log_encoder.output_length;
 
         ret = FLB_FILTER_MODIFIED;
-
         flb_log_event_encoder_claim_internal_buffer_ownership(&log_encoder);
     }
     else {
-        flb_plg_error(ctx->ins,
-                      "Log event encoder error : %d", ret);
-
+        flb_plg_error(ctx->ins, "Log event encoder error : %d", ret);
         ret = FLB_FILTER_NOTOUCH;
     }
 
@@ -400,7 +404,6 @@ static int cb_parser_filter(const void *data, size_t bytes,
     return ret;
 }
 
-
 static int cb_parser_exit(void *data, struct flb_config *config)
 {
     struct filter_parser_ctx *ctx = data;
diff --git a/tests/runtime/filter_parser.c b/tests/runtime/filter_parser.c
index 8f25fec0e6e..aa5fc7a2d1b 100644
--- a/tests/runtime/filter_parser.c
+++ b/tests/runtime/filter_parser.c
@@ -493,32 +493,61 @@ void flb_test_filter_parser_handle_time_key_with_time_zone()
 }
 
 void test_parser_timestamp_timezone(char *tz,
-                                     char *time_fmt, 
-                                     char *timestamp, 
-                                     char *expected_epoch,
-                                     int use_system_timezone) 
+                                  char *time_fmt,
+                                  char *timestamp,
+                                  char *expected_epoch,
+                                  int use_system_timezone)
 {
     int ret;
     int bytes;
     char *output, *original_tz = NULL;
+    char *saved_tz = NULL;
     char p[256];
-    char expected[12];
+    char expected[256];
     flb_ctx_t *ctx;
     int in_ffd;
     int out_ffd;
     int filter_ffd;
     struct flb_parser *parser;
+    struct flb_lib_out_cb *cb;
 
-    struct flb_lib_out_cb cb;
-    cb.cb   = callback_test;
-    cb.data = NULL;
+    /* Allocate and initialize callback */
+    cb = flb_malloc(sizeof(struct flb_lib_out_cb));
+    if (!cb) {
+        flb_errno();
+        return;
+    }
+    cb->cb = callback_test;
+    cb->data = NULL;
 
     clear_output();
 
+    /* Save current TZ if exists */
+    original_tz = getenv("TZ");
+    if (original_tz) {
+        saved_tz = strdup(original_tz);
+        if (!saved_tz) {
+            flb_free(cb);
+            return;
+        }
+    }
+
+    /* Set new timezone if provided */
+    if (tz) {
+        ret = setenv("TZ", tz, 1);
+        TEST_CHECK(ret == 0);
+        tzset(); /* Make sure timezone changes take effect */
+    }
+
     ctx = flb_create();
+    TEST_CHECK(ctx != NULL);
 
     /* Configure service */
-    flb_service_set(ctx, "Flush", FLUSH_INTERVAL, "Grace", "1", "Log_Level", "debug", NULL);
+    flb_service_set(ctx,
+                    "Flush", FLUSH_INTERVAL,
+                    "Grace", "1",
+                    "Log_Level", "debug",
+                    NULL);
 
     /* Input */
     in_ffd = flb_input(ctx, (char *) "lib", NULL);
@@ -528,21 +557,20 @@ void test_parser_timestamp_timezone(char *tz,
                   NULL);
 
     /* Parser */
-    parser = flb_parser_create("timestamp", // name
-                               "regex", // format
-                               "^(?<time>.*)$", // regex
-                               FLB_TRUE, // skip_empty
-                               time_fmt, // time_fmt
-                               "time", // time_key
-                               NULL, // time_offset
-                               MK_FALSE, // time_keep
-                               MK_TRUE, // time_strict
-                               use_system_timezone, // time_system_timezone
-                               MK_FALSE, // logfmt_no_bare_keys
-                               NULL, // types
-                               0, // types_len
-                               NULL, // decoders
-                               ctx->config); // config
+    parser = flb_parser_create("timestamp",
+                             "regex",
+                             "^(?<time>.*)$",
+                             FLB_TRUE,
+                             time_fmt,
+                             "time",
+                             NULL,
+                             MK_FALSE,
+                             MK_TRUE,
+                             use_system_timezone,
+                             MK_FALSE,
+                             NULL, 0,
+                             NULL,
+                             ctx->config);
     TEST_CHECK(parser != NULL);
 
     /* Filter */
@@ -557,62 +585,50 @@ void test_parser_timestamp_timezone(char *tz,
     TEST_CHECK(ret == 0);
 
     /* Output */
-    out_ffd = flb_output(ctx, (char *) "lib", &cb);
+    out_ffd = flb_output(ctx, (char *) "lib", cb);
     TEST_CHECK(out_ffd >= 0);
     flb_output_set(ctx, out_ffd,
                    "Match", "*",
                    "format", "json",
                    NULL);
 
-    /* Temporarily set the system timezone for this test */
-    if (tz) {
-        original_tz = getenv("TZ");
-        ret = setenv("TZ", tz, 1);
-        TEST_ASSERT(ret == 0);
-    }
-
     /* Start the engine */
     ret = flb_start(ctx);
     TEST_CHECK(ret == 0);
-    if (ret != 0) {
-        flb_destroy(ctx);
-        return;
-    }
 
-    /* Ingest data. */
-    snprintf(p, 256, "[1448403340, {\"@timestamp\":\"%s\", \"message\":\"This is an example\"}]", timestamp);
+    /* Ingest data */
+    snprintf(p, sizeof(p), "[1448403340, {\"@timestamp\":\"%s\", \"message\":\"This is an example\"}]",
+             timestamp);
     bytes = flb_lib_push(ctx, in_ffd, p, strlen(p));
-    TEST_CHECK(bytes == strlen(p));
+    TEST_CHECK(bytes > 0);
 
-    wait_with_timeout(2000, &output); /* waiting flush and ensuring data flush */
-    TEST_CHECK_(output != NULL, "Expected output to not be NULL");
-    if (output != NULL) {
-        snprintf(expected, 12, "[%s", expected_epoch);
-        TEST_CHECK_(strstr(output, expected) != NULL, "Expected output to contain '%s', got '%s'", expected, output);
-    }
+    wait_with_timeout(2000, &output);
+    TEST_CHECK(output != NULL);
 
-    /* Reset original timezone */
-    if (original_tz) {
-        ret = setenv("TZ", original_tz, 1);
-        TEST_CHECK(ret == 0);
+    if (output != NULL) {
+        snprintf(expected, sizeof(expected), "[%s", expected_epoch);
+        TEST_CHECK_(strstr(output, expected) != NULL,
+                   "Expected output to contain '%s', got '%s'", expected, output);
+        free(output);
     }
 
     flb_stop(ctx);
     flb_destroy(ctx);
+    flb_free(cb);
+
+    /* Restore original timezone */
+    if (saved_tz) {
+        setenv("TZ", saved_tz, 1);
+        free(saved_tz);
+    } else if (original_tz == NULL) {
+        unsetenv("TZ");
+    }
+    tzset();
 }
 
-/**
- * In all the following tests, you can verify the correctness of the expected results using the
- * `date` command in Linux:
- *
- * TZ=<desired TZ> date -d "<timestamp>" "+%s"
- *
- * Any output for a timestamp parsed by Fluent Bit using Time_System_Timezone should match
- * the result of this date command.
- */
 void flb_test_filter_parser_use_system_timezone()
 {
-    struct {
+    struct test_case {
         char *tz;
         char *timestamp;
         char *expected_epoch;
@@ -624,19 +640,18 @@ void flb_test_filter_parser_use_system_timezone()
         /* Examples from https://github.com/fluent/fluent-bit/issues/9197. */
         {"Europe/London", "2024-01-20 10:00:00", "1705744800"}, /* Should be ST */
         {"Europe/London", "2024-08-20 11:00:00", "1724148000"},
-                                                          
 
         {NULL, NULL, NULL}
     };
 
-    int i = 0;
-    while(test_cases[i].tz) {
-        test_parser_timestamp_timezone(test_cases[i].tz, /* char *tz */
-                                       "%Y-%m-%d %H:%M:%S", /* char *time_fmt */ 
-                                       test_cases[i].timestamp, /* char *timestamp */
-                                       test_cases[i].expected_epoch, /* char *expected_epoch */
-                                       FLB_TRUE); /* int use_system_timezone */
-        i++;
+    for (int i = 0; test_cases[i].tz != NULL; i++) {
+        test_parser_timestamp_timezone(
+            test_cases[i].tz,
+            "%Y-%m-%d %H:%M:%S",
+            test_cases[i].timestamp,
+            test_cases[i].expected_epoch,
+            FLB_TRUE
+        );
     }
 }
 
@@ -976,6 +991,313 @@ void flb_test_filter_parser_skip_empty_values_false()
     flb_destroy(ctx);
 }
 
+struct test_ctx {
+    flb_ctx_t *flb;
+    int i_ffd;
+    int f_ffd;
+    int o_ffd;
+    struct flb_lib_out_cb *cb;  /* Store callback pointer */
+};
+
+static struct test_ctx *test_ctx_create(char *reserve_data, char *preserve_key)
+{
+    struct test_ctx *ctx = flb_malloc(sizeof(struct test_ctx));
+    struct flb_parser *parser;
+    int ret;
+
+    if (!ctx) {
+        flb_errno();
+        return NULL;
+    }
+
+    /* Create callback structure */
+    ctx->cb = flb_malloc(sizeof(struct flb_lib_out_cb));
+    if (!ctx->cb) {
+        flb_free(ctx);
+        return NULL;
+    }
+
+    ctx->cb->cb = callback_test;
+    ctx->cb->data = NULL;
+
+    /* Create Fluent Bit context */
+    ctx->flb = flb_create();
+    if (!ctx->flb) {
+        flb_free(ctx->cb);
+        flb_free(ctx);
+        return NULL;
+    }
+
+    /* Service config */
+    flb_service_set(ctx->flb,
+                    "Flush", FLUSH_INTERVAL,
+                    "Grace", "1",
+                    "Log_Level", "debug",
+                    NULL);
+
+    /* Input */
+    ctx->i_ffd = flb_input(ctx->flb, (char *) "lib", NULL);
+    if (ctx->i_ffd < 0) {
+        flb_destroy(ctx->flb);
+        flb_free(ctx->cb);
+        flb_free(ctx);
+        return NULL;
+    }
+    flb_input_set(ctx->flb, ctx->i_ffd, "Tag", "test", NULL);
+
+    /* Parser */
+    parser = flb_parser_create("dummy_test",
+                              "regex",
+                              "^(?<INT>[^ ]+) (?<FLOAT>[^ ]+) (?<BOOL>[^ ]+) (?<STRING>.+)$",
+                              FLB_TRUE,
+                              NULL, NULL, NULL,
+                              MK_FALSE, MK_TRUE,
+                              FLB_FALSE, FLB_FALSE,
+                              NULL, 0,
+                              NULL, ctx->flb->config);
+    if (!parser) {
+        flb_destroy(ctx->flb);
+        flb_free(ctx->cb);
+        flb_free(ctx);
+        return NULL;
+    }
+
+    /* Filter */
+    ctx->f_ffd = flb_filter(ctx->flb, (char *) "parser", NULL);
+    if (ctx->f_ffd < 0) {
+        flb_destroy(ctx->flb);
+        flb_free(ctx->cb);
+        flb_free(ctx);
+        return NULL;
+    }
+
+    ret = flb_filter_set(ctx->flb, ctx->f_ffd,
+                        "Match", "test",
+                        "Key_Name", "data",
+                        "Parser", "dummy_test",
+                        "Reserve_Data", reserve_data,
+                        "Preserve_Key", preserve_key,
+                        NULL);
+    if (ret != 0) {
+        flb_destroy(ctx->flb);
+        flb_free(ctx->cb);
+        flb_free(ctx);
+        return NULL;
+    }
+
+    /* Output with properly aligned callback */
+    ctx->o_ffd = flb_output(ctx->flb, (char *) "lib", ctx->cb);
+    if (ctx->o_ffd < 0) {
+        flb_destroy(ctx->flb);
+        flb_free(ctx->cb);
+        flb_free(ctx);
+        return NULL;
+    }
+
+    flb_output_set(ctx->flb, ctx->o_ffd,
+                   "Match", "*",
+                   "format", "json",
+                   NULL);
+
+    return ctx;
+}
+
+static void test_ctx_destroy(struct test_ctx *ctx)
+{
+    if (!ctx) {
+        return;
+    }
+
+    if (ctx->flb) {
+        flb_stop(ctx->flb);
+        flb_destroy(ctx->flb);
+    }
+
+    if (ctx->cb) {
+        flb_free(ctx->cb);
+    }
+
+    flb_free(ctx);
+}
+
+/* Test case 1: Reserve_Data=Off, Preserve_Key=Off */
+void flb_test_filter_parser_reserve_off_preserve_off()
+{
+    char *output = NULL;
+    struct test_ctx *ctx;
+
+    ctx = test_ctx_create("Off", "Off");
+    TEST_CHECK(ctx != NULL);
+    clear_output();
+
+    int ret = flb_start(ctx->flb);
+    TEST_CHECK(ret == 0);
+
+    /* Ingest data */
+    int bytes = flb_lib_push(ctx->flb, ctx->i_ffd,
+                            "[1448403340,{\"data\":\"100 0.5 true This is an example\", \"extra\":\"should not appear\"}]",
+                            strlen("[1448403340,{\"data\":\"100 0.5 true This is an example\", \"extra\":\"should not appear\"}]"));
+    TEST_CHECK(bytes > 0);
+
+    wait_with_timeout(2000, &output);
+    TEST_CHECK(output != NULL);
+
+    if (output != NULL) {
+        /* Positive checks: should contain parsed fields */
+        TEST_CHECK_(strstr(output, "\"INT\":\"100\"") != NULL,
+                   "Expected output to contain INT field");
+        TEST_CHECK_(strstr(output, "\"FLOAT\":\"0.5\"") != NULL,
+                   "Expected output to contain FLOAT field");
+        TEST_CHECK_(strstr(output, "\"BOOL\":\"true\"") != NULL,
+                   "Expected output to contain BOOL field");
+        TEST_CHECK_(strstr(output, "\"STRING\":\"This is an example\"") != NULL,
+                   "Expected output to contain STRING field");
+
+        /* Negative checks: should not contain original fields */
+        TEST_CHECK_(strstr(output, "\"data\":") == NULL,
+                   "Expected output to not contain original data field");
+        TEST_CHECK_(strstr(output, "\"extra\":") == NULL,
+                   "Expected output to not contain extra field");
+
+        free(output);
+    }
+
+    test_ctx_destroy(ctx);
+}
+
+/* Test case 2: Reserve_Data=Off, Preserve_Key=On */
+void flb_test_filter_parser_reserve_off_preserve_on()
+{
+    char *output = NULL;
+    struct test_ctx *ctx;
+
+    ctx = test_ctx_create("Off", "On");
+    TEST_CHECK(ctx != NULL);
+    clear_output();
+
+    int ret = flb_start(ctx->flb);
+    TEST_CHECK(ret == 0);
+
+    /* Ingest data */
+    int bytes = flb_lib_push(ctx->flb, ctx->i_ffd,
+                            "[1448403340,{\"data\":\"100 0.5 true This is an example\", \"extra\":\"should not appear\"}]",
+                            strlen("[1448403340,{\"data\":\"100 0.5 true This is an example\", \"extra\":\"should not appear\"}]"));
+    TEST_CHECK(bytes > 0);
+
+    wait_with_timeout(2000, &output);
+    TEST_CHECK(output != NULL);
+
+    if (output != NULL) {
+        /* Positive checks */
+        TEST_CHECK_(strstr(output, "\"data\":\"100 0.5 true This is an example\"") != NULL,
+                   "Expected output to contain original data field");
+        TEST_CHECK_(strstr(output, "\"INT\":\"100\"") != NULL,
+                   "Expected output to contain INT field");
+        TEST_CHECK_(strstr(output, "\"FLOAT\":\"0.5\"") != NULL,
+                   "Expected output to contain FLOAT field");
+        TEST_CHECK_(strstr(output, "\"BOOL\":\"true\"") != NULL,
+                   "Expected output to contain BOOL field");
+        TEST_CHECK_(strstr(output, "\"STRING\":\"This is an example\"") != NULL,
+                   "Expected output to contain STRING field");
+
+        /* Negative checks */
+        TEST_CHECK_(strstr(output, "\"extra\":") == NULL,
+                   "Expected output to not contain extra field");
+
+        free(output);
+    }
+
+    test_ctx_destroy(ctx);
+}
+
+/* Test case 3: Reserve_Data=On, Preserve_Key=Off */
+void flb_test_filter_parser_reserve_on_preserve_off()
+{
+    char *output = NULL;
+    struct test_ctx *ctx;
+
+    ctx = test_ctx_create("On", "Off");
+    TEST_CHECK(ctx != NULL);
+    clear_output();
+
+    int ret = flb_start(ctx->flb);
+    TEST_CHECK(ret == 0);
+
+    /* Ingest data */
+    int bytes = flb_lib_push(ctx->flb, ctx->i_ffd,
+                            "[1448403340,{\"data\":\"100 0.5 true This is an example\", \"extra\":\"should appear\"}]",
+                            strlen("[1448403340,{\"data\":\"100 0.5 true This is an example\", \"extra\":\"should appear\"}]"));
+    TEST_CHECK(bytes > 0);
+
+    wait_with_timeout(2000, &output);
+    TEST_CHECK(output != NULL);
+
+    if (output != NULL) {
+        /* Positive checks */
+        TEST_CHECK_(strstr(output, "\"extra\":\"should appear\"") != NULL,
+                   "Expected output to contain extra field");
+        TEST_CHECK_(strstr(output, "\"INT\":\"100\"") != NULL,
+                   "Expected output to contain INT field");
+        TEST_CHECK_(strstr(output, "\"FLOAT\":\"0.5\"") != NULL,
+                   "Expected output to contain FLOAT field");
+        TEST_CHECK_(strstr(output, "\"BOOL\":\"true\"") != NULL,
+                   "Expected output to contain BOOL field");
+        TEST_CHECK_(strstr(output, "\"STRING\":\"This is an example\"") != NULL,
+                   "Expected output to contain STRING field");
+
+        /* Negative checks */
+        TEST_CHECK_(strstr(output, "\"data\":") == NULL,
+                   "Expected output to not contain original data field");
+
+        free(output);
+    }
+
+    test_ctx_destroy(ctx);
+}
+
+/* Test case 4: Reserve_Data=On, Preserve_Key=On */
+void flb_test_filter_parser_reserve_on_preserve_on()
+{
+    char *output = NULL;
+    struct test_ctx *ctx;
+
+    ctx = test_ctx_create("On", "On");
+    TEST_CHECK(ctx != NULL);
+    clear_output();
+
+    int ret = flb_start(ctx->flb);
+    TEST_CHECK(ret == 0);
+
+    /* Ingest data */
+    int bytes = flb_lib_push(ctx->flb, ctx->i_ffd,
+                            "[1448403340,{\"data\":\"100 0.5 true This is an example\", \"extra\":\"should appear\"}]",
+                            strlen("[1448403340,{\"data\":\"100 0.5 true This is an example\", \"extra\":\"should appear\"}]"));
+    TEST_CHECK(bytes > 0);
+
+    wait_with_timeout(2000, &output);
+    TEST_CHECK(output != NULL);
+
+    if (output != NULL) {
+        /* Positive checks - all fields should be present */
+        TEST_CHECK_(strstr(output, "\"data\":\"100 0.5 true This is an example\"") != NULL,
+                   "Expected output to contain original data field");
+        TEST_CHECK_(strstr(output, "\"extra\":\"should appear\"") != NULL,
+                   "Expected output to contain extra field");
+        TEST_CHECK_(strstr(output, "\"INT\":\"100\"") != NULL,
+                   "Expected output to contain INT field");
+        TEST_CHECK_(strstr(output, "\"FLOAT\":\"0.5\"") != NULL,
+                   "Expected output to contain FLOAT field");
+        TEST_CHECK_(strstr(output, "\"BOOL\":\"true\"") != NULL,
+                   "Expected output to contain BOOL field");
+        TEST_CHECK_(strstr(output, "\"STRING\":\"This is an example\"") != NULL,
+                   "Expected output to contain STRING field");
+
+        free(output);
+    }
+
+    test_ctx_destroy(ctx);
+}
+
 TEST_LIST = {
     {"filter_parser_extract_fields", flb_test_filter_parser_extract_fields },
     {"filter_parser_reserve_data_off", flb_test_filter_parser_reserve_data_off },
@@ -986,6 +1308,10 @@ TEST_LIST = {
     {"filter_parser_preserve_original_field", flb_test_filter_parser_preserve_original_field },
     {"filter_parser_first_matched_when_multiple_parser", flb_test_filter_parser_first_matched_when_mutilple_parser },
     {"filter_parser_skip_empty_values_false", flb_test_filter_parser_skip_empty_values_false},
+    {"filter_parser_reserve_off_preserve_off", flb_test_filter_parser_reserve_off_preserve_off},
+    {"filter_parser_reserve_off_preserve_on", flb_test_filter_parser_reserve_off_preserve_on},
+    {"filter_parser_reserve_on_preserve_off", flb_test_filter_parser_reserve_on_preserve_off},
+    {"filter_parser_reserve_on_preserve_on", flb_test_filter_parser_reserve_on_preserve_on},
     {NULL, NULL}
 };