From c8decb23c25bcadd1eacc7e1975346d733e8385f Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Sat, 17 Oct 2026 12:22:19 -0600 Subject: [PATCH 1/5] input_chunk: keep a log group in one chunk A group whose start and end arrive in separate appends, such as forward messages from a sender that splits a group, was split across chunks when the chunk was flushed or closed for size between the appends. Outputs then received the end of the group without its start and lost its context. Track whether a log chunk ends inside a group, from the group markers of each append, and keep such a chunk open until the group ends: it is not closed when it grows past its size limit, up to twice that limit, and flb_input_chunk_holds_open_group() tells the dispatcher to wait for it. The wait is bounded to five seconds and ends on shutdown. A cheap byte scan skips buffers that cannot contain markers. Signed-off-by: Eduardo Silva --- include/fluent-bit/flb_input_chunk.h | 11 ++ src/flb_input_chunk.c | 211 ++++++++++++++++++++++++++- 2 files changed, 220 insertions(+), 2 deletions(-) diff --git a/include/fluent-bit/flb_input_chunk.h b/include/fluent-bit/flb_input_chunk.h index 681ae9afa69..1a6fff1709c 100644 --- a/include/fluent-bit/flb_input_chunk.h +++ b/include/fluent-bit/flb_input_chunk.h @@ -42,6 +42,14 @@ struct cio_chunk; */ #define FLB_INPUT_CHUNK_FS_MAX_SIZE 2048000 /* 2MB */ +/* + * A log chunk that ends inside a group is not flushed or closed for size until + * the group ends, so the start and end markers of a group stay in the same + * chunk. These limits bound the wait for senders that never end the group. + */ +#define FLB_INPUT_CHUNK_GROUP_HOLD_SEC 5.0 +#define FLB_INPUT_CHUNK_GROUP_MAX_SIZE (FLB_INPUT_CHUNK_FS_MAX_SIZE * 2) + /* Number of bytes reserved for Metadata Header on Chunks */ #define FLB_INPUT_CHUNK_META_HEADER 4 @@ -103,12 +111,15 @@ struct flb_input_chunk { #endif /* FLB_HAVE_CHUNK_TRACE */ double create_time; /* chunk creation time in seconds with fractional precision) */ flb_route_mask_element *routes_mask; /* track the output plugins the chunk routes to */ + int group_open; /* log chunk ends inside a group */ + double group_open_time; /* when the open group started */ struct mk_list _head; }; struct flb_input_chunk *flb_input_chunk_create(struct flb_input_instance *in, int event_type, const char *tag, int tag_len); int flb_input_chunk_destroy(struct flb_input_chunk *ic, int del); +int flb_input_chunk_holds_open_group(struct flb_input_chunk *ic); void flb_input_chunk_destroy_all(struct flb_input_instance *in); int flb_input_chunk_destroy_corrupted(struct flb_input_chunk *ic, const char *tag_buf, int tag_len, diff --git a/src/flb_input_chunk.c b/src/flb_input_chunk.c index f3b9490239b..8cea6434930 100644 --- a/src/flb_input_chunk.c +++ b/src/flb_input_chunk.c @@ -39,6 +39,7 @@ #include #include #include +#include #include #include #include @@ -2826,6 +2827,197 @@ int flb_input_chunk_destroy(struct flb_input_chunk *ic, int del) } /* Return or create an available chunk to write data */ +/* + * Group markers are events whose timestamp is -1 (start) or -2 (end), stored + * as an EventTime with 0xffffffff or 0xfffffffe seconds, or as a negative + * number. Every event header is a two entries array, so a marker can only + * follow the bytes 0x92 0x92. Buffers without such a candidate, nearly all + * buffers of inputs that do not create groups, skip the event scan. + */ +static int log_buffer_may_have_groups(const char *buf, size_t size) +{ + size_t left; + const unsigned char *p; + const unsigned char *end; + + p = (const unsigned char *) buf; + end = p + size; + + while (p + 2 < end) { + p = memchr(p, 0x92, end - p - 2); + if (p == NULL) { + return FLB_FALSE; + } + + if (p[1] != 0x92) { + p++; + continue; + } + + left = end - (p + 2); + switch (p[2]) { + case 0xd7: + /* EventTime: fixext 8, type 0, seconds 0xfffffffe or 0xffffffff */ + if (left >= 6 && p[3] == 0x00 && p[4] == 0xff && p[5] == 0xff && + p[6] == 0xff && (p[7] == 0xff || p[7] == 0xfe)) { + return FLB_TRUE; + } + break; + case 0xd0: case 0xd1: case 0xd2: case 0xd3: case 0xca: case 0xcb: + /* signed integers and floats */ + return FLB_TRUE; + default: + if (p[2] >= 0xe0) { + /* negative fixint */ + return FLB_TRUE; + } + break; + } + p++; + } + + return FLB_FALSE; +} + +/* 1 for a group start, 2 for a group end, 0 for a record, -1 on error */ +static int log_event_group_marker(mpack_reader_t *reader) +{ + int header; + int marker = 0; + char ext[8]; + double value; + uint32_t seconds; + mpack_tag_t tag; + + tag = mpack_read_tag(reader); + if (mpack_tag_type(&tag) != mpack_type_array || mpack_tag_array_count(&tag) != 2) { + return -1; + } + + tag = mpack_read_tag(reader); + header = mpack_tag_type(&tag) == mpack_type_array; + if (header) { + if (mpack_tag_array_count(&tag) != 2) { + return -1; + } + tag = mpack_read_tag(reader); + } + + if (mpack_tag_type(&tag) == mpack_type_ext) { + if (mpack_tag_ext_exttype(&tag) == 0 && mpack_tag_ext_length(&tag) == 8) { + mpack_read_bytes(reader, ext, sizeof(ext)); + seconds = ((uint32_t) (unsigned char) ext[0] << 24) | + ((uint32_t) (unsigned char) ext[1] << 16) | + ((uint32_t) (unsigned char) ext[2] << 8) | + (uint32_t) (unsigned char) ext[3]; + if (seconds == UINT32_MAX) { + marker = 1; + } + else if (seconds == UINT32_MAX - 1) { + marker = 2; + } + } + else { + mpack_skip_bytes(reader, mpack_tag_ext_length(&tag)); + } + mpack_done_ext(reader); + } + else if (mpack_tag_type(&tag) == mpack_type_int) { + if (mpack_tag_int_value(&tag) == FLB_LOG_EVENT_GROUP_START) { + marker = 1; + } + else if (mpack_tag_int_value(&tag) == FLB_LOG_EVENT_GROUP_END) { + marker = 2; + } + } + else if (mpack_tag_type(&tag) == mpack_type_float || + mpack_tag_type(&tag) == mpack_type_double) { + value = mpack_tag_type(&tag) == mpack_type_float ? + mpack_tag_float_value(&tag) : mpack_tag_double_value(&tag); + if ((int) value == FLB_LOG_EVENT_GROUP_START) { + marker = 1; + } + else if ((int) value == FLB_LOG_EVENT_GROUP_END) { + marker = 2; + } + } + + if (header) { + mpack_discard(reader); + mpack_done_array(reader); + } + mpack_discard(reader); + mpack_done_array(reader); + + if (mpack_reader_error(reader) != mpack_ok) { + return -1; + } + + return marker; +} + +/* + * Whether a log chunk still ends inside a group after appending 'buf': the + * last group marker of the buffer decides, a buffer without markers keeps + * the previous state. Unreadable buffers keep it as well. + */ +static int log_buffer_group_open(const char *buf, size_t size, int open) +{ + int marker; + int state; + mpack_reader_t reader; + + if (buf == NULL || size == 0 || !log_buffer_may_have_groups(buf, size)) { + return open; + } + + state = open; + mpack_reader_init_data(&reader, buf, size); + while (mpack_reader_remaining(&reader, NULL) > 0) { + marker = log_event_group_marker(&reader); + if (marker == -1) { + state = open; + break; + } + else if (marker == 1) { + state = FLB_TRUE; + } + else if (marker == 2) { + state = FLB_FALSE; + } + } + mpack_reader_destroy(&reader); + + return state; +} + +/* + * A log chunk ending inside a group waits for the group end, so the group is + * not split across chunks. The wait is bounded, and ends on shutdown. + */ +int flb_input_chunk_holds_open_group(struct flb_input_chunk *ic) +{ + struct flb_config *config; + + if (ic == NULL || ic->group_open == FLB_FALSE) { + return FLB_FALSE; + } + + config = ic->in->config; + if (config->is_shutting_down == FLB_TRUE || + config->is_ingestion_active == FLB_FALSE) { + return FLB_FALSE; + } + + if (flb_time_now() - ic->group_open_time >= FLB_INPUT_CHUNK_GROUP_HOLD_SEC) { + flb_debug("[input chunk] flushing a chunk whose last group did not end " + "within %.0f seconds", FLB_INPUT_CHUNK_GROUP_HOLD_SEC); + return FLB_FALSE; + } + + return FLB_TRUE; +} + static struct flb_input_chunk *input_chunk_get(struct flb_input_instance *in, int event_type, const char *tag, int tag_len, @@ -3266,6 +3458,7 @@ static int input_chunk_append_raw(struct flb_input_instance *in, #endif int set_down = FLB_FALSE; int min; + int open; int new_chunk = FLB_FALSE; uint64_t ts; char *name; @@ -3466,6 +3659,16 @@ static int input_chunk_append_raw(struct flb_input_instance *in, ret = 0; } + /* track whether the chunk now ends inside a group */ + if (ret == 0 && event_type == FLB_INPUT_LOGS && final_data_size > 0) { + open = log_buffer_group_open(final_data_buffer, final_data_size, + ic->group_open); + if (open == FLB_TRUE && ic->group_open == FLB_FALSE) { + ic->group_open_time = flb_time_now(); + } + ic->group_open = open; + } + if (filtered_data_buffer != NULL && filtered_data_buffer != buf) { flb_free(filtered_data_buffer); @@ -3507,8 +3710,12 @@ static int input_chunk_append_raw(struct flb_input_instance *in, real_diff = 0; } - /* Lock buffers where size > 2MB */ - if (content_size > FLB_INPUT_CHUNK_FS_MAX_SIZE) { + /* + * Lock buffers where size > 2MB. A chunk that ends inside a group stays + * open for the rest of the group, up to a hard limit. + */ + if (content_size > FLB_INPUT_CHUNK_GROUP_MAX_SIZE || + (content_size > FLB_INPUT_CHUNK_FS_MAX_SIZE && ic->group_open == FLB_FALSE)) { cio_chunk_lock(ic->chunk); } From 91ca30ef748c7857ad930568afe233c45c77ff3a Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Sat, 17 Oct 2026 12:22:19 -0600 Subject: [PATCH 2/5] engine_dispatch: wait for the end of an open log group Skip a chunk that still ends inside a log group, so the rest of the group is appended to it instead of a new chunk. Signed-off-by: Eduardo Silva --- src/flb_engine_dispatch.c | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/flb_engine_dispatch.c b/src/flb_engine_dispatch.c index ff13720110d..c3b1b45cef2 100644 --- a/src/flb_engine_dispatch.c +++ b/src/flb_engine_dispatch.c @@ -384,6 +384,11 @@ int flb_engine_dispatch(uint64_t id, struct flb_input_instance *in, continue; } + /* keep appending the rest of an open group to this chunk */ + if (flb_input_chunk_holds_open_group(ic) == FLB_TRUE) { + continue; + } + if (flb_task_map_get_task_id(config) == -1) { /* * There isn't a task available, no more chunks can have a task From 8c06f20a8d67407eb58638cdcecc74782ad71997 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Sat, 17 Oct 2026 12:22:19 -0600 Subject: [PATCH 3/5] input_log: keep routed groups whole across appends When an append left a group open, the route payload repeated its start marker before every matching record, and the end of a group started by a previous append was only kept by accident when no other group followed. Emit the start of a group once, and emit the end of a continued group after its records. Signed-off-by: Eduardo Silva --- src/flb_input_log.c | 54 +++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 50 insertions(+), 4 deletions(-) diff --git a/src/flb_input_log.c b/src/flb_input_log.c index e14c84820c1..e82f559415f 100644 --- a/src/flb_input_log.c +++ b/src/flb_input_log.c @@ -452,6 +452,8 @@ static int build_payload_for_route(struct flb_input_instance *ins, struct flb_log_event_encoder *encoder; struct flb_mp_chunk_record *group_end = NULL; struct flb_mp_chunk_record *group_start_record = NULL; + int group_start_emitted = FLB_FALSE; + int group_continued = FLB_FALSE; uint8_t *matched_by_route = NULL; if (!payload || !records || record_count == 0 || !matched_non_default || @@ -509,7 +511,12 @@ static int build_payload_for_route(struct flb_input_instance *ins, return 0; } - /* Second pass: encode records in order, preserving group structure */ + /* + * Second pass: encode records in order, preserving group structure. The + * first pass leaves group_end on the last end marker; it is set again for + * each group below. + */ + group_end = NULL; for (i = 0; i < record_count; i++) { if (flb_log_event_decoder_get_record_type(&records[i]->event, &record_type) != 0) { continue; @@ -517,6 +524,7 @@ static int build_payload_for_route(struct flb_input_instance *ins, if (record_type == FLB_LOG_EVENT_GROUP_START) { group_start_record = records[i]; + group_start_emitted = FLB_FALSE; group_end = NULL; continue; } @@ -532,13 +540,25 @@ static int build_payload_for_route(struct flb_input_instance *ins, } group_end = NULL; } + else if (group_start_record == NULL && group_continued == FLB_TRUE) { + /* end of a group started by the previous append */ + ret = encode_chunk_record(encoder, records[i]); + if (ret != 0) { + flb_free(matched_by_route); + flb_log_event_encoder_destroy(encoder); + return -1; + } + } group_start_record = NULL; + group_continued = FLB_FALSE; continue; } else if (record_type == FLB_LOG_EVENT_NORMAL && matched_by_route[i]) { if (group_start_record != NULL && records[i]->cobj_group_metadata == group_start_record->cobj_group_metadata) { - if (group_end == NULL) { + /* a group without an end in this append starts only once */ + if (group_start_emitted == FLB_FALSE) { + group_start_emitted = FLB_TRUE; ret = encode_chunk_record(encoder, group_start_record); if (ret != 0) { flb_free(matched_by_route); @@ -555,6 +575,9 @@ static int build_payload_for_route(struct flb_input_instance *ins, } } } + else if (group_start_record == NULL) { + group_continued = FLB_TRUE; + } ret = encode_chunk_record(encoder, records[i]); if (ret != 0) { @@ -615,6 +638,8 @@ static int build_payload_for_default_route(struct flb_input_instance *ins, struct flb_log_event_encoder *encoder; struct flb_mp_chunk_record *group_end = NULL; struct flb_mp_chunk_record *group_start_record = NULL; + int group_start_emitted = FLB_FALSE; + int group_continued = FLB_FALSE; int *matched_by_default = NULL; if (!payload || !records || !matched_non_default || !matched_any_route) { @@ -685,7 +710,12 @@ static int build_payload_for_default_route(struct flb_input_instance *ins, } } - /* Second pass: encode records in order, preserving group structure */ + /* + * Second pass: encode records in order, preserving group structure. The + * first pass leaves group_end on the last end marker; it is set again for + * each group below. + */ + group_end = NULL; for (i = 0; i < record_count; i++) { if (flb_log_event_decoder_get_record_type(&records[i]->event, &record_type) != 0) { continue; @@ -693,6 +723,7 @@ static int build_payload_for_default_route(struct flb_input_instance *ins, if (record_type == FLB_LOG_EVENT_GROUP_START) { group_start_record = records[i]; + group_start_emitted = FLB_FALSE; group_end = NULL; continue; } @@ -708,13 +739,25 @@ static int build_payload_for_default_route(struct flb_input_instance *ins, } group_end = NULL; } + else if (group_start_record == NULL && group_continued == FLB_TRUE) { + /* end of a group started by the previous append */ + ret = encode_chunk_record(encoder, records[i]); + if (ret != 0) { + flb_free(matched_by_default); + flb_log_event_encoder_destroy(encoder); + return -1; + } + } group_start_record = NULL; + group_continued = FLB_FALSE; continue; } else if (record_type == FLB_LOG_EVENT_NORMAL && matched_by_default[i]) { if (group_start_record != NULL && records[i]->cobj_group_metadata == group_start_record->cobj_group_metadata) { - if (group_end == NULL) { + /* a group without an end in this append starts only once */ + if (group_start_emitted == FLB_FALSE) { + group_start_emitted = FLB_TRUE; ret = encode_chunk_record(encoder, group_start_record); if (ret != 0) { flb_free(matched_by_default); @@ -731,6 +774,9 @@ static int build_payload_for_default_route(struct flb_input_instance *ins, } } } + else if (group_start_record == NULL) { + group_continued = FLB_TRUE; + } ret = encode_chunk_record(encoder, records[i]); if (ret != 0) { From cad3d33dccc4f0e585058784a062cb2493a65835 Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Sat, 17 Oct 2026 12:22:19 -0600 Subject: [PATCH 4/5] in_forward: split large forward frames only between groups A Forward mode frame larger than the ingress byte limit was split into batches by size. The decoder skipped group markers without counting their bytes, so a batch could cut the end of its last record and drop markers, and a split could fall inside a group. Read the markers as events, and when a group does not fit, ingest the events before it and start the next batch with the group. Signed-off-by: Eduardo Silva --- plugins/in_forward/fw_prot.c | 52 +++++++++++++++++++++++++++++++----- 1 file changed, 46 insertions(+), 6 deletions(-) diff --git a/plugins/in_forward/fw_prot.c b/plugins/in_forward/fw_prot.c index 82213c61173..c201c9c67f6 100644 --- a/plugins/in_forward/fw_prot.c +++ b/plugins/in_forward/fw_prot.c @@ -1023,6 +1023,9 @@ static int fw_ingest_forward_mode_batch(struct fw_conn *conn, { int decode_result; int ingest_result; + int in_group; + int32_t record_type; + size_t group_offset; size_t batch_length; size_t byte_limit; const char *batch; @@ -1040,29 +1043,66 @@ static int fw_ingest_forward_mode_batch(struct fw_conn *conn, return -1; } + /* + * Group markers are events too: count their bytes, so every batch is the + * exact contiguous run of its events, and split batches only between + * groups: when a group does not fit, the events before it are ingested + * and the group starts the next batch. A group larger than the limit is + * sent whole, like a single record larger than the limit. + */ + flb_log_event_decoder_read_groups(&decoder, FLB_TRUE); + batch = NULL; batch_length = 0; ingest_result = 0; + in_group = FLB_FALSE; + group_offset = 0; while ((decode_result = flb_log_event_decoder_next(&decoder, &event)) == FLB_EVENT_DECODER_SUCCESS) { + if (flb_log_event_decoder_get_record_type(&event, &record_type) != 0) { + record_type = FLB_LOG_EVENT_NORMAL; + } + if (batch_length > 0 && (batch_length > byte_limit || decoder.record_length > byte_limit - batch_length)) { - ingest_result = fw_ingest_forward_mode_chunk( - conn, tag, tag_len, batch, batch_length); - if (ingest_result != 0) { - break; + if (in_group == FLB_FALSE && record_type != FLB_LOG_EVENT_GROUP_END) { + /* between groups: send the whole batch */ + ingest_result = fw_ingest_forward_mode_chunk( + conn, tag, tag_len, batch, batch_length); + if (ingest_result != 0) { + break; + } + batch_length = 0; + } + else if (in_group == FLB_TRUE && group_offset > 0) { + /* send the events before the open group, keep the group */ + ingest_result = fw_ingest_forward_mode_chunk( + conn, tag, tag_len, batch, group_offset); + if (ingest_result != 0) { + break; + } + batch += group_offset; + batch_length -= group_offset; + group_offset = 0; } - - batch_length = 0; } if (batch_length == 0) { batch = decoder.record_base; } + if (record_type == FLB_LOG_EVENT_GROUP_START) { + in_group = FLB_TRUE; + group_offset = batch_length; + } + batch_length += decoder.record_length; + + if (record_type == FLB_LOG_EVENT_GROUP_END) { + in_group = FLB_FALSE; + } } if (ingest_result == 0) { From a4f6105a9ec69deca62824cecb2e3c2ed8c1f26b Mon Sep 17 00:00:00 2001 From: Eduardo Silva Date: Sat, 17 Oct 2026 12:22:19 -0600 Subject: [PATCH 5/5] tests: integration: in_forward: cover log groups split across appends Signed-off-by: Eduardo Silva --- ...onditional_groups_to_forward_receiver.yaml | 38 ++++ ...ward_split_groups_to_forward_receiver.yaml | 21 +++ ...rd_workers_groups_to_forward_receiver.yaml | 25 +++ .../in_forward/tests/test_in_forward_001.py | 177 ++++++++++++++++++ 4 files changed, 261 insertions(+) create mode 100644 tests/integration/scenarios/in_forward/config/in_forward_conditional_groups_to_forward_receiver.yaml create mode 100644 tests/integration/scenarios/in_forward/config/in_forward_split_groups_to_forward_receiver.yaml create mode 100644 tests/integration/scenarios/in_forward/config/in_forward_workers_groups_to_forward_receiver.yaml diff --git a/tests/integration/scenarios/in_forward/config/in_forward_conditional_groups_to_forward_receiver.yaml b/tests/integration/scenarios/in_forward/config/in_forward_conditional_groups_to_forward_receiver.yaml new file mode 100644 index 00000000000..ca93f003ae4 --- /dev/null +++ b/tests/integration/scenarios/in_forward/config/in_forward_conditional_groups_to_forward_receiver.yaml @@ -0,0 +1,38 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: forward + alias: routed_forward + listen: 127.0.0.1 + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + routes: + logs: + - name: kept + condition: + op: and + rules: + - field: $keep + op: eq + value: "yes" + to: + outputs: + - kept_forward + + outputs: + - name: forward + alias: kept_forward + host: 127.0.0.1 + port: ${FORWARD_RECEIVER_PORT} + send_options: true + require_ack_response: true + retain_metadata_in_forward_mode: true + + - name: null + alias: unrouted + match: unrouted.* diff --git a/tests/integration/scenarios/in_forward/config/in_forward_split_groups_to_forward_receiver.yaml b/tests/integration/scenarios/in_forward/config/in_forward_split_groups_to_forward_receiver.yaml new file mode 100644 index 00000000000..b320f5606c8 --- /dev/null +++ b/tests/integration/scenarios/in_forward/config/in_forward_split_groups_to_forward_receiver.yaml @@ -0,0 +1,21 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: forward + listen: 127.0.0.1 + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + + outputs: + - name: forward + match: test + host: 127.0.0.1 + port: ${FORWARD_RECEIVER_PORT} + send_options: true + require_ack_response: true + retain_metadata_in_forward_mode: true diff --git a/tests/integration/scenarios/in_forward/config/in_forward_workers_groups_to_forward_receiver.yaml b/tests/integration/scenarios/in_forward/config/in_forward_workers_groups_to_forward_receiver.yaml new file mode 100644 index 00000000000..f21a9dc896f --- /dev/null +++ b/tests/integration/scenarios/in_forward/config/in_forward_workers_groups_to_forward_receiver.yaml @@ -0,0 +1,25 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + storage.path: ${FORWARD_STORAGE_PATH} + +pipeline: + inputs: + - name: forward + listen: 127.0.0.1 + port: ${FLUENT_BIT_TEST_LISTENER_PORT} + workers: 2 + mem_buf_limit: 128k + storage.type: filesystem + + outputs: + - name: forward + match: test + host: 127.0.0.1 + port: ${FORWARD_RECEIVER_PORT} + send_options: true + require_ack_response: true + retain_metadata_in_forward_mode: true diff --git a/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py b/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py index 3ce3f13620e..3f577163273 100644 --- a/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py +++ b/tests/integration/scenarios/in_forward/tests/test_in_forward_001.py @@ -2539,3 +2539,180 @@ def shared_success_eviction_snapshot(): assert any(b"shared-two" in content for content in chunk_contents) assert any(b"shared-three" in content for content in chunk_contents) assert failed_chunks & evicted_chunks + + +def _event_time(seconds): + return ("__ext__", 0, seconds.to_bytes(4, "big") + bytes(4)) + + +def _group_start(name): + return [[_event_time(0xFFFFFFFF), {"schema": "otlp", "resource_id": 0, "scope_id": 0}], + {"resource": {"attributes": {"service.name": name}}, "scope": {"name": "scope"}}] + + +def _group_end(): + return [[_event_time(0xFFFFFFFE), {}], {}] + + +def _group_record(record_id): + return [[_event_time(1700000000), {"otlp": {}}], {"id": record_id}] + + +def _send_forward_entries(port, entries): + with socket.create_connection(("127.0.0.1", port), timeout=10) as connection: + connection.sendall(_pack_obj([TEST_TAG, entries])) + + +def _marker(record): + timestamp = record.get("timestamp") + if isinstance(timestamp, dict) and "__ext__" in timestamp: + seconds = int.from_bytes(timestamp["__ext__"]["data"][:4], "big") + if seconds == 0xFFFFFFFF: + return "start" + if seconds == 0xFFFFFFFE: + return "end" + return None + + +def _flushed_entries(): + """Events of each captured chunk as 'start', 'end' or a record id.""" + chunks = [] + for message in forward_data_storage["messages"]: + events = [] + for record in message["records"]: + marker = _marker(record) + body = record.get("body") + events.append(marker if marker else (body.get("id") if isinstance(body, dict) else None)) + chunks.append(events) + return chunks + + +def _flushed_record_ids(): + return sorted(event for events in _flushed_entries() for event in events + if isinstance(event, int)) + + +def test_in_forward_group_split_across_messages_stays_in_one_chunk(): + """The end of a group received after a flush lands in the chunk of its start.""" + service = ForwardReceiverService("in_forward_split_groups_to_forward_receiver.yaml") + service.start() + try: + _send_forward_entries(service.flb_listener_port, + [_group_start("split"), _group_record(0), _group_record(1)]) + # let at least one flush happen while the group is open + time.sleep(2.5) + _send_forward_entries(service.flb_listener_port, [_group_record(2), _group_end()]) + + service.service.wait_for_condition( + lambda: _flushed_record_ids() == [0, 1, 2] or None, + timeout=15, interval=0.2, description="all records of the split group") + chunks = [events for events in _flushed_entries() if events] + assert chunks == [["start", 0, 1, 2, "end"]] + finally: + service.stop() + + +def test_in_forward_open_group_is_flushed_after_the_hold_limit(): + """A group that never ends does not hold its records forever.""" + service = ForwardReceiverService("in_forward_split_groups_to_forward_receiver.yaml") + service.start() + try: + _send_forward_entries(service.flb_listener_port, [_group_start("open"), _group_record(0)]) + started = time.monotonic() + service.service.wait_for_condition( + lambda: _flushed_record_ids() == [0] or None, + timeout=20, interval=0.2, description="record of the open group") + # held for the open group, then released + assert time.monotonic() - started >= 3 + finally: + service.stop() + + +def test_in_forward_group_crossing_the_chunk_size_stays_in_one_chunk(): + """A chunk that grows past its size limit inside a group takes the group end too.""" + service = ForwardReceiverService("in_forward_split_groups_to_forward_receiver.yaml") + service.start() + try: + large = [[_event_time(1700000000), {"otlp": {}}], {"id": 0, "payload": "x" * (2100 * 1024)}] + with socket.create_connection(("127.0.0.1", service.flb_listener_port), timeout=10) as connection: + connection.sendall(_pack_obj([TEST_TAG, [_group_start("large"), large]])) + connection.sendall(_pack_obj([TEST_TAG, [_group_record(1), _group_end()]])) + + service.service.wait_for_condition( + lambda: _flushed_record_ids() == [0, 1] or None, + timeout=20, interval=0.2, description="records of the large group") + chunks = [events for events in _flushed_entries() if events] + assert chunks == [["start", 0, 1, "end"]] + finally: + service.stop() + + +def test_in_forward_workers_split_large_frames_between_groups(tmp_path): + """A frame above the ingress byte limit is split only between whole groups.""" + service = ForwardReceiverService( + "in_forward_workers_groups_to_forward_receiver.yaml", + extra_env={"FORWARD_STORAGE_PATH": str(tmp_path / "storage")}) + service.start() + try: + entries = [] + expected = [] + record_id = 0 + for group in range(6): + entries.append(_group_start(f"group-{group}")) + for _ in range(5): + entries.append([[_event_time(1700000000), {"otlp": {}}], + {"id": record_id, "group": group, "payload": "y" * 20000}]) + expected.append(record_id) + record_id += 1 + entries.append(_group_end()) + _send_forward_entries(service.flb_listener_port, entries) + + service.service.wait_for_condition( + lambda: _flushed_record_ids() == expected or None, + timeout=20, interval=0.2, description="every record of the large frame") + + groups = [] + for message in forward_data_storage["messages"]: + current = None + for record in message["records"]: + marker = _marker(record) + if marker == "start": + assert current is None + current = [] + elif marker == "end": + assert current is not None + groups.append(current) + current = None + else: + assert current is not None + current.append(record["body"]["group"]) + assert current is None + assert sorted(groups) == [[group] * 5 for group in range(6)] + finally: + service.stop() + + +def test_in_forward_conditional_route_keeps_a_split_group_whole(): + """A routed group split across appends starts once and keeps its end.""" + service = ForwardReceiverService("in_forward_conditional_groups_to_forward_receiver.yaml") + service.start() + try: + def record(record_id): + return [[_event_time(1700000000), {"otlp": {}}], {"id": record_id, "keep": "yes"}] + + _send_forward_entries(service.flb_listener_port, + [_group_start("routed"), record(0), record(1)]) + time.sleep(2.5) + # the end of the open group, then a whole group + _send_forward_entries(service.flb_listener_port, + [record(2), _group_end(), _group_start("next"), record(3), + _group_end()]) + + service.service.wait_for_condition( + lambda: _flushed_record_ids() == [0, 1, 2, 3] or None, + timeout=15, interval=0.2, description="routed records of both groups") + time.sleep(1.5) + chunks = [events for events in _flushed_entries() if events] + assert chunks == [["start", 0, 1, 2, "end", "start", 3, "end"]] + finally: + service.stop()