From e44f1225ef7a47ffdfd4fc26be6812ba1cfded84 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Sat, 17 Oct 2026 23:21:58 +0900 Subject: [PATCH 1/9] log_event_encoder: bound serialized output size Signed-off-by: Hiroshi Hatake --- include/fluent-bit/flb_log_event_encoder.h | 2 ++ src/flb_log_event_encoder.c | 32 ++++++++++++++++++---- 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/include/fluent-bit/flb_log_event_encoder.h b/include/fluent-bit/flb_log_event_encoder.h index 306911373bb..d893e647293 100644 --- a/include/fluent-bit/flb_log_event_encoder.h +++ b/include/fluent-bit/flb_log_event_encoder.h @@ -234,6 +234,8 @@ struct flb_log_event_encoder { int dynamically_allocated; char *output_buffer; size_t output_length; + /* Maximum serialized output bytes; SIZE_MAX by default. */ + size_t output_limit; int initialized; struct flb_time timestamp; diff --git a/src/flb_log_event_encoder.c b/src/flb_log_event_encoder.c index 9bb1968373e..caeb8d98fb1 100644 --- a/src/flb_log_event_encoder.c +++ b/src/flb_log_event_encoder.c @@ -59,6 +59,21 @@ void flb_log_event_encoder_reset(struct flb_log_event_encoder *context) flb_log_event_encoder_update_internal_state(context); } +static int encoder_buffer_write(void *data, const char *buffer, size_t length) +{ + struct flb_log_event_encoder *context; + + context = (struct flb_log_event_encoder *) + ((char *) data - offsetof(struct flb_log_event_encoder, buffer)); + + if (context->buffer.size > context->output_limit || + length > context->output_limit - context->buffer.size) { + return -1; + } + + return msgpack_sbuffer_write(&context->buffer, buffer, length); +} + int flb_log_event_encoder_init(struct flb_log_event_encoder *context, int format) { if (context == NULL) { @@ -75,11 +90,12 @@ int flb_log_event_encoder_init(struct flb_log_event_encoder *context, int format context->dynamically_allocated = FLB_FALSE; context->initialized = FLB_TRUE; context->format = format; + context->output_limit = SIZE_MAX; msgpack_sbuffer_init(&context->buffer); msgpack_packer_init(&context->packer, &context->buffer, - msgpack_sbuffer_write); + encoder_buffer_write); flb_log_event_encoder_dynamic_field_init(&context->metadata, MSGPACK_OBJECT_MAP); @@ -174,6 +190,7 @@ int flb_log_event_encoder_emit_record(struct flb_log_event_encoder *context) int result; char ts_buf[8]; char rec_header[5]; + size_t original_size; uint32_t sec; uint32_t nsec; @@ -181,6 +198,7 @@ int flb_log_event_encoder_emit_record(struct flb_log_event_encoder *context) return FLB_EVENT_ENCODER_ERROR_INVALID_CONTEXT; } + original_size = context->buffer.size; result = FLB_EVENT_ENCODER_SUCCESS; if (context->root.size == 0) { @@ -290,6 +308,10 @@ int flb_log_event_encoder_emit_record(struct flb_log_event_encoder *context) } } + if (result != FLB_EVENT_ENCODER_SUCCESS) { + context->buffer.size = original_size; + } + flb_log_event_encoder_update_internal_state(context); flb_log_event_encoder_reset_record(context); @@ -487,8 +509,8 @@ int flb_log_event_encoder_group_header_end(struct flb_log_event_encoder *context int ret; ret = flb_log_event_encoder_commit_record(context); - if (ret == -1) { - return -1; + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + return ret; } return 0; @@ -511,8 +533,8 @@ int flb_log_event_encoder_group_end(struct flb_log_event_encoder *context) } ret = flb_log_event_encoder_commit_record(context); - if (ret == -1) { - return -1; + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + return ret; } return 0; From c75b0b5e60002826d5385a4ce2536f158da57634 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Sat, 17 Oct 2026 23:22:03 +0900 Subject: [PATCH 2/9] opentelemetry: honor log encoder output limits Signed-off-by: Hiroshi Hatake --- src/opentelemetry/flb_opentelemetry_logs.c | 67 ++++++++++++++++++---- 1 file changed, 55 insertions(+), 12 deletions(-) diff --git a/src/opentelemetry/flb_opentelemetry_logs.c b/src/opentelemetry/flb_opentelemetry_logs.c index 63687cc833b..8631f0d01e0 100644 --- a/src/opentelemetry/flb_opentelemetry_logs.c +++ b/src/opentelemetry/flb_opentelemetry_logs.c @@ -452,7 +452,7 @@ static int process_json_payload_log_records_entry( if (result == FLB_EVENT_ENCODER_SUCCESS) { result = flb_log_event_encoder_commit_record(encoder); } - else { + if (result != FLB_EVENT_ENCODER_SUCCESS) { if (error_status) { *error_status = FLB_OTEL_LOGS_ERR_ENCODER_FAILURE; } @@ -516,8 +516,10 @@ static int process_json_payload_scope_logs_entry( sizeof(FLB_OTEL_LOGS_METADATA_KEY) - 1, logs_body_key, &entry_status); - if (result < 0 && error_status) { - *error_status = entry_status; + if (result < 0) { + if (error_status) { + *error_status = entry_status; + } return result; } } @@ -629,6 +631,22 @@ static int process_json_payload_resource_logs_entry (struct flb_log_event_encode return -FLB_OTEL_LOGS_ERR_UNEXPECTED_SCOPELOGS_TYPE; } + /* Empty scopes must not repeatedly serialize discarded resource metadata. */ + result = flb_otel_utils_find_map_entry_by_key(&scope_logs->ptr[index].via.map, + "logRecords", 0, FLB_TRUE); + if (result < 0) { + if (error_status) { + *error_status = FLB_OTEL_LOGS_ERR_EMPTY_PAYLOAD; + } + result = 0; + continue; + } + obj = &scope_logs->ptr[index].via.map.ptr[result].val; + if (obj->type == MSGPACK_OBJECT_ARRAY && obj->via.array.size == 0) { + result = 0; + continue; + } + /* * we use a temporary encoder to hold the group information, if no record entries are added * we will discard it. @@ -640,6 +658,7 @@ static int process_json_payload_resource_logs_entry (struct flb_log_event_encode } return -FLB_OTEL_LOGS_ERR_ENCODER_FAILURE; } + tmp_encoder->output_limit = encoder->output_limit - encoder->buffer.size; flb_log_event_encoder_group_init(tmp_encoder); /* pack internal schema */ @@ -794,7 +813,14 @@ static int process_json_payload_resource_logs_entry (struct flb_log_event_encode /* close scope map */ flb_log_event_encoder_commit_map(tmp_encoder, FLB_LOG_EVENT_BODY); } - flb_log_event_encoder_group_header_end(tmp_encoder); + ret = flb_log_event_encoder_group_header_end(tmp_encoder); + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + if (error_status) { + *error_status = FLB_OTEL_LOGS_ERR_ENCODER_FAILURE; + } + flb_log_event_encoder_destroy(tmp_encoder); + return -FLB_OTEL_LOGS_ERR_ENCODER_FAILURE; + } /* before processing the scope logs, grab the number of bytes written */ size_before = tmp_encoder->buffer.size; @@ -817,13 +843,26 @@ static int process_json_payload_resource_logs_entry (struct flb_log_event_encode * the main encoder */ if (size_after > size_before) { - flb_log_event_encoder_group_end(tmp_encoder); + ret = flb_log_event_encoder_group_end(tmp_encoder); + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + if (error_status) { + *error_status = FLB_OTEL_LOGS_ERR_ENCODER_FAILURE; + } + flb_log_event_encoder_destroy(tmp_encoder); + return -FLB_OTEL_LOGS_ERR_ENCODER_FAILURE; + } /* Append the temporary encoder output to the main encoder */ - msgpack_sbuffer_write(&encoder->buffer, - tmp_encoder->output_buffer, - tmp_encoder->output_length); - + ret = msgpack_pack_str_body(&encoder->packer, + tmp_encoder->output_buffer, + tmp_encoder->output_length); + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + if (error_status) { + *error_status = FLB_OTEL_LOGS_ERR_ENCODER_FAILURE; + } + flb_log_event_encoder_destroy(tmp_encoder); + return -FLB_OTEL_LOGS_ERR_ENCODER_FAILURE; + } encoder->output_buffer = encoder->buffer.data; encoder->output_length = encoder->buffer.size; } @@ -932,6 +971,9 @@ int flb_opentelemetry_logs_json_to_msgpack(struct flb_log_event_encoder *encoder if (encoder == NULL) { return -1; } + if (encoder->buffer.size > encoder->output_limit) { + return -FLB_OTEL_LOGS_ERR_ENCODER_FAILURE; + } /* Convert JSON to messagepack */ result = flb_pack_json(body, len, &msgpack_body, &msgpack_body_length, @@ -947,6 +989,7 @@ int flb_opentelemetry_logs_json_to_msgpack(struct flb_log_event_encoder *encoder flb_free(msgpack_body); return -1; } + local_log_encoder.output_limit = encoder->output_limit - encoder->buffer.size; offset = 0; msgpack_unpacked_init(&unpacked_root); @@ -983,9 +1026,9 @@ int flb_opentelemetry_logs_json_to_msgpack(struct flb_log_event_encoder *encoder /* copy local buffer into caller encoder buffer */ if (local_log_encoder.output_length > 0) { - msgpack_sbuffer_write(&encoder->buffer, - local_log_encoder.output_buffer, - local_log_encoder.output_length); + result = msgpack_pack_str_body(&encoder->packer, + local_log_encoder.output_buffer, + local_log_encoder.output_length); encoder->output_buffer = encoder->buffer.data; encoder->output_length = encoder->buffer.size; } From 8fff3d1013f7ec9ab93801101024a424d86e412c Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Sat, 17 Oct 2026 23:22:07 +0900 Subject: [PATCH 3/9] in_opentelemetry: limit generated log bytes per request Signed-off-by: Hiroshi Hatake --- plugins/in_opentelemetry/opentelemetry.c | 18 ++++++++++++ plugins/in_opentelemetry/opentelemetry.h | 1 + plugins/in_opentelemetry/opentelemetry_logs.c | 28 +++++++++++++++++-- plugins/in_opentelemetry/opentelemetry_logs.h | 6 ++++ plugins/in_opentelemetry/opentelemetry_prot.c | 7 +++-- 5 files changed, 56 insertions(+), 4 deletions(-) diff --git a/plugins/in_opentelemetry/opentelemetry.c b/plugins/in_opentelemetry/opentelemetry.c index 6ff23d5d66f..8b7d1fbf010 100644 --- a/plugins/in_opentelemetry/opentelemetry.c +++ b/plugins/in_opentelemetry/opentelemetry.c @@ -29,6 +29,8 @@ static int in_opentelemetry_init(struct flb_input_instance *ins, struct flb_config *config, void *data) { int ret; + int64_t logs_max_size_bytes; + const char *logs_max_size; struct flb_opentelemetry *ctx; struct flb_http_server_options http_server_options; @@ -49,6 +51,17 @@ static int in_opentelemetry_init(struct flb_input_instance *ins, return -1; } + logs_max_size = flb_input_get_property("logs_max_size", ins); + logs_max_size_bytes = ctx->logs_max_size; + if (logs_max_size != NULL) { + logs_max_size_bytes = flb_utils_size_to_bytes(logs_max_size); + } + if (logs_max_size_bytes <= 0 || (uint64_t) logs_max_size_bytes > SIZE_MAX) { + flb_plg_error(ctx->ins, "logs_max_size must be a valid positive size"); + opentelemetry_config_destroy(ctx); + return -1; + } + if (ctx->oauth2_cfg.validate) { if (!ctx->oauth2_cfg.issuer || !ctx->oauth2_cfg.jwks_url) { flb_plg_error(ctx->ins, "oauth2.issuer and oauth2.jwks_url are required when oauth2.validate is enabled"); @@ -205,6 +218,11 @@ static struct flb_config_map config_map[] = { "Forward traces without processing" }, + { + FLB_CONFIG_MAP_SIZE, "logs_max_size", "4M", + 0, FLB_TRUE, offsetof(struct flb_opentelemetry, logs_max_size), + "Maximum generated MessagePack bytes per logs request, including group metadata" + }, { FLB_CONFIG_MAP_STR, "logs_metadata_key", "otlp", 0, FLB_TRUE, offsetof(struct flb_opentelemetry, logs_metadata_key), diff --git a/plugins/in_opentelemetry/opentelemetry.h b/plugins/in_opentelemetry/opentelemetry.h index c82d60b1804..b3834416dd0 100644 --- a/plugins/in_opentelemetry/opentelemetry.h +++ b/plugins/in_opentelemetry/opentelemetry.h @@ -39,6 +39,7 @@ struct flb_opentelemetry { int tag_from_uri; flb_sds_t logs_metadata_key; flb_sds_t logs_body_key; + size_t logs_max_size; int profile_support_enabled; int encode_profiles_as_log; struct flb_oauth2_jwt_cfg oauth2_cfg; diff --git a/plugins/in_opentelemetry/opentelemetry_logs.c b/plugins/in_opentelemetry/opentelemetry_logs.c index 9d2db14a734..52dc92dab38 100644 --- a/plugins/in_opentelemetry/opentelemetry_logs.c +++ b/plugins/in_opentelemetry/opentelemetry_logs.c @@ -30,6 +30,7 @@ #include "opentelemetry_protobuf.h" #include "opentelemetry.h" +#include "opentelemetry_logs.h" #include "opentelemetry_utils.h" /* @@ -655,7 +656,10 @@ static int binary_payload_to_msgpack(struct flb_opentelemetry *ctx, goto binary_payload_to_msgpack_end; } - flb_log_event_encoder_group_header_end(encoder); + ret = flb_log_event_encoder_group_header_end(encoder); + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + goto binary_payload_to_msgpack_end; + } for (log_record_index=0; log_record_index < scope_log->n_log_records; log_record_index++) { ret = flb_log_event_encoder_begin_record(encoder); @@ -757,6 +761,9 @@ static int binary_payload_to_msgpack(struct flb_opentelemetry *ctx, if (ret == FLB_EVENT_ENCODER_SUCCESS) { (*record_count)++; } + else { + goto binary_payload_to_msgpack_end; + } } else { flb_plg_error(ctx->ins, "marshalling error"); @@ -764,7 +771,10 @@ static int binary_payload_to_msgpack(struct flb_opentelemetry *ctx, } } - flb_log_event_encoder_group_end(encoder); + ret = flb_log_event_encoder_group_end(encoder); + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + goto binary_payload_to_msgpack_end; + } } } @@ -791,6 +801,18 @@ int opentelemetry_process_logs(struct flb_opentelemetry *ctx, flb_sds_t tag, size_t tag_len, void *data, size_t size) +{ + size_t remaining = ctx->logs_max_size; + + return opentelemetry_process_logs_with_budget(ctx, content_type, tag, tag_len, + data, size, &remaining); +} + +int opentelemetry_process_logs_with_budget(struct flb_opentelemetry *ctx, + flb_sds_t content_type, + flb_sds_t tag, size_t tag_len, + void *data, size_t size, + size_t *remaining) { int ret = -1; int is_proto = FLB_FALSE; /* default to JSON */ @@ -828,6 +850,7 @@ int opentelemetry_process_logs(struct flb_opentelemetry *ctx, if (encoder == NULL) { return -1; } + encoder->output_limit = *remaining; if (is_proto == FLB_TRUE) { ret = binary_payload_to_msgpack(ctx, encoder, @@ -853,6 +876,7 @@ int opentelemetry_process_logs(struct flb_opentelemetry *ctx, } if (ret >= 0) { + *remaining -= encoder->output_length; if (opentelemetry_uses_worker_ingress_queue(ctx)) { size_t allocation_size; void *resized_buffer; diff --git a/plugins/in_opentelemetry/opentelemetry_logs.h b/plugins/in_opentelemetry/opentelemetry_logs.h index b36eef7676a..470771c7249 100644 --- a/plugins/in_opentelemetry/opentelemetry_logs.h +++ b/plugins/in_opentelemetry/opentelemetry_logs.h @@ -28,4 +28,10 @@ int opentelemetry_process_logs(struct flb_opentelemetry *ctx, size_t tag_len, void *data, size_t size); +int opentelemetry_process_logs_with_budget(struct flb_opentelemetry *ctx, + flb_sds_t content_type, + flb_sds_t tag, size_t tag_len, + void *data, size_t size, + size_t *remaining); + #endif diff --git a/plugins/in_opentelemetry/opentelemetry_prot.c b/plugins/in_opentelemetry/opentelemetry_prot.c index acad5d43d2e..b2d90e2dbcf 100644 --- a/plugins/in_opentelemetry/opentelemetry_prot.c +++ b/plugins/in_opentelemetry/opentelemetry_prot.c @@ -838,12 +838,14 @@ int opentelemetry_prot_handle_ng(struct flb_http_request *request, char payload_type; char *payload = NULL; size_t payload_size = 0; + size_t logs_remaining; struct flb_opentelemetry *context; struct cfl_list frames; struct otlp_grpc_frame *frame; cfl_list_init(&frames); context = (struct flb_opentelemetry *) response->stream->user_data; + logs_remaining = context->logs_max_size; if (request->path == NULL || request->path[0] != '/') { send_response_ng(response, 400, "error: invalid request\n"); @@ -1018,8 +1020,9 @@ int opentelemetry_prot_handle_ng(struct flb_http_request *request, goto cleanup; } - ret = opentelemetry_process_logs(context, request->content_type, tag, flb_sds_len(tag), - payload, payload_size); + ret = opentelemetry_process_logs_with_budget(context, request->content_type, + tag, flb_sds_len(tag), + payload, payload_size, &logs_remaining); } else if (context->profile_support_enabled && is_profiles_export_path(request->path) == FLB_TRUE) { From 335da450fc89fca4cc75851524c5038fbb8dca45 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Sat, 17 Oct 2026 23:22:11 +0900 Subject: [PATCH 4/9] tests: internal: cover bounded log event encoding Signed-off-by: Hiroshi Hatake --- tests/internal/log_event_encoder.c | 67 ++++++++++++++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/tests/internal/log_event_encoder.c b/tests/internal/log_event_encoder.c index 1b367ee0577..468f998c78c 100644 --- a/tests/internal/log_event_encoder.c +++ b/tests/internal/log_event_encoder.c @@ -834,7 +834,74 @@ static void timestamp_encoding() flb_log_event_encoder_destroy(&encoder); } +static void output_limit(void) +{ + struct flb_log_event_encoder encoder; + int ret; + + TEST_CHECK(flb_log_event_encoder_init(&encoder, FLB_LOG_EVENT_FORMAT_FLUENT_BIT_V2) == 0); + TEST_CHECK(encoder.output_limit == SIZE_MAX); + encoder.output_limit = 3; + TEST_CHECK(flb_log_event_encoder_emit_raw_record(&encoder, "abc", 3) == 0); + TEST_CHECK(encoder.output_length == 3); + TEST_CHECK(flb_log_event_encoder_emit_raw_record(&encoder, "d", 1) < 0); + TEST_CHECK(encoder.output_length == 3); + + flb_log_event_encoder_reset(&encoder); + TEST_CHECK(encoder.output_limit == 3); + TEST_CHECK(flb_log_event_encoder_group_init(&encoder) == 0); + ret = flb_log_event_encoder_group_header_end(&encoder); + TEST_CHECK(ret < 0); + TEST_CHECK(encoder.output_length <= encoder.output_limit); + + flb_log_event_encoder_reset(&encoder); + TEST_CHECK(flb_log_event_encoder_group_end(&encoder) < 0); + TEST_CHECK(encoder.output_length <= encoder.output_limit); + + flb_log_event_encoder_reset(&encoder); + encoder.output_limit = 0; + TEST_CHECK(flb_log_event_encoder_emit_raw_record(&encoder, "a", 1) < 0); + TEST_CHECK(encoder.output_length == 0); + encoder.output_limit = SIZE_MAX; + TEST_CHECK(flb_log_event_encoder_group_init(&encoder) == 0); + TEST_CHECK(flb_log_event_encoder_group_header_end(&encoder) == 0); + TEST_CHECK(flb_log_event_encoder_group_end(&encoder) == 0); + flb_log_event_encoder_destroy(&encoder); +} + +static void output_limit_record_rollback(void) +{ + struct flb_log_event_encoder encoder; + size_t record_size; + size_t index; + int formats[] = {FLB_LOG_EVENT_FORMAT_FORWARD, FLB_LOG_EVENT_FORMAT_FLUENT_BIT_V2}; + + for (index = 0; index < sizeof(formats) / sizeof(formats[0]); index++) { + TEST_CHECK(flb_log_event_encoder_init(&encoder, formats[index]) == 0); + TEST_CHECK(flb_log_event_encoder_begin_record(&encoder) == 0); + TEST_CHECK(flb_log_event_encoder_commit_record(&encoder) == 0); + record_size = encoder.output_length; + flb_log_event_encoder_reset(&encoder); + + TEST_CHECK(flb_log_event_encoder_emit_raw_record(&encoder, "\x01", 1) == 0); + encoder.output_limit = record_size; + TEST_CHECK(flb_log_event_encoder_begin_record(&encoder) == 0); + TEST_CHECK(flb_log_event_encoder_commit_record(&encoder) < 0); + TEST_CHECK(encoder.output_length == 1); + TEST_CHECK(encoder.buffer.size == 1); + TEST_CHECK(encoder.output_buffer[0] == '\x01'); + + encoder.output_limit = SIZE_MAX; + TEST_CHECK(flb_log_event_encoder_begin_record(&encoder) == 0); + TEST_CHECK(flb_log_event_encoder_commit_record(&encoder) == 0); + TEST_CHECK(encoder.output_length == 1 + record_size); + flb_log_event_encoder_destroy(&encoder); + } +} + TEST_LIST = { + { "output_limit_record_rollback", output_limit_record_rollback}, + { "output_limit", output_limit}, { "basic_format_fluent_bit_v2", basic_format_fluent_bit_v2}, { "basic_format_fluent_bit_v1", basic_format_fluent_bit_v1}, { "basic_metadata_format_fluent_bit_v2", basic_metadata_format_fluent_bit_v2}, From 44324f1c9e087d3d2e917755c8ed6bcb76d0ad4a Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Sat, 17 Oct 2026 23:22:16 +0900 Subject: [PATCH 5/9] tests: internal: cover bounded OTLP JSON conversion Signed-off-by: Hiroshi Hatake --- tests/internal/opentelemetry.c | 39 ++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/tests/internal/opentelemetry.c b/tests/internal/opentelemetry.c index 0967329a861..e2d6afbd241 100644 --- a/tests/internal/opentelemetry.c +++ b/tests/internal/opentelemetry.c @@ -3533,7 +3533,46 @@ void test_opentelemetry_metrics_cutoff() } /* Test list */ +static void test_opentelemetry_logs_generated_limit(void) +{ + struct flb_log_event_encoder encoder; + const char *payload = "{\"resourceLogs\":[{\"scopeLogs\":[{\"logRecords\":[{},{}]}]}]}"; + const char *empty = "{\"resourceLogs\":[{\"resource\":{\"attributes\":[" + "{\"key\":\"k\",\"value\":{\"stringValue\":\"v\"}}]}," + "\"scopeLogs\":[{},{\"logRecords\":[]}]}]}"; + size_t length; + int ret; + + TEST_CHECK(flb_log_event_encoder_init(&encoder, FLB_LOG_EVENT_FORMAT_FLUENT_BIT_V2) == 0); + TEST_CHECK(flb_opentelemetry_logs_json_to_msgpack(&encoder, payload, strlen(payload), + NULL, NULL) == 0); + length = encoder.output_length; + TEST_CHECK(length > 0); + flb_log_event_encoder_reset(&encoder); + encoder.output_limit = length; + TEST_CHECK(flb_opentelemetry_logs_json_to_msgpack(&encoder, payload, strlen(payload), + NULL, NULL) == 0); + TEST_CHECK(encoder.output_length == length); + + /* Existing caller output counts against the budget and survives rejection. */ + ret = flb_opentelemetry_logs_json_to_msgpack(&encoder, payload, strlen(payload), NULL, NULL); + TEST_CHECK(ret < 0); + TEST_CHECK(encoder.output_length == length); + flb_log_event_encoder_reset(&encoder); + encoder.output_limit = length - 1; + TEST_CHECK(flb_opentelemetry_logs_json_to_msgpack(&encoder, payload, strlen(payload), + NULL, NULL) < 0); + TEST_CHECK(encoder.output_length == 0); + /* Empty scopes must succeed without serializing even a temporary group header. */ + encoder.output_limit = 0; + TEST_CHECK(flb_opentelemetry_logs_json_to_msgpack(&encoder, empty, strlen(empty), + NULL, NULL) == 0); + TEST_CHECK(encoder.output_length == 0); + flb_log_event_encoder_destroy(&encoder); +} + TEST_LIST = { + { "opentelemetry_logs_generated_limit", test_opentelemetry_logs_generated_limit }, { "hex_to_id", test_hex_to_id }, { "hex_to_id_error_cases", test_hex_to_id_error_cases }, { "convert_string_number_to_u64", test_convert_string_number_to_u64 }, From a1b7e8de5b52b75ef3b51cea1ebbae3d693abb77 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Sat, 17 Oct 2026 23:22:21 +0900 Subject: [PATCH 6/9] tests: integration: cover generated OTLP log limits Signed-off-by: Hiroshi Hatake --- .../tests/test_in_opentelemetry_001.py | 154 +++++++++++++++++- 1 file changed, 150 insertions(+), 4 deletions(-) diff --git a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py index 4c846a37465..24762259932 100644 --- a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py +++ b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py @@ -36,6 +36,7 @@ # local imports from utils.data_utils import read_json_file +from utils.fluent_bit_manager import FluentBitStartupError from utils.http_matrix import PROTOCOL_CASES, run_curl_request from utils.input_pause_resume import ( assert_pause_resume_cycles, @@ -174,7 +175,7 @@ def iter_spans(output): } -def read_stdout_otlp_json(service, root_key, timeout=10, interval=0.25): +def read_stdout_otlp_json(service, root_key, timeout=10, interval=0.25, min_entries=None): deadline = time.time() + timeout decoder = json.JSONDecoder() @@ -198,7 +199,11 @@ def read_stdout_otlp_json(service, root_key, timeout=10, interval=0.25): matches.append(payload) if matches: - return matches[-1] + if min_entries is None: + return matches[-1] + entries = [entry for match in matches for entry in match[root_key]] + if len(entries) >= min_entries: + return {root_key: entries} time.sleep(interval) @@ -463,6 +468,146 @@ def build_histogram_json_payload(explicit_bounds, name="histogram_bounds"): return json.dumps(payload).encode("utf-8") +def generated_limit_payload(attributes, scopes, records=1, resources=1): + request = ExportLogsServiceRequest() + for _ in range(resources): + resource = request.resource_logs.add() + for index in range(attributes): + attribute = resource.resource.attributes.add(key=f"key{index:04d}") + attribute.value.string_value = "v" * 64 + for _ in range(scopes): + scope = resource.scope_logs.add() + for _ in range(records): + scope.log_records.add().body.string_value = "generated-limit" + return request + + +def generated_limit_service(tmp_path, workers, limit="16k"): + config = tmp_path / "generated-limit.yaml" + limit_option = f"logs_max_size: {limit}" if limit else "" + config.write_text(f""" +service: + flush: 0.2 + grace: 1 + log_level: info + http_server: on + http_port: ${{FLUENT_BIT_HTTP_MONITORING_PORT}} +pipeline: + inputs: + - name: opentelemetry + port: ${{FLUENT_BIT_TEST_LISTENER_PORT}} + http2: on + http_server.workers: {workers} + {limit_option} + outputs: + - name: opentelemetry + match: '*' + host: 127.0.0.1 + port: ${{TEST_SUITE_HTTP_PORT}} +""", encoding="utf-8") + return Service(str(config)) + + +@pytest.mark.parametrize("workers", [1, pytest.param( + 2, marks=pytest.mark.skipif(os.name == "nt", + reason="HTTP shared listener ports are unavailable on Windows"))]) +@pytest.mark.parametrize("encoding", ["json", "protobuf"]) +@pytest.mark.parametrize("attributes,scopes,records,resources,accepted", [ + (1, 32, 1, 1, True), + (64, 1, 1, 1, True), + (64, 32, 1, 1, False), + (32, 64, 1, 1, False), + (64, 2, 1, 2, False), + (1, 1, 512, 1, False), + (64, 1000, 0, 1, False), +]) +def test_in_opentelemetry_generated_limit(tmp_path, workers, encoding, attributes, + scopes, records, resources, accepted): + service = generated_limit_service(tmp_path, workers) + request = generated_limit_payload(attributes, scopes, records, resources) + if encoding == "json": + payload = json_format.MessageToJson(request).encode() + content_type = "application/json" + else: + payload = request.SerializeToString() + content_type = "application/x-protobuf" + # All requests fit the wire limit; rejection must be due to generated bytes. + assert len(payload) < 4_000_000 + try: + service.start() + response = service.send_raw_request("/v1/logs", payload, content_type) + assert response.status_code == (201 if accepted else 400) + # Exercise cleanup and ensure a rejected request never appends its prefix. + small = generated_limit_payload(1, 1) + small.resource_logs[0].scope_logs[0].log_records[0].body.string_value = "recovery" + response = service.send_raw_request("/v1/logs", small.SerializeToString()) + assert response.status_code == 201 + + def received_records(): + return [record for request in data_storage["logs"] + for resource in request.resource_logs + for scope in resource.scope_logs for record in scope.log_records] + + expected = scopes * records * resources if accepted else 0 + service.service.wait_for_condition( + lambda: len(received_records()) >= expected + 1, + timeout=20, interval=0.1, description="recovery after generated byte limit") + actual = received_records() + assert len(actual) == expected + 1 + assert sum(record.body.string_value == "recovery" for record in actual) == 1 + assert sum(record.body.string_value == "generated-limit" for record in actual) == expected + finally: + service.stop() + + +@pytest.mark.parametrize("encoding", ["json", "protobuf"]) +def test_in_opentelemetry_generated_limit_default(tmp_path, encoding): + service = generated_limit_service(tmp_path, 1, limit=None) + request = generated_limit_payload(64, 1000) + if encoding == "json": + payload = json_format.MessageToJson(request).encode() + content_type = "application/json" + else: + payload = request.SerializeToString() + content_type = "application/x-protobuf" + assert len(payload) < 4_000_000 + try: + service.start() + assert service.send_raw_request("/v1/logs", payload, content_type).status_code == 400 + finally: + service.stop() + + +@pytest.mark.parametrize("limit", ["invalid", "-1", "-2M", "0"]) +def test_in_opentelemetry_generated_limit_invalid_config(tmp_path, limit): + service = generated_limit_service(tmp_path, 1, limit=limit) + try: + with pytest.raises(FluentBitStartupError): + service.start() + with open(service.service.flb.log_file, encoding="utf-8", errors="replace") as log: + assert "logs_max_size must be a valid positive size" in log.read() + finally: + service.stop() + + +def test_in_opentelemetry_generated_limit_grpc_request_budget(tmp_path): + service = generated_limit_service(tmp_path, 1) + payload = generated_limit_payload(64, 2).SerializeToString() + endpoint = "/opentelemetry.proto.collector.logs.v1.LogsService/Export" + try: + service.start() + with grpc.insecure_channel(f"127.0.0.1:{service.flb_listener_port}") as channel: + export = channel.stream_unary(endpoint) + # Each message fits alone; two exceed the shared HTTP request budget. + export(iter([payload]), timeout=10) + with pytest.raises(grpc.RpcError) as rejected: + export(iter([payload, payload]), timeout=10) + assert rejected.value.code() == grpc.StatusCode.UNKNOWN + export(iter([payload]), timeout=10) + finally: + service.stop() + + class Service: def __init__(self, config_file, *, use_auth_server=False): # Compose the absolute path for the Fluent Bit configuration file @@ -1156,7 +1301,8 @@ def test_in_opentelemetry_stdout_otlp_json_logs_preserve_resources_across_reques ) assert 200 <= response.status_code < 300 - output = read_stdout_otlp_json(service, "resourceLogs", timeout=10) + # The requests may land on opposite sides of a flush boundary. + output = read_stdout_otlp_json(service, "resourceLogs", timeout=10, min_entries=2) service.stop() records = list(iter_log_records(output)) @@ -1198,7 +1344,7 @@ def test_in_opentelemetry_stdout_otlp_json_logs_preserve_resource_schema_urls( ) assert 200 <= response.status_code < 300 - output = read_stdout_otlp_json(service, "resourceLogs", timeout=10) + output = read_stdout_otlp_json(service, "resourceLogs", timeout=10, min_entries=2) service.stop() body_to_schema_url = { From 322f89764a4bafbe6a14c106f64d626dbcf9c7b8 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 19 Oct 2026 11:10:21 +0900 Subject: [PATCH 7/9] tests: internal: initialize OTEL router encoders Signed-off-by: Hiroshi Hatake --- tests/internal/router_config.c | 16 +++++----------- 1 file changed, 5 insertions(+), 11 deletions(-) diff --git a/tests/internal/router_config.c b/tests/internal/router_config.c index fb1ae28b4a8..0fb0788fb02 100644 --- a/tests/internal/router_config.c +++ b/tests/internal/router_config.c @@ -265,6 +265,11 @@ static int build_log_chunk_with_otel(const char *service_name, return -1; } + ret = flb_log_event_encoder_init(encoder, FLB_LOG_EVENT_FORMAT_FLUENT_BIT_V2); + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + return -1; + } + /* Extract attribute key from "scope.attr" format if needed */ if (strncmp(scope_attribute_key, "scope.", 6) == 0) { attr_key = scope_attribute_key + 6; /* Skip "scope." prefix */ @@ -343,16 +348,11 @@ static int build_log_chunk_with_otel(const char *service_name, "}", service_name, scope_name, scope_version, attr_key, scope_attribute_value); - /* Initialize encoder buffer (needed for msgpack_sbuffer_write) */ - memset(encoder, 0, sizeof(*encoder)); - msgpack_sbuffer_init(&encoder->buffer); - /* Convert OTLP JSON to msgpack using the actual converter */ ret = flb_opentelemetry_logs_json_to_msgpack(encoder, otlp_json, json_len, NULL, &error_status); flb_free(otlp_json); if (ret != 0) { - msgpack_sbuffer_destroy(&encoder->buffer); return -1; } @@ -2188,7 +2188,6 @@ static void test_router_condition_eval_logs_otel_contexts() TEST_CHECK(flb_condition_eval_logs(&chunk, &context, &route) == FLB_TRUE); } flb_router_chunk_context_reset(&context); - msgpack_sbuffer_destroy(&encoder.buffer); flb_log_event_encoder_destroy(&encoder); ret = build_log_chunk_with_otel("api", "demo", "1.0.0", @@ -2199,7 +2198,6 @@ static void test_router_condition_eval_logs_otel_contexts() TEST_CHECK(flb_condition_eval_logs(&chunk, &context, &route) == FLB_FALSE); } flb_router_chunk_context_reset(&context); - msgpack_sbuffer_destroy(&encoder.buffer); flb_log_event_encoder_destroy(&encoder); free_route_condition(condition); @@ -2265,7 +2263,6 @@ static void test_router_condition_eval_logs_otel_contexts() TEST_CHECK(flb_condition_eval_logs(&chunk, &context, &route) == FLB_TRUE); } flb_router_chunk_context_reset(&context); - msgpack_sbuffer_destroy(&encoder.buffer); flb_log_event_encoder_destroy(&encoder); ret = build_log_chunk_with_otel("backend", "other", "1.0.0", @@ -2276,7 +2273,6 @@ static void test_router_condition_eval_logs_otel_contexts() TEST_CHECK(flb_condition_eval_logs(&chunk, &context, &route) == FLB_FALSE); } flb_router_chunk_context_reset(&context); - msgpack_sbuffer_destroy(&encoder.buffer); flb_log_event_encoder_destroy(&encoder); free_route_condition(condition); @@ -2342,7 +2338,6 @@ static void test_router_condition_eval_logs_otel_contexts() TEST_CHECK(flb_condition_eval_logs(&chunk, &context, &route) == FLB_TRUE); } flb_router_chunk_context_reset(&context); - msgpack_sbuffer_destroy(&encoder.buffer); flb_log_event_encoder_destroy(&encoder); ret = build_log_chunk_with_otel("backend", "demo", "1.0.0", @@ -2353,7 +2348,6 @@ static void test_router_condition_eval_logs_otel_contexts() TEST_CHECK(flb_condition_eval_logs(&chunk, &context, &route) == FLB_FALSE); } flb_router_chunk_context_reset(&context); - msgpack_sbuffer_destroy(&encoder.buffer); flb_log_event_encoder_destroy(&encoder); free_route_condition(condition); From 9ff07ef7f291b42c1c4d7bc9fd7db7d3e9469abb Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Tue, 20 Oct 2026 21:18:22 +0900 Subject: [PATCH 8/9] in_opentelemetry: raise generated log size default to 100M Signed-off-by: Hiroshi Hatake --- plugins/in_opentelemetry/opentelemetry.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/in_opentelemetry/opentelemetry.c b/plugins/in_opentelemetry/opentelemetry.c index 8b7d1fbf010..1df6fe06b43 100644 --- a/plugins/in_opentelemetry/opentelemetry.c +++ b/plugins/in_opentelemetry/opentelemetry.c @@ -219,7 +219,7 @@ static struct flb_config_map config_map[] = { }, { - FLB_CONFIG_MAP_SIZE, "logs_max_size", "4M", + FLB_CONFIG_MAP_SIZE, "logs_max_size", "100M", 0, FLB_TRUE, offsetof(struct flb_opentelemetry, logs_max_size), "Maximum generated MessagePack bytes per logs request, including group metadata" }, From a4b269f7cc37466bbf063691be652b72297b398d Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Tue, 20 Oct 2026 21:18:23 +0900 Subject: [PATCH 9/9] tests: integration: verify 100M generated log default Signed-off-by: Hiroshi Hatake --- .../tests/test_in_opentelemetry_001.py | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py index 24762259932..4a1d824b442 100644 --- a/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py +++ b/tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py @@ -561,9 +561,14 @@ def received_records(): @pytest.mark.parametrize("encoding", ["json", "protobuf"]) -def test_in_opentelemetry_generated_limit_default(tmp_path, encoding): +@pytest.mark.parametrize("attributes,scopes,expected_status", [ + (64, 1000, 201), + (512, 3000, 400), +]) +def test_in_opentelemetry_generated_limit_default(tmp_path, encoding, attributes, + scopes, expected_status): service = generated_limit_service(tmp_path, 1, limit=None) - request = generated_limit_payload(64, 1000) + request = generated_limit_payload(attributes, scopes) if encoding == "json": payload = json_format.MessageToJson(request).encode() content_type = "application/json" @@ -571,9 +576,17 @@ def test_in_opentelemetry_generated_limit_default(tmp_path, encoding): payload = request.SerializeToString() content_type = "application/x-protobuf" assert len(payload) < 4_000_000 + # Count only repeated key/value bytes: MessagePack headers add more bytes. + generated_lower_bound = attributes * scopes * (len("key0000") + 64) + assert generated_lower_bound > 4_000_000 + if expected_status == 400: + assert generated_lower_bound > 100_000_000 try: service.start() - assert service.send_raw_request("/v1/logs", payload, content_type).status_code == 400 + assert service.send_raw_request("/v1/logs", payload, content_type).status_code == expected_status + # Large conversion buffers must be cleaned up without poisoning later requests. + small = generated_limit_payload(1, 1).SerializeToString() + assert service.send_raw_request("/v1/logs", small).status_code == 201 finally: service.stop()