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/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) { 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 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); } 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) { 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()