From 9afc4e7e7d43c44145865a717366a75259a9ae5b Mon Sep 17 00:00:00 2001 From: Nour Douffir Date: Fri, 21 Nov 2025 15:25:40 -0500 Subject: [PATCH 1/4] in_systemd: add FLUENT_BIT_PARSER support Co-authored-by: Dennis Kaarsemaker Signed-off-by: Nour Douffir --- plugins/in_systemd/systemd.c | 111 ++++++++++++++++++++++++++++++++++- 1 file changed, 108 insertions(+), 3 deletions(-) diff --git a/plugins/in_systemd/systemd.c b/plugins/in_systemd/systemd.c index 2f2e6f4885b..a55b20303aa 100644 --- a/plugins/in_systemd/systemd.c +++ b/plugins/in_systemd/systemd.c @@ -21,6 +21,8 @@ #include #include #include +#include +#include #include "systemd_config.h" #include "systemd_db.h" @@ -70,6 +72,60 @@ static int tag_compose(const char *tag, const char *unit_name, return 0; } +/* Helper function to unpack and repack msgpack data from parser output */ +static int flb_systemd_repack_map(struct flb_log_event_encoder *encoder, + char *data, + size_t data_size) +{ + msgpack_unpacked source_map; + size_t offset; + int result; + size_t index; + msgpack_object value; + msgpack_object key; + + result = FLB_EVENT_ENCODER_SUCCESS; + + if (data_size > 0) { + msgpack_unpacked_init(&source_map); + + offset = 0; + result = msgpack_unpack_next(&source_map, + data, + data_size, + &offset); + + if (result == MSGPACK_UNPACK_SUCCESS) { + result = FLB_EVENT_ENCODER_SUCCESS; + } + else { + result = FLB_EVENT_DECODER_ERROR_DESERIALIZATION_FAILURE; + } + + for (index = 0; + index < source_map.data.via.map.size && + result == FLB_EVENT_ENCODER_SUCCESS; + index++) { + key = source_map.data.via.map.ptr[index].key; + value = source_map.data.via.map.ptr[index].val; + + result = flb_log_event_encoder_append_body_msgpack_object( + encoder, + &key); + + if (result == FLB_EVENT_ENCODER_SUCCESS) { + result = flb_log_event_encoder_append_body_msgpack_object( + encoder, + &value); + } + } + + msgpack_unpacked_destroy(&source_map); + } + + return result; +} + static int append_enumerate_data(struct flb_systemd_config *ctx, struct cfl_kvlist *kvlist) { int i; @@ -129,7 +185,10 @@ static int systemd_enumerate_data_store(struct flb_config *config, struct flb_input_instance *ins, void *plugin_context, void *format_context, - const void *data, size_t data_size) + const void *data, size_t data_size, + struct flb_parser *parser, + void **out_buf, size_t *out_size, + struct flb_time *out_time) { int i; int len; @@ -155,6 +214,24 @@ static int systemd_enumerate_data_store(struct flb_config *config, len = (sep - key); key_len = len; + + /* Skip FLUENT_BIT_PARSER field - it's metadata, not log content */ + if (strncmp(key, "FLUENT_BIT_PARSER", key_len) == 0) { + return 0; + } + + /* If this is MESSAGE field and parser is specified, apply parser */ + if (parser && strncmp(key, "MESSAGE", key_len) == 0) { + val = sep + 1; + len = length - (sep - key) - 1; + int ret_parser = flb_parser_do(parser, val, len, out_buf, out_size, out_time); + if (ret_parser != -1) { + /* Return special code to indicate parsed content should be added */ + return -3; + } + /* If parser failed, continue with unparsed message */ + } + list_key = flb_sds_create_len(key, key_len); if (!list_key) { @@ -261,19 +338,23 @@ static int in_systemd_collect(struct flb_input_instance *ins, long nsec; uint64_t usec; size_t length; + size_t plength; const char *key; #ifdef FLB_HAVE_SQLDB char *cursor = NULL; #endif char *tag = NULL; + char *name; char new_tag[PATH_MAX]; char last_tag[PATH_MAX] = {0}; size_t tag_len; size_t last_tag_len = 0; const void *data; + void *pbuf = NULL; struct flb_systemd_config *ctx = in_context; struct flb_time tm; struct cfl_kvlist *kvlist = NULL; + struct flb_parser *parser; /* Restricted by mem_buf_limit */ if (flb_input_buf_paused(ins) == FLB_TRUE) { @@ -337,6 +418,18 @@ static int in_systemd_collect(struct flb_input_instance *ins, tag_len = ctx->ins->tag_len; } + /* Find the parser, if specified */ + parser = NULL; + ret = sd_journal_get_data(ctx->j, "FLUENT_BIT_PARSER", &data, &length); + if (ret == 0) { + name = flb_strndup((const char *)(data+18), length-18); + parser = flb_parser_get(name, config); + if (!parser) { + flb_plg_error(ctx->ins, "no such parser: '%s'", name); + } + flb_free(name); + } + if (last_tag_len == 0) { strncpy(last_tag, tag, tag_len); last_tag_len = tag_len; @@ -412,11 +505,22 @@ static int in_systemd_collect(struct flb_input_instance *ins, ret = systemd_enumerate_data_store(config, ctx->ins, (void *)ctx, (void *)kvlist, - key, length); + key, length, parser, &pbuf, &plength, &tm); if (ret == -2) { skip_entries++; continue; } + else if (ret == -3) { + /* Parsed content - add it to encoder as msgpack */ + ret = flb_systemd_repack_map(ctx->log_encoder, pbuf, plength); + flb_free(pbuf); + pbuf = NULL; + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + continue; + } + entries++; + continue; + } else if (ret == -1) { continue; } @@ -688,7 +792,8 @@ static int cb_systemd_format_test(struct flb_config *config, cur = cfl_list_entry(head, struct cfl_split_entry, _head); ret = systemd_enumerate_data_store(config, ctx->ins, (void *)ctx, (void *)kvlist, - cur->value, cur->len); + cur->value, cur->len, + NULL, NULL, NULL, &tm); if (ret == -2 || ret == -1) { continue; From 3588941b0412d974352be9ff55853e444690626e Mon Sep 17 00:00:00 2001 From: Nour Douffir Date: Wed, 7 Jan 2026 18:38:16 -0500 Subject: [PATCH 2/4] in_systemd: improve error handling This patch addresses three issues: - Add NULL check after flb_strndup to prevent potential segfault when memory allocation fails - Add type validation for msgpack object before accessing map fields to prevent undefined behavior with non-map data - Fix FLUENT_BIT_PARSER counting toward max_fields limit by returning -1 instead of 0 Signed-off-by: Nour Douffir --- plugins/in_systemd/systemd.c | 25 ++++++++++++++++++------- 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/plugins/in_systemd/systemd.c b/plugins/in_systemd/systemd.c index a55b20303aa..bdb10fead8a 100644 --- a/plugins/in_systemd/systemd.c +++ b/plugins/in_systemd/systemd.c @@ -96,7 +96,12 @@ static int flb_systemd_repack_map(struct flb_log_event_encoder *encoder, &offset); if (result == MSGPACK_UNPACK_SUCCESS) { - result = FLB_EVENT_ENCODER_SUCCESS; + if (source_map.data.type == MSGPACK_OBJECT_MAP) { + result = FLB_EVENT_ENCODER_SUCCESS; + } + else { + result = FLB_EVENT_DECODER_ERROR_DESERIALIZATION_FAILURE; + } } else { result = FLB_EVENT_DECODER_ERROR_DESERIALIZATION_FAILURE; @@ -215,9 +220,10 @@ static int systemd_enumerate_data_store(struct flb_config *config, len = (sep - key); key_len = len; - /* Skip FLUENT_BIT_PARSER field - it's metadata, not log content */ + /* Skip FLUENT_BIT_PARSER field - it's metadata, not log content + * Return -1 so it doesn't count toward max_fields */ if (strncmp(key, "FLUENT_BIT_PARSER", key_len) == 0) { - return 0; + return -1; } /* If this is MESSAGE field and parser is specified, apply parser */ @@ -423,11 +429,16 @@ static int in_systemd_collect(struct flb_input_instance *ins, ret = sd_journal_get_data(ctx->j, "FLUENT_BIT_PARSER", &data, &length); if (ret == 0) { name = flb_strndup((const char *)(data+18), length-18); - parser = flb_parser_get(name, config); - if (!parser) { - flb_plg_error(ctx->ins, "no such parser: '%s'", name); + if (name == NULL) { + flb_plg_error(ctx->ins, "failed to allocate parser name"); + } + else { + parser = flb_parser_get(name, config); + if (!parser) { + flb_plg_error(ctx->ins, "no such parser: '%s'", name); + } + flb_free(name); } - flb_free(name); } if (last_tag_len == 0) { From 3f9651b930acfb6cb0c4624c61180d7109eb8810 Mon Sep 17 00:00:00 2001 From: Nour Douffir Date: Fri, 10 Jul 2026 14:02:22 -0400 Subject: [PATCH 3/4] in_systemd: address review feedback - Replace magic strings/numbers with named constants for FLUENT_BIT_PARSER - Fix key comparisons to require exact length match - Add NULL check for pbuf before repacking parsed content - Initialize plength to 0 for defensive safety - Document return code convention for systemd_enumerate_data_store Signed-off-by: Nour Douffir --- plugins/in_systemd/systemd.c | 67 ++++++++++++++++++++++++++++++++---- 1 file changed, 61 insertions(+), 6 deletions(-) diff --git a/plugins/in_systemd/systemd.c b/plugins/in_systemd/systemd.c index bdb10fead8a..9793fd027ee 100644 --- a/plugins/in_systemd/systemd.c +++ b/plugins/in_systemd/systemd.c @@ -33,6 +33,11 @@ #define pack_uint16(buf, d) _msgpack_store16(buf, (uint16_t) d) #define pack_uint32(buf, d) _msgpack_store32(buf, (uint32_t) d) +/* FLUENT_BIT_PARSER journal field: "FLUENT_BIT_PARSER=" */ +#define FLUENT_BIT_PARSER_FIELD "FLUENT_BIT_PARSER" +#define FLUENT_BIT_PARSER_FIELD_LEN 17 +#define FLUENT_BIT_PARSER_PREFIX_LEN 18 /* includes the '=' separator */ + /* tag composer */ static int tag_compose(const char *tag, const char *unit_name, int unit_size, char **out_buf, size_t *out_size) @@ -186,6 +191,15 @@ static int append_enumerate_data(struct flb_systemd_config *ctx, struct cfl_kvli return ret; } +/* + * Process a single journal field (key=value). + * + * Return values: + * 0 : success, field stored in kvlist + * -1 : error or metadata-only field (skip, don't count) + * -2 : malformed field, no '=' separator found (skip) + * -3 : parsed content available in out_buf/out_size (caller must repack) + */ static int systemd_enumerate_data_store(struct flb_config *config, struct flb_input_instance *ins, void *plugin_context, @@ -222,12 +236,13 @@ static int systemd_enumerate_data_store(struct flb_config *config, /* Skip FLUENT_BIT_PARSER field - it's metadata, not log content * Return -1 so it doesn't count toward max_fields */ - if (strncmp(key, "FLUENT_BIT_PARSER", key_len) == 0) { + if (key_len == FLUENT_BIT_PARSER_FIELD_LEN && + strncmp(key, FLUENT_BIT_PARSER_FIELD, FLUENT_BIT_PARSER_FIELD_LEN) == 0) { return -1; } /* If this is MESSAGE field and parser is specified, apply parser */ - if (parser && strncmp(key, "MESSAGE", key_len) == 0) { + if (parser && key_len == 7 && strncmp(key, "MESSAGE", 7) == 0) { val = sep + 1; len = length - (sep - key) - 1; int ret_parser = flb_parser_do(parser, val, len, out_buf, out_size, out_time); @@ -344,7 +359,7 @@ static int in_systemd_collect(struct flb_input_instance *ins, long nsec; uint64_t usec; size_t length; - size_t plength; + size_t plength = 0; const char *key; #ifdef FLB_HAVE_SQLDB char *cursor = NULL; @@ -426,9 +441,10 @@ static int in_systemd_collect(struct flb_input_instance *ins, /* Find the parser, if specified */ parser = NULL; - ret = sd_journal_get_data(ctx->j, "FLUENT_BIT_PARSER", &data, &length); + ret = sd_journal_get_data(ctx->j, FLUENT_BIT_PARSER_FIELD, &data, &length); if (ret == 0) { - name = flb_strndup((const char *)(data+18), length-18); + name = flb_strndup((const char *)(data + FLUENT_BIT_PARSER_PREFIX_LEN), + length - FLUENT_BIT_PARSER_PREFIX_LEN); if (name == NULL) { flb_plg_error(ctx->ins, "failed to allocate parser name"); } @@ -523,6 +539,12 @@ static int in_systemd_collect(struct flb_input_instance *ins, } else if (ret == -3) { /* Parsed content - add it to encoder as msgpack */ + if (pbuf == NULL) { + flb_plg_warn(ctx->ins, + "parser returned success but NULL buffer"); + skip_entries++; + continue; + } ret = flb_systemd_repack_map(ctx->log_encoder, pbuf, plength); flb_free(pbuf); pbuf = NULL; @@ -778,6 +800,9 @@ static int cb_systemd_format_test(struct flb_config *config, struct cfl_list *kvs = NULL; struct cfl_split_entry *cur = NULL; struct cfl_kvlist *kvlist = NULL; + struct flb_parser *parser = NULL; + void *pbuf = NULL; + size_t plength = 0; const char *keys; ret = flb_log_event_encoder_begin_record(ctx->log_encoder); @@ -799,16 +824,46 @@ static int cb_systemd_format_test(struct flb_config *config, goto split_error; } + /* + * First pass: look for a FLUENT_BIT_PARSER field so we can resolve the + * parser before processing the remaining fields (mirrors what + * in_systemd_collect does with sd_journal_get_data). + */ + parser = NULL; + cfl_list_foreach(head, kvs) { + cur = cfl_list_entry(head, struct cfl_split_entry, _head); + if (cur->len > FLUENT_BIT_PARSER_PREFIX_LEN && + strncmp(cur->value, FLUENT_BIT_PARSER_FIELD "=", + FLUENT_BIT_PARSER_PREFIX_LEN) == 0) { + char *pname = flb_strndup( + cur->value + FLUENT_BIT_PARSER_PREFIX_LEN, + cur->len - FLUENT_BIT_PARSER_PREFIX_LEN); + if (pname) { + parser = flb_parser_get(pname, config); + flb_free(pname); + } + break; + } + } + + /* Second pass: enumerate all fields */ cfl_list_foreach(head, kvs) { cur = cfl_list_entry(head, struct cfl_split_entry, _head); ret = systemd_enumerate_data_store(config, ctx->ins, (void *)ctx, (void *)kvlist, cur->value, cur->len, - NULL, NULL, NULL, &tm); + parser, &pbuf, &plength, &tm); if (ret == -2 || ret == -1) { continue; } + else if (ret == -3) { + if (pbuf != NULL) { + ret = flb_systemd_repack_map(ctx->log_encoder, pbuf, plength); + flb_free(pbuf); + pbuf = NULL; + } + } } /* Interpret cfl_kvlist as logs type of events later. */ From 92a47562d82d86dd0d2c421e27f9519191b1194d Mon Sep 17 00:00:00 2001 From: Nour Douffir Date: Fri, 10 Jul 2026 14:02:22 -0400 Subject: [PATCH 4/4] tests: in_systemd: add parser tests - Update test formatter to exercise parser code path - Add tests: parser_field_excluded, parser_logfmt, parser_fallback Signed-off-by: Nour Douffir --- tests/runtime/data/in_systemd/parsers.conf | 7 + tests/runtime/in_systemd.c | 325 ++++++++++++++++++++- 2 files changed, 330 insertions(+), 2 deletions(-) create mode 100644 tests/runtime/data/in_systemd/parsers.conf diff --git a/tests/runtime/data/in_systemd/parsers.conf b/tests/runtime/data/in_systemd/parsers.conf new file mode 100644 index 00000000000..f36b7bf1ce1 --- /dev/null +++ b/tests/runtime/data/in_systemd/parsers.conf @@ -0,0 +1,7 @@ +[PARSER] + Name logfmt + Format logfmt + +[PARSER] + Name json_test + Format json diff --git a/tests/runtime/in_systemd.c b/tests/runtime/in_systemd.c index d3ca436ebce..804221a67a6 100644 --- a/tests/runtime/in_systemd.c +++ b/tests/runtime/in_systemd.c @@ -24,6 +24,10 @@ #include "flb_tests_runtime.h" +#define DPATH FLB_TESTS_DATA_PATH "/data/in_systemd" + +/* ---------- existing test: duplicated keys ---------- */ + static void cb_check_cfl_variant_properties(void *ctx, int ffd, int res_ret, void *res_data, size_t res_size, void *data) @@ -183,10 +187,327 @@ void flb_test_namespace_path_conflict() flb_destroy(ctx); } +/* ---------- parser: FLUENT_BIT_PARSER field is excluded, MESSAGE is parsed ---------- */ + +static void cb_check_parser_field_excluded(void *ctx, int ffd, + int res_ret, void *res_data, size_t res_size, + void *data) +{ + flb_sds_t output; + char *result = NULL; + + output = flb_msgpack_raw_to_json_sds(res_data, res_size, FLB_TRUE); + TEST_CHECK(output != NULL); + + /* FLUENT_BIT_PARSER must NOT appear in the output */ + result = strstr(output, "FLUENT_BIT_PARSER"); + if (TEST_CHECK(result == NULL)) { + TEST_MSG("FLUENT_BIT_PARSER should be excluded, output:%s\n", output); + } + + /* Raw MESSAGE must NOT appear: parsed fields replace it on success */ + result = strstr(output, "\"MESSAGE\":"); + if (TEST_CHECK(result == NULL)) { + TEST_MSG("raw MESSAGE should be replaced by parsed fields, output:%s\n", output); + } + + /* The parsed fields from logfmt should appear instead of raw MESSAGE */ + result = strstr(output, "\"level\""); + if (TEST_CHECK(result != NULL)) { + TEST_MSG("expected parsed 'level' field, output:%s\n", output); + } + + result = strstr(output, "\"msg\""); + if (TEST_CHECK(result != NULL)) { + TEST_MSG("expected parsed 'msg' field, output:%s\n", output); + } + + /* UNIT field should still be present */ + result = strstr(output, "\"UNIT\":\"myservice\""); + if (TEST_CHECK(result != NULL)) { + TEST_MSG("expected UNIT field, output:%s\n", output); + } + + flb_sds_destroy(output); +} + +void flb_test_parser_field_excluded() +{ + int ret; + int in_ffd; + int out_ffd; + flb_ctx_t *ctx; + /* logfmt message with FLUENT_BIT_PARSER specifying the parser */ + char *message = "FLUENT_BIT_PARSER=logfmt\n" + "MESSAGE=level=info msg=\"hello world\" caller=main.go\n" + "UNIT=myservice\n"; + + ctx = flb_create(); + flb_service_set(ctx, + "flush", "2", + "grace", "1", + "Log_Level", "error", + "Parsers_File", DPATH "/parsers.conf", + NULL); + + in_ffd = flb_input(ctx, (char *) "systemd", NULL); + flb_input_set(ctx, in_ffd, + "tag", "test", + "Read_From_Tail", "On", + NULL); + + out_ffd = flb_output(ctx, (char *) "null", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + NULL); + + ret = flb_input_set_test(ctx, in_ffd, "formatter", + cb_check_parser_field_excluded, + NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + ret = flb_lib_push(ctx, in_ffd, message, strlen(message)); + TEST_CHECK(ret == 0); + + sleep(2); + flb_stop(ctx); + flb_destroy(ctx); +} + +/* ---------- parser: logfmt MESSAGE is parsed into structured fields ---------- */ + +static void cb_check_parser_logfmt(void *ctx, int ffd, + int res_ret, void *res_data, size_t res_size, + void *data) +{ + flb_sds_t output; + char *result = NULL; + + output = flb_msgpack_raw_to_json_sds(res_data, res_size, FLB_TRUE); + TEST_CHECK(output != NULL); + + /* Parsed logfmt fields should be present */ + result = strstr(output, "\"level\":\"info\""); + if (TEST_CHECK(result != NULL)) { + TEST_MSG("expected level=info, output:%s\n", output); + } + + result = strstr(output, "\"msg\":\"hello world\""); + if (TEST_CHECK(result != NULL)) { + TEST_MSG("expected msg=hello world, output:%s\n", output); + } + + result = strstr(output, "\"caller\":\"main.go\""); + if (TEST_CHECK(result != NULL)) { + TEST_MSG("expected caller=main.go, output:%s\n", output); + } + + flb_sds_destroy(output); +} + +void flb_test_parser_logfmt() +{ + int ret; + int in_ffd; + int out_ffd; + flb_ctx_t *ctx; + char *message = "FLUENT_BIT_PARSER=logfmt\n" + "MESSAGE=level=info msg=\"hello world\" caller=main.go\n" + "UNIT=myservice\n"; + + ctx = flb_create(); + flb_service_set(ctx, + "flush", "2", + "grace", "1", + "Log_Level", "error", + "Parsers_File", DPATH "/parsers.conf", + NULL); + + in_ffd = flb_input(ctx, (char *) "systemd", NULL); + flb_input_set(ctx, in_ffd, + "tag", "test", + "Read_From_Tail", "On", + NULL); + + out_ffd = flb_output(ctx, (char *) "null", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + NULL); + + ret = flb_input_set_test(ctx, in_ffd, "formatter", + cb_check_parser_logfmt, + NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + ret = flb_lib_push(ctx, in_ffd, message, strlen(message)); + TEST_CHECK(ret == 0); + + sleep(2); + flb_stop(ctx); + flb_destroy(ctx); +} + +/* ---------- parser: unknown parser falls back to raw MESSAGE ---------- */ + +static void cb_check_parser_fallback(void *ctx, int ffd, + int res_ret, void *res_data, size_t res_size, + void *data) +{ + flb_sds_t output; + char *result = NULL; + + output = flb_msgpack_raw_to_json_sds(res_data, res_size, FLB_TRUE); + TEST_CHECK(output != NULL); + + /* With an unknown parser, MESSAGE should be passed through raw */ + result = strstr(output, "\"MESSAGE\":"); + if (TEST_CHECK(result != NULL)) { + TEST_MSG("expected raw MESSAGE field, output:%s\n", output); + } + + /* The raw value should contain the unparsed logfmt content */ + result = strstr(output, "level=info"); + if (TEST_CHECK(result != NULL)) { + TEST_MSG("expected raw logfmt content in MESSAGE, output:%s\n", output); + } + + flb_sds_destroy(output); +} + +void flb_test_parser_fallback() +{ + int ret; + int in_ffd; + int out_ffd; + flb_ctx_t *ctx; + /* Use a parser name that doesn't exist */ + char *message = "FLUENT_BIT_PARSER=nonexistent_parser\n" + "MESSAGE=level=info msg=\"hello world\"\n" + "UNIT=myservice\n"; + + ctx = flb_create(); + flb_service_set(ctx, + "flush", "2", + "grace", "1", + "Log_Level", "error", + "Parsers_File", DPATH "/parsers.conf", + NULL); + + in_ffd = flb_input(ctx, (char *) "systemd", NULL); + flb_input_set(ctx, in_ffd, + "tag", "test", + "Read_From_Tail", "On", + NULL); + + out_ffd = flb_output(ctx, (char *) "null", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + NULL); + + ret = flb_input_set_test(ctx, in_ffd, "formatter", + cb_check_parser_fallback, + NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + ret = flb_lib_push(ctx, in_ffd, message, strlen(message)); + TEST_CHECK(ret == 0); + + sleep(2); + flb_stop(ctx); + flb_destroy(ctx); +} + +/* ---------- parser: parser execution failure falls back to raw MESSAGE ---------- */ + +static void cb_check_parser_exec_failure(void *ctx, int ffd, + int res_ret, void *res_data, size_t res_size, + void *data) +{ + flb_sds_t output; + char *result = NULL; + + output = flb_msgpack_raw_to_json_sds(res_data, res_size, FLB_TRUE); + TEST_CHECK(output != NULL); + + /* + * The parser exists but fails to parse the message (invalid JSON), so the + * plugin must fall back to emitting the raw MESSAGE field. This exercises + * the flb_parser_do() == -1 path, unlike the nonexistent-parser case which + * fails at parser lookup. + */ + result = strstr(output, "\"MESSAGE\":"); + if (TEST_CHECK(result != NULL)) { + TEST_MSG("expected raw MESSAGE field on parse failure, output:%s\n", output); + } + + /* The raw value should contain the unparsed content */ + result = strstr(output, "not valid json"); + if (TEST_CHECK(result != NULL)) { + TEST_MSG("expected raw MESSAGE content, output:%s\n", output); + } + + flb_sds_destroy(output); +} + +void flb_test_parser_exec_failure() +{ + int ret; + int in_ffd; + int out_ffd; + flb_ctx_t *ctx; + /* Parser exists (json_test) but the MESSAGE is not valid JSON */ + char *message = "FLUENT_BIT_PARSER=json_test\n" + "MESSAGE=this is not valid json\n" + "UNIT=myservice\n"; + + ctx = flb_create(); + flb_service_set(ctx, + "flush", "2", + "grace", "1", + "Log_Level", "error", + "Parsers_File", DPATH "/parsers.conf", + NULL); + + in_ffd = flb_input(ctx, (char *) "systemd", NULL); + flb_input_set(ctx, in_ffd, + "tag", "test", + "Read_From_Tail", "On", + NULL); + + out_ffd = flb_output(ctx, (char *) "null", NULL); + flb_output_set(ctx, out_ffd, + "match", "test", + NULL); + + ret = flb_input_set_test(ctx, in_ffd, "formatter", + cb_check_parser_exec_failure, + NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + ret = flb_lib_push(ctx, in_ffd, message, strlen(message)); + TEST_CHECK(ret == 0); + + sleep(2); + flb_stop(ctx); + flb_destroy(ctx); +} + /* Test list */ TEST_LIST = { - { "duplicated_keys", flb_test_duplicated_keys }, - { "namespace", flb_test_namespace }, + { "duplicated_keys", flb_test_duplicated_keys }, + { "namespace", flb_test_namespace }, { "namespace_path_conflict", flb_test_namespace_path_conflict }, + { "parser_field_excluded", flb_test_parser_field_excluded }, + { "parser_logfmt", flb_test_parser_logfmt }, + { "parser_fallback", flb_test_parser_fallback }, + { "parser_exec_failure", flb_test_parser_exec_failure }, { NULL, NULL} };