utils: fix SIMD boundary for JSON string encoding

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]);
 

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

tests/integration/scenarios/out_loki/config/parsers.conf

@@ -0,0 +1,3 @@
+[PARSER]
+    Name        loki-integration-json
+    Format      json

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"<TEST {'®' * ((sequence % 15) + 1)}>"
+    suffix = "</TEST>"
+    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)

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/<path:subpath>', methods=['POST'])
 def receive_data(subpath=None):
     _record_request()

tests/internal/utils.c

@@ -2,6 +2,7 @@
 
 #include <fluent-bit/flb_info.h>
 #include <fluent-bit/flb_mem.h>
+#include <fluent-bit/flb_simd.h>
 #include <fluent-bit/flb_utils.h>
 #include <stdarg.h>
 #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},