diff --git a/plugins/out_chronicle/chronicle.c b/plugins/out_chronicle/chronicle.c index 10afd972d6f..b46f009f31b 100644 --- a/plugins/out_chronicle/chronicle.c +++ b/plugins/out_chronicle/chronicle.c @@ -533,9 +533,12 @@ static int flb_pack_msgpack_extract_log_key(void *out_context, uint64_t bytes, int ret; struct flb_chronicle *ctx = out_context; char *val_buf; + char *tmp_buf; char *key_str = NULL; size_t key_str_size = 0; size_t msgpack_size = bytes + bytes / 4; + size_t new_size; + size_t avail; size_t val_offset = 0; flb_sds_t out_buf; msgpack_object map; @@ -601,13 +604,33 @@ static int flb_pack_msgpack_extract_log_key(void *out_context, uint64_t bytes, val_offset++; } else { - ret = flb_msgpack_to_json(val_buf + val_offset, - msgpack_size - val_offset, &val, - config->json_escape_unicode); - if (ret <= 0) { - flb_plg_error(ctx->ins, "Could not convert log_key value to JSON"); - flb_free(val_buf); - return CHRONICLE_LOG_KEY_ERROR; + /* + * Serialize the value as JSON, growing the scratch buffer + * and retrying if it does not fit, so an oversized value + * is neither truncated nor dropped. + */ + ret = -1; + while (1) { + avail = msgpack_size - val_offset; + if (avail > 1) { + ret = flb_msgpack_to_json(val_buf + val_offset, + avail, &val, + config->json_escape_unicode); + if (ret > 0) { + break; + } + } + new_size = msgpack_size * 2; + tmp_buf = flb_realloc(val_buf, new_size); + if (tmp_buf == NULL) { + flb_errno(); + flb_plg_error(ctx->ins, "Could not grow buffer to " + "convert log_key value to JSON"); + flb_free(val_buf); + return CHRONICLE_LOG_KEY_ERROR; + } + val_buf = tmp_buf; + msgpack_size = new_size; } val_offset += ret; val_buf[val_offset] = '\0'; diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index e2c730d20b5..8f8f3f4968d 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -4475,9 +4475,12 @@ static flb_sds_t flb_pack_msgpack_extract_log_key(void *out_context, const char int alloc_error = 0; struct flb_s3 *ctx = out_context; char *val_buf; + char *tmp_buf; char *key_str = NULL; size_t key_str_size = 0; size_t msgpack_size = bytes + bytes / 4; + size_t new_size; + size_t avail; size_t val_offset = 0; flb_sds_t out_buf; msgpack_object map; @@ -4570,10 +4573,33 @@ static flb_sds_t flb_pack_msgpack_extract_log_key(void *out_context, const char val_offset++; } else { - ret = flb_msgpack_to_json(val_buf + val_offset, - msgpack_size - val_offset, &val, - config->json_escape_unicode); - if (ret < 0) { + /* + * Serialize the value as JSON, growing the scratch + * buffer and retrying if it does not fit, so an + * oversized value is neither truncated nor dropped. + */ + ret = -1; + while (1) { + avail = msgpack_size - val_offset; + if (avail > 1) { + ret = flb_msgpack_to_json(val_buf + val_offset, + avail, &val, + config->json_escape_unicode); + if (ret > 0) { + break; + } + } + new_size = msgpack_size * 2; + tmp_buf = flb_realloc(val_buf, new_size); + if (tmp_buf == NULL) { + flb_errno(); + alloc_error = 1; + break; + } + val_buf = tmp_buf; + msgpack_size = new_size; + } + if (alloc_error == 1) { break; } val_offset += ret; @@ -4600,6 +4626,15 @@ static flb_sds_t flb_pack_msgpack_extract_log_key(void *out_context, const char flb_log_event_decoder_destroy(&log_decoder); + /* + * If the scratch buffer could not be grown mid-chunk, do not return a + * partial payload: fail so the chunk can be retried as a whole. + */ + if (alloc_error == 1) { + flb_free(val_buf); + return NULL; + } + /* If nothing was read, destroy buffer */ if (val_offset == 0) { flb_free(val_buf); diff --git a/tests/runtime/out_chronicle.c b/tests/runtime/out_chronicle.c index ddfd54412a0..c6cf6ff64e4 100644 --- a/tests/runtime/out_chronicle.c +++ b/tests/runtime/out_chronicle.c @@ -175,15 +175,45 @@ static void cb_check_format_multiple_records(void *ctx, int ffd, flb_sds_destroy(res_data); } -static void cb_check_format_log_key_error(void *ctx, int ffd, - int res_ret, void *res_data, - size_t res_size, void *data) +static void cb_check_format_log_key_oversized(void *ctx, int ffd, + int res_ret, void *res_data, + size_t res_size, void *data) { - if (res_ret == 0) { - set_callback_error("log_key conversion failure was not propagated"); + char *out_json = res_data; + char *p; + int count = 0; + + /* + * A log_key value whose JSON form exceeds the buffer initially sized + * from the MessagePack record must now be serialized in full: the + * extraction buffer grows and retries instead of dropping or + * truncating the value. + */ + if (res_ret != 0 || out_json == NULL) { + set_callback_error("oversized log_key value should be serialized, not dropped"); + flb_sds_destroy(res_data); + return; + } + + if (strstr(out_json, "record one") == NULL) { + set_callback_error("first record was not found"); + } + if (strstr(out_json, "record three") == NULL) { + set_callback_error("third record was not found"); + } + + /* + * The middle record embeds 256 escaped control characters; every one + * must survive, proving the oversized value was neither truncated nor + * dropped. + */ + p = out_json; + while ((p = strstr(p, "u0001")) != NULL) { + count++; + p += 5; } - if (res_data != NULL || res_size != 0) { - set_callback_error("failed extraction returned a partial payload"); + if (count < 256) { + set_callback_error("oversized middle record value was truncated or dropped"); } increment_output_invoked(); @@ -511,7 +541,7 @@ void test_format_multiple_records() stop_and_check(ctx, 1); } -void test_format_with_log_key_conversion_error() +void test_format_with_log_key_oversized_value() { flb_ctx_t *ctx; int in_ffd, out_ffd; @@ -536,7 +566,7 @@ void test_format_with_log_key_conversion_error() "namespace", "tenant-a", "label", "env production", NULL); - flb_output_set_test(ctx, out_ffd, "formatter", cb_check_format_log_key_error, NULL, NULL); + flb_output_set_test(ctx, out_ffd, "formatter", cb_check_format_log_key_oversized, NULL, NULL); flb_start(ctx); clear_output_invoked(); @@ -551,7 +581,10 @@ void test_format_with_log_key_conversion_error() snprintf(records + offset, sizeof(records) - offset, "\"]}][3, {\"message\": \"record three\"}]"); - /* The failing middle record must abort the batch, preserving it for retry. */ + /* + * The oversized middle record must be serialized in full alongside its + * neighbours: the extraction buffer grows instead of dropping the value. + */ ret = flb_lib_push(ctx, in_ffd, records, strlen(records)); TEST_CHECK(ret == strlen(records)); @@ -733,7 +766,7 @@ TEST_LIST = { { "format_no_log_key", test_format_no_log_key }, { "format_with_log_key_found", test_format_with_log_key_found }, { "format_with_log_key_not_found", test_format_with_log_key_not_found }, - { "format_with_log_key_conversion_error", test_format_with_log_key_conversion_error }, + { "format_with_log_key_oversized_value", test_format_with_log_key_oversized_value }, { "format_multiple_records", test_format_multiple_records }, { "format_partially_suceeded_records", test_format_partially_suceeded_records }, { "format_namespace_and_labels", test_format_namespace_and_labels },