From c323e9e58b51667fa7636763e0eaaf89f207848d Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 14 Aug 2026 15:25:55 +0900 Subject: [PATCH 1/4] utils: Fix the boundary length of SIMD offloading Signed-off-by: Hiroshi Hatake --- src/flb_utils.c | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) diff --git a/src/flb_utils.c b/src/flb_utils.c index 7a6077b145d..5f536a5c53f 100644 --- a/src/flb_utils.c +++ b/src/flb_utils.c @@ -878,7 +878,7 @@ static int flb_utils_write_str_escaped(char *buf, int *off, size_t size, const c { int i, b, ret, len, hex_bytes, utf_sequence_length, utf_sequence_number; int processed_bytes = 0; - int is_valid, copypos = 0, vlen; + int is_valid, copypos = 0; uint32_t c; uint32_t codepoint = 0; uint32_t state = 0; @@ -902,11 +902,9 @@ static int flb_utils_write_str_escaped(char *buf, int *off, size_t size, const c p = buf + *off; - /* align length to the nearest multiple of the vector size for safe SIMD processing */ - vlen = str_len & ~(inst_len - 1); for (i = 0;;) { /* SIMD optimization: Process chunk of input string */ - for (; i < vlen; i += inst_len) { + for (; i + inst_len <= str_len; i += inst_len) { flb_vector8 chunk; flb_vector8_load(&chunk, (const uint8_t *)&str[i]); @@ -1247,7 +1245,7 @@ static inline int flb_utf8_validate_char(const unsigned char *str, int max_len) static int flb_utils_write_str_raw(char *buf, int *off, size_t size, const char *str, size_t str_len) { - int i, b, vlen, len, utf_len, copypos = 0; + int i, b, len, utf_len, copypos = 0; size_t available; char *p; off_t offset = 0; @@ -1258,15 +1256,12 @@ static int flb_utils_write_str_raw(char *buf, int *off, size_t size, available = size - *off; p = buf + *off; - /* align length to the nearest multiple of the vector size for safe SIMD processing */ - vlen = str_len & ~(inst_len - 1); - for (i = 0;;) { /* * Process chunks of the input string using SIMD instructions. * This loop continues as long as it finds "safe" ASCII characters. */ - for (; i < vlen; i += inst_len) { + for (; i + inst_len <= str_len; i += inst_len) { flb_vector8 chunk; flb_vector8_load(&chunk, (const uint8_t *)&str[i]); From b39bc81978f7386f8dfde682dfaf26f556e49053 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 14 Aug 2026 15:26:31 +0900 Subject: [PATCH 2/4] tests: internal: Add a test case for the boundary of SIMD op length Signed-off-by: Hiroshi Hatake --- tests/internal/utils.c | 63 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/tests/internal/utils.c b/tests/internal/utils.c index 888644c826b..2dbb67812bb 100644 --- a/tests/internal/utils.c +++ b/tests/internal/utils.c @@ -2,6 +2,7 @@ #include #include +#include #include #include #include "flb_tests_internal.h" @@ -339,6 +340,67 @@ void test_write_str() TEST_CHECK(ret == FLB_FALSE); } +static void check_write_str_simd_boundary(size_t input_len) +{ + int off; + int ret; + size_t output_size; + char *input; + char *output; + + output_size = input_len + FLB_SIMD_VEC8_INST_LEN + 8; + + input = flb_malloc(input_len + FLB_SIMD_VEC8_INST_LEN); + output = flb_calloc(output_size, sizeof(char)); + if (!TEST_CHECK(input != NULL && output != NULL)) { + flb_free(input); + flb_free(output); + return; + } + + /* + * A multibyte character moves the input cursor off its original SIMD + * alignment. Poison the bytes beyond the declared string length to catch + * a subsequent vector copy that crosses that boundary. + */ + memset(input, 'x', input_len + FLB_SIMD_VEC8_INST_LEN); + input[0] = '\xc2'; + input[1] = '\xae'; + + off = 0; + ret = flb_utils_write_str(output, &off, output_size, input, input_len, FLB_TRUE); + TEST_CHECK(ret == FLB_TRUE); + TEST_CHECK_(off == input_len + 4, "expected %zu escaped bytes, got %d", + input_len + 4, off); + TEST_CHECK(memcmp(output, "\\u00ae", 6) == 0); + TEST_CHECK(memcmp(output + 6, input + 2, input_len - 2) == 0); + + memset(output, 0, output_size); + off = 0; + ret = flb_utils_write_str(output, &off, output_size, input, input_len, FLB_FALSE); + TEST_CHECK(ret == FLB_TRUE); + TEST_CHECK_(off == input_len, "expected %zu raw bytes, got %d", input_len, off); + TEST_CHECK(memcmp(output, input, input_len) == 0); + + flb_free(input); + flb_free(output); +} + +void test_write_str_simd_boundary() +{ + size_t i; + size_t input_lengths[] = { + FLB_SIMD_VEC8_INST_LEN * 2, + /* Historical x86_64 and arm64 macOS coroutine stack sizes. */ + 24 * 1024, + 36 * 1024 + }; + + for (i = 0; i < sizeof(input_lengths) / sizeof(input_lengths[0]); i++) { + check_write_str_simd_boundary(input_lengths[i]); + } +} + void test_write_str_invalid_trailing_bytes() { struct write_str_case cases[] = { @@ -1023,6 +1085,7 @@ TEST_LIST = { { "url_split", test_url_split }, { "url_split_sds", test_url_split_sds }, { "write_str", test_write_str }, + { "write_str_simd_boundary", test_write_str_simd_boundary }, { "write_str_special_bytes", test_write_str_special_bytes }, { "write_raw_str_special_bytes", test_write_raw_str_special_bytes }, { "write_raw_str_invalid_bytes", test_write_raw_str_invalid_sequences}, From 8a5b6381968c7f82e1f335de41459817118a221a Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 14 Aug 2026 15:27:11 +0900 Subject: [PATCH 3/4] tests: runtime: Add a test case for long payloads Signed-off-by: Hiroshi Hatake --- tests/runtime/out_loki.c | 150 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 150 insertions(+) diff --git a/tests/runtime/out_loki.c b/tests/runtime/out_loki.c index 2e31c78273d..bdc49caf9e6 100644 --- a/tests/runtime/out_loki.c +++ b/tests/runtime/out_loki.c @@ -1216,6 +1216,155 @@ void flb_test_tenant_id_key_partial_error() FLB_TRUE); } +static char *create_long_unicode_input(size_t message_size, size_t *input_size) +{ + size_t json_prefix_len; + size_t message_prefix_len; + size_t message_suffix_len; + size_t json_suffix_len; + char *input; + char *message; + const char json_prefix[] = "[12345678,{\"seq\":0,\"msg\":\""; + const char message_prefix[] = ""; + const char message_suffix[] = ""; + const char json_suffix[] = "\"}]"; + + json_prefix_len = sizeof(json_prefix) - 1; + message_prefix_len = sizeof(message_prefix) - 1; + message_suffix_len = sizeof(message_suffix) - 1; + json_suffix_len = sizeof(json_suffix) - 1; + + if (message_size < message_prefix_len + message_suffix_len) { + return NULL; + } + + *input_size = json_prefix_len + message_size + json_suffix_len; + input = flb_malloc(*input_size + 1); + if (input == NULL) { + return NULL; + } + + memcpy(input, json_prefix, json_prefix_len); + message = input + json_prefix_len; + memset(message, '1', message_size); + memcpy(message, message_prefix, message_prefix_len); + memcpy(message + message_size - message_suffix_len, + message_suffix, message_suffix_len); + memcpy(message + message_size, json_suffix, json_suffix_len); + input[*input_size] = '\0'; + + return input; +} + +struct long_unicode_test_result { + size_t message_size; + int callback_called; +}; + +static void cb_check_long_unicode_payload(void *ctx, int ffd, + int res_ret, void *res_data, + size_t res_size, void *data) +{ + char *ending; + char *removed_key; + flb_sds_t out_js; + struct long_unicode_test_result *result; + + out_js = res_data; + result = data; + result->callback_called = FLB_TRUE; + + TEST_CHECK(res_ret == 0); + if (!TEST_CHECK(out_js != NULL)) { + return; + } + + ending = strstr(out_js, "\\\"}"); + if (!TEST_CHECK(ending != NULL)) { + TEST_MSG("%zu-byte message did not end at ", result->message_size); + } + + removed_key = strstr(out_js, "\\\"seq\\\":"); + TEST_CHECK(removed_key == NULL); + + flb_sds_destroy(out_js); +} + +static void run_long_unicode_payload_boundary(size_t message_size) +{ + int ret; + int in_ffd; + int out_ffd; + size_t input_size; + char *input; + flb_ctx_t *ctx; + struct long_unicode_test_result result; + + input = create_long_unicode_input(message_size, &input_size); + if (!TEST_CHECK(input != NULL)) { + return; + } + + ctx = flb_create(); + if (!TEST_CHECK(ctx != NULL)) { + flb_free(input); + return; + } + + ret = flb_service_set(ctx, + "flush", "1", + "grace", "1", + "log_level", "error", + NULL); + TEST_CHECK(ret == 0); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(in_ffd >= 0); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "loki", NULL); + TEST_CHECK(out_ffd >= 0); + ret = flb_output_set(ctx, out_ffd, + "match", "test", + "line_format", "json", + "labels", "job=mwe", + "remove_keys", "seq", + NULL); + TEST_CHECK(ret == 0); + + result.message_size = message_size; + result.callback_called = FLB_FALSE; + ret = flb_output_set_test(ctx, out_ffd, "formatter", + cb_check_long_unicode_payload, + &result, NULL); + TEST_CHECK(ret == 0); + + ret = flb_start(ctx); + if (!TEST_CHECK(ret == 0)) { + flb_destroy(ctx); + flb_free(input); + return; + } + + ret = flb_lib_push(ctx, in_ffd, input, input_size); + TEST_CHECK(ret >= 0); + + sleep(2); + flb_stop(ctx); + flb_destroy(ctx); + flb_free(input); + + if (!TEST_CHECK(result.callback_called == FLB_TRUE)) { + TEST_MSG("formatter was not called for %zu-byte message", message_size); + } +} + +void flb_test_long_unicode_payload_boundaries() +{ + run_long_unicode_payload_boundary(24 * 1024); + run_long_unicode_payload_boundary(36 * 1024); +} + static void cb_check_label_map_path(void *ctx, int ffd, int res_ret, void *res_data, size_t res_size, void *data) @@ -1548,6 +1697,7 @@ TEST_LIST = { {"tenant_id_key_splits_requests", flb_test_tenant_id_key_splits_requests }, {"tenant_id_key_partial_success", flb_test_tenant_id_key_partial_success }, {"tenant_id_key_partial_error", flb_test_tenant_id_key_partial_error }, + {"long_unicode_payload_boundaries", flb_test_long_unicode_payload_boundaries }, {"basic" , flb_test_basic }, {"labels" , flb_test_labels }, {"label_keys" , flb_test_label_keys }, From 3710135f56d45d98bdcd274365e2cf20cc188dca Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 14 Aug 2026 16:00:20 +0900 Subject: [PATCH 4/4] tests: integration: Add an integration test case for SIMD boundaries Signed-off-by: Hiroshi Hatake --- .../config/out_loki_long_unicode.yaml | 29 ++++ .../scenarios/out_loki/config/parsers.conf | 3 + .../out_loki/tests/test_out_loki_001.py | 133 ++++++++++++++++++ tests/integration/src/server/http_server.py | 1 + 4 files changed, 166 insertions(+) create mode 100644 tests/integration/scenarios/out_loki/config/out_loki_long_unicode.yaml create mode 100644 tests/integration/scenarios/out_loki/config/parsers.conf create mode 100644 tests/integration/scenarios/out_loki/tests/test_out_loki_001.py diff --git a/tests/integration/scenarios/out_loki/config/out_loki_long_unicode.yaml b/tests/integration/scenarios/out_loki/config/out_loki_long_unicode.yaml new file mode 100644 index 00000000000..4bd4d59c78c --- /dev/null +++ b/tests/integration/scenarios/out_loki/config/out_loki_long_unicode.yaml @@ -0,0 +1,29 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + parsers_file: ${LOKI_PARSERS_FILE} + +pipeline: + inputs: + - name: tail + tag: loki.integration + path: ${LOKI_INPUT_PATH} + parser: loki-integration-json + read_from_head: true + refresh_interval: 1 + buffer_chunk_size: 256k + buffer_max_size: 20m + mem_buf_limit: 512m + + outputs: + - name: loki + match: loki.integration + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + line_format: json + labels: job=integration-test + remove_keys: seq + retry_limit: no_limits diff --git a/tests/integration/scenarios/out_loki/config/parsers.conf b/tests/integration/scenarios/out_loki/config/parsers.conf new file mode 100644 index 00000000000..8f9f5dd2c63 --- /dev/null +++ b/tests/integration/scenarios/out_loki/config/parsers.conf @@ -0,0 +1,3 @@ +[PARSER] + Name loki-integration-json + Format json diff --git a/tests/integration/scenarios/out_loki/tests/test_out_loki_001.py b/tests/integration/scenarios/out_loki/tests/test_out_loki_001.py new file mode 100644 index 00000000000..63a52f0082b --- /dev/null +++ b/tests/integration/scenarios/out_loki/tests/test_out_loki_001.py @@ -0,0 +1,133 @@ +import json +import os +from pathlib import Path +import tempfile + +import requests + +from server.http_server import data_storage, http_server_run +from utils.memory_check import memory_check_enabled +from utils.test_service import FluentBitTestService + + +MESSAGE_SIZE = 40 * 1024 +RECORD_COUNT = 200 + + +def create_message(sequence): + prefix = f"" + suffix = "" + token = f"{sequence}," + remaining = MESSAGE_SIZE - len(prefix.encode("utf-8")) - len(suffix) + repetitions = remaining // len(token) + padding = remaining - repetitions * len(token) + + return prefix + token * repetitions + "-" * padding + suffix + + +def write_input(path): + expected_messages = [] + + with path.open("w", encoding="utf-8") as input_file: + for sequence in range(RECORD_COUNT): + message = create_message(sequence) + expected_messages.append(message) + json.dump({"seq": sequence, "msg": message}, input_file, ensure_ascii=False) + input_file.write("\n") + + return expected_messages + + +def count_loki_values(): + count = 0 + + for payload in data_storage["payloads"]: + if not isinstance(payload, dict): + continue + + for stream in payload.get("streams", []): + count += len(stream.get("values", [])) + + return count + + +def collect_loki_records(): + records = [] + + for payload in data_storage["payloads"]: + assert isinstance(payload, dict) + + for stream in payload.get("streams", []): + for value in stream.get("values", []): + records.append(json.loads(value[1])) + + return records + + +class Service: + def __init__(self, input_path): + test_directory = Path(__file__).resolve().parent + config_directory = test_directory.parent / "config" + self.service = FluentBitTestService( + str(config_directory / "out_loki_long_unicode.yaml"), + data_storage=data_storage, + data_keys=["payloads", "requests"], + extra_env={ + "LOKI_INPUT_PATH": str(input_path), + "LOKI_PARSERS_FILE": str(config_directory / "parsers.conf"), + }, + pre_start=self._start_receiver, + post_stop=self._stop_receiver, + ) + + def _start_receiver(self, service): + http_server_run(service.test_suite_http_port) + self.service.wait_for_http_endpoint( + f"http://127.0.0.1:{service.test_suite_http_port}/ping", + timeout=10, + interval=0.5, + ) + + def _stop_receiver(self, service): + try: + requests.post( + f"http://127.0.0.1:{service.test_suite_http_port}/shutdown", + timeout=2, + ) + except requests.RequestException: + pass + + def start(self): + self.service.start() + + def stop(self): + self.service.stop() + + def wait_for_records(self): + timeout = 60 if memory_check_enabled() else 20 + + self.service.wait_for_condition( + lambda: count_loki_values() >= RECORD_COUNT, + timeout=timeout, + interval=0.5, + description=f"{RECORD_COUNT} Loki values", + ) + + +def test_out_loki_preserves_long_unicode_json_strings(): + with tempfile.TemporaryDirectory(prefix="flb-out-loki-it-") as temp_directory: + input_path = Path(temp_directory) / "input.log" + expected_messages = write_input(input_path) + service = Service(input_path) + + try: + service.start() + service.wait_for_records() + finally: + service.stop() + + records = collect_loki_records() + + assert len(records) == RECORD_COUNT + assert all(set(record) == {"msg"} for record in records) + assert sorted(record["msg"] for record in records) == sorted(expected_messages) diff --git a/tests/integration/src/server/http_server.py b/tests/integration/src/server/http_server.py index 7295fffe12d..f06fe909a06 100644 --- a/tests/integration/src/server/http_server.py +++ b/tests/integration/src/server/http_server.py @@ -256,6 +256,7 @@ def _decode_json_payload(decoded_payload): @app.route('/data', methods=['POST']) @app.route('/shared', methods=['POST']) @app.route('/solo', methods=['POST']) +@app.route('/loki/api/v1/push', methods=['POST']) @app.route('/dataCollectionRules/', methods=['POST']) def receive_data(subpath=None): _record_request()