From 7303b69315271686cfc5669805fed49b93857748 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 21 Aug 2026 18:29:01 +0900 Subject: [PATCH 01/14] output: add route-scoped retry payload state Signed-off-by: Hiroshi Hatake --- include/fluent-bit/flb_output.h | 47 +++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/include/fluent-bit/flb_output.h b/include/fluent-bit/flb_output.h index 08eced889dc..499fa5a212d 100644 --- a/include/fluent-bit/flb_output.h +++ b/include/fluent-bit/flb_output.h @@ -600,6 +600,47 @@ struct flb_output_flush { struct mk_list _head; /* Link to flb_task->threads */ }; +static FLB_INLINE void *flb_output_get_retry_context( + struct flb_output_flush *out_flush, + int *records, + size_t *bytes) +{ + void *context; + + flb_task_acquire_lock(out_flush->task); + context = flb_task_get_route_retry_context(out_flush->task, + out_flush->o_ins, + records, bytes); + flb_task_release_lock(out_flush->task); + + return context; +} + +static FLB_INLINE int flb_output_set_retry_context( + struct flb_output_flush *out_flush, + void *context, + void (*destroy)(void *), + int records, + size_t bytes) +{ + int result; + + flb_task_acquire_lock(out_flush->task); + result = flb_task_set_route_retry_context(out_flush->task, + out_flush->o_ins, + context, destroy, + records, bytes); + flb_task_release_lock(out_flush->task); + + return result; +} + +static FLB_INLINE int flb_output_clear_retry_context( + struct flb_output_flush *out_flush) +{ + return flb_output_set_retry_context(out_flush, NULL, NULL, 0, 0); +} + static FLB_INLINE int flb_output_is_threaded(struct flb_output_instance *ins) { return ins->is_threaded; @@ -1272,6 +1313,12 @@ static inline void flb_output_return(int ret, struct flb_coro *co) { bytes = counted_event_chunk->size; flb_task_acquire_lock(task); + if (ret != FLB_OK && + flb_task_get_route_retry_context(task, o_ins, + &records, &bytes) == NULL) { + records = counted_event_chunk->total_events; + bytes = counted_event_chunk->size; + } flb_task_set_route_data(task, o_ins, records, bytes); flb_task_deactivate_route(task, o_ins); flb_task_release_lock(task); From a97454d669e9b41671e273a0a4ebbf8838f688f7 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 21 Aug 2026 18:32:03 +0900 Subject: [PATCH 02/14] task: Add retry contexts Signed-off-by: Hiroshi Hatake --- include/fluent-bit/flb_task.h | 69 +++++++++++++++++++++++++++++++++++ src/flb_task.c | 8 ++++ 2 files changed, 77 insertions(+) diff --git a/include/fluent-bit/flb_task.h b/include/fluent-bit/flb_task.h index de7720d19b8..4ef8b7cd00e 100644 --- a/include/fluent-bit/flb_task.h +++ b/include/fluent-bit/flb_task.h @@ -59,6 +59,10 @@ struct flb_task_route { int status; int records; size_t bytes; + void *retry_context; + void (*retry_context_destroy)(void *); + int retry_records; + size_t retry_bytes; struct flb_output_instance *out; struct mk_list _head; }; @@ -305,6 +309,71 @@ static FLB_INLINE int flb_task_get_route_data( return -1; } +static FLB_INLINE void *flb_task_get_route_retry_context( + struct flb_task *task, + struct flb_output_instance *o_ins, + int *records, + size_t *bytes) +{ + struct mk_list *iterator; + struct flb_task_route *route; + + mk_list_foreach(iterator, &task->routes) { + route = mk_list_entry(iterator, struct flb_task_route, _head); + + if (route->out == o_ins) { + if (records != NULL) { + *records = route->retry_records; + } + if (bytes != NULL) { + *bytes = route->retry_bytes; + } + return route->retry_context; + } + } + + return NULL; +} + +static FLB_INLINE int flb_task_set_route_retry_context( + struct flb_task *task, + struct flb_output_instance *o_ins, + void *context, + void (*destroy)(void *), + int records, + size_t bytes) +{ + struct mk_list *iterator; + struct flb_task_route *route; + + mk_list_foreach(iterator, &task->routes) { + route = mk_list_entry(iterator, struct flb_task_route, _head); + + if (route->out == o_ins) { + if (route->retry_context != NULL && + route->retry_context != context && + route->retry_context_destroy != NULL) { + route->retry_context_destroy(route->retry_context); + } + + route->retry_context = context; + route->retry_context_destroy = destroy; + route->retry_records = records; + route->retry_bytes = bytes; + return 0; + } + } + + return -1; +} + +static FLB_INLINE int flb_task_clear_route_retry_context( + struct flb_task *task, + struct flb_output_instance *o_ins) +{ + return flb_task_set_route_retry_context(task, o_ins, NULL, NULL, 0, 0); +} + static FLB_INLINE void flb_task_activate_route( struct flb_task *task, diff --git a/src/flb_task.c b/src/flb_task.c index 928f75d5bb0..dcefdc66a89 100644 --- a/src/flb_task.c +++ b/src/flb_task.c @@ -277,6 +277,8 @@ void flb_task_retry_destroy(struct flb_task_retry *retry) retry); } + flb_task_clear_route_retry_context(retry->parent, retry->o_ins); + mk_list_del(&retry->_head); flb_free(retry); } @@ -429,6 +431,8 @@ int flb_task_retry_clean(struct flb_task *task, struct flb_output_instance *ins) } } + flb_task_clear_route_retry_context(task, ins); + return -1; } @@ -907,6 +911,10 @@ void flb_task_destroy(struct flb_task *task, int del) /* Remove routes */ mk_list_foreach_safe(head, tmp, &task->routes) { route = mk_list_entry(head, struct flb_task_route, _head); + if (route->retry_context != NULL && + route->retry_context_destroy != NULL) { + route->retry_context_destroy(route->retry_context); + } mk_list_del(&route->_head); flb_free(route); } From fface97e05cbe3487bdcf33e22419b681ea69c63 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 21 Aug 2026 18:32:44 +0900 Subject: [PATCH 03/14] search_bulk: Make consistency of retrying on bulk Signed-off-by: Hiroshi Hatake --- include/fluent-bit/flb_search_bulk.h | 36 ++++ src/CMakeLists.txt | 1 + src/flb_search_bulk.c | 255 +++++++++++++++++++++++++++ 3 files changed, 292 insertions(+) create mode 100644 include/fluent-bit/flb_search_bulk.h create mode 100644 src/flb_search_bulk.c diff --git a/include/fluent-bit/flb_search_bulk.h b/include/fluent-bit/flb_search_bulk.h new file mode 100644 index 00000000000..602a7dafcd0 --- /dev/null +++ b/include/fluent-bit/flb_search_bulk.h @@ -0,0 +1,36 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2026 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + */ + +#ifndef FLB_SEARCH_BULK_H +#define FLB_SEARCH_BULK_H + +#include + +#define FLB_SEARCH_BULK_COMPLETE 0 +#define FLB_SEARCH_BULK_RETRY 1 +#define FLB_SEARCH_BULK_INVALID -1 + +struct flb_search_bulk_retry { + char *payload; + size_t size; + int records; +}; + +int flb_search_bulk_process_response(const char *response, + size_t response_size, + const char *payload, + size_t payload_size, + struct flb_search_bulk_retry **retry); +void flb_search_bulk_retry_destroy(void *data); + +#endif diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 178fe16a534..6e11193d78c 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -47,6 +47,7 @@ set(src flb_engine.c flb_engine_dispatch.c flb_task.c + flb_search_bulk.c flb_unescape.c flb_scheduler.c flb_io.c diff --git a/src/flb_search_bulk.c b/src/flb_search_bulk.c new file mode 100644 index 00000000000..75d234b12cf --- /dev/null +++ b/src/flb_search_bulk.c @@ -0,0 +1,255 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2026 The Fluent Bit Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + */ + +#include +#include +#include + +#include +#include + +static int object_key_equals(msgpack_object key, const char *value, size_t length) +{ + if (key.type != MSGPACK_OBJECT_STR || key.via.str.size != length) { + return FLB_FALSE; + } + + return strncmp(key.via.str.ptr, value, length) == 0; +} + +static int item_is_acknowledged(msgpack_object item) +{ + int index; + int status; + msgpack_object operation; + msgpack_object key; + msgpack_object value; + + if (item.type != MSGPACK_OBJECT_MAP || item.via.map.size != 1) { + return -1; + } + + key = item.via.map.ptr[0].key; + operation = item.via.map.ptr[0].val; + if (key.type != MSGPACK_OBJECT_STR || operation.type != MSGPACK_OBJECT_MAP) { + return -1; + } + + status = -1; + for (index = 0; index < operation.via.map.size; index++) { + value = operation.via.map.ptr[index].val; + if (object_key_equals(operation.via.map.ptr[index].key, + "status", 6) == FLB_TRUE) { + if (value.type != MSGPACK_OBJECT_POSITIVE_INTEGER) { + return -1; + } + status = value.via.u64; + break; + } + } + + if (status < 0) { + return -1; + } + if (status >= 200 && status < 300) { + return FLB_TRUE; + } + if (status == 409 && object_key_equals(key, "create", 6) == FLB_TRUE) { + return FLB_TRUE; + } + + return FLB_FALSE; +} + +static int next_entry(const char *payload, size_t payload_size, + size_t *offset, size_t *entry_start, size_t *entry_size) +{ + const char *line_end; + const char *record_end; + size_t remaining; + + if (*offset >= payload_size) { + return -1; + } + + *entry_start = *offset; + remaining = payload_size - *offset; + line_end = memchr(payload + *offset, '\n', remaining); + if (line_end == NULL) { + return -1; + } + + *offset = (line_end - payload) + 1; + remaining = payload_size - *offset; + record_end = memchr(payload + *offset, '\n', remaining); + if (record_end == NULL) { + return -1; + } + + *offset = (record_end - payload) + 1; + *entry_size = *offset - *entry_start; + return 0; +} + +void flb_search_bulk_retry_destroy(void *data) +{ + struct flb_search_bulk_retry *retry; + + retry = data; + if (retry == NULL) { + return; + } + + flb_free(retry->payload); + flb_free(retry); +} + +int flb_search_bulk_process_response(const char *response, + size_t response_size, + const char *payload, + size_t payload_size, + struct flb_search_bulk_retry **out_retry) +{ + int index; + int result; + int root_type; + int errors_found; + int has_errors; + int acknowledged; + char *packed_response; + size_t packed_size; + size_t unpack_offset; + size_t payload_offset; + size_t entry_start; + size_t entry_size; + msgpack_unpacked unpacked; + msgpack_object root; + msgpack_object items; + msgpack_object key; + msgpack_object value; + struct flb_search_bulk_retry *retry; + + *out_retry = NULL; + packed_response = NULL; + retry = NULL; + items.type = MSGPACK_OBJECT_NIL; + errors_found = FLB_FALSE; + has_errors = FLB_FALSE; + + result = flb_pack_json(response, response_size, + &packed_response, &packed_size, + &root_type, NULL); + if (result != 0) { + return FLB_SEARCH_BULK_INVALID; + } + + msgpack_unpacked_init(&unpacked); + unpack_offset = 0; + result = msgpack_unpack_next(&unpacked, packed_response, + packed_size, &unpack_offset); + if (result != MSGPACK_UNPACK_SUCCESS) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + + root = unpacked.data; + if (root.type != MSGPACK_OBJECT_MAP) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + + for (index = 0; index < root.via.map.size; index++) { + key = root.via.map.ptr[index].key; + value = root.via.map.ptr[index].val; + + if (object_key_equals(key, "errors", 6) == FLB_TRUE) { + if (value.type != MSGPACK_OBJECT_BOOLEAN) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + errors_found = FLB_TRUE; + has_errors = value.via.boolean; + } + else if (object_key_equals(key, "items", 5) == FLB_TRUE) { + if (value.type != MSGPACK_OBJECT_ARRAY) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + items = value; + } + } + + if (errors_found == FLB_FALSE) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + if (has_errors == FLB_FALSE) { + result = FLB_SEARCH_BULK_COMPLETE; + goto done; + } + if (items.type != MSGPACK_OBJECT_ARRAY) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + + retry = flb_calloc(1, sizeof(struct flb_search_bulk_retry)); + if (retry == NULL) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + retry->payload = flb_malloc(payload_size); + if (retry->payload == NULL) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + + payload_offset = 0; + for (index = 0; index < items.via.array.size; index++) { + if (next_entry(payload, payload_size, &payload_offset, + &entry_start, &entry_size) != 0) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + + acknowledged = item_is_acknowledged(items.via.array.ptr[index]); + if (acknowledged < 0) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + if (acknowledged == FLB_FALSE) { + memcpy(retry->payload + retry->size, + payload + entry_start, entry_size); + retry->size += entry_size; + retry->records++; + } + } + + if (payload_offset != payload_size) { + result = FLB_SEARCH_BULK_INVALID; + goto done; + } + if (retry->records == 0) { + result = FLB_SEARCH_BULK_COMPLETE; + goto done; + } + + *out_retry = retry; + retry = NULL; + result = FLB_SEARCH_BULK_RETRY; + + done: + flb_search_bulk_retry_destroy(retry); + msgpack_unpacked_destroy(&unpacked); + flb_free(packed_response); + return result; +} From dcc18fdb0bb2cf48bbb339dbf0ab9ac4c9ad5109 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 21 Aug 2026 18:33:58 +0900 Subject: [PATCH 04/14] out_es: Add a state of retrying bulk Signed-off-by: Hiroshi Hatake --- plugins/out_es/es.c | 98 +++++++++++++++++++++++++++++++-------------- 1 file changed, 68 insertions(+), 30 deletions(-) diff --git a/plugins/out_es/es.c b/plugins/out_es/es.c index 2f362703dc7..aecee3bee5a 100644 --- a/plugins/out_es/es.c +++ b/plugins/out_es/es.c @@ -31,6 +31,7 @@ #include #include #include +#include #include #include @@ -1216,6 +1217,10 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, flb_sds_t signature = NULL; flb_sds_t uri = NULL; int compressed = FLB_FALSE; + void *final_payload_buf; + size_t final_payload_size; + struct flb_search_bulk_retry *retry_payload; + struct flb_search_bulk_retry *next_retry_payload; int compress_gzip; size_t buffer_size; flb_sds_t header_line = NULL; @@ -1230,6 +1235,12 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, int has_aws_auth; #endif + pack = NULL; + out_buf = NULL; + final_payload_buf = NULL; + final_payload_size = 0; + next_retry_payload = NULL; + node_ctx = NULL; if (ctx->ha_mode == FLB_TRUE) { node = flb_upstream_ha_node_get(ctx->ha); @@ -1253,16 +1264,28 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, FLB_OUTPUT_RETURN(FLB_RETRY); } - /* Convert format */ - ret = elasticsearch_format(config, ins, - ctx, node, - event_chunk->type, - event_chunk->tag, flb_sds_len(event_chunk->tag), - event_chunk->data, event_chunk->size, - &out_buf, &out_size); - if (ret != 0) { - flb_upstream_conn_release(u_conn); - FLB_OUTPUT_RETURN(FLB_ERROR); + retry_payload = flb_output_get_retry_context(out_flush, NULL, NULL); + if (retry_payload != NULL) { + out_buf = flb_malloc(retry_payload->size); + if (out_buf == NULL) { + flb_upstream_conn_release(u_conn); + FLB_OUTPUT_RETURN(FLB_RETRY); + } + memcpy(out_buf, retry_payload->payload, retry_payload->size); + out_size = retry_payload->size; + } + else { + /* Convert format */ + ret = elasticsearch_format(config, ins, + ctx, node, + event_chunk->type, + event_chunk->tag, flb_sds_len(event_chunk->tag), + event_chunk->data, event_chunk->size, + &out_buf, &out_size); + if (ret != 0) { + flb_upstream_conn_release(u_conn); + FLB_OUTPUT_RETURN(FLB_ERROR); + } } if (out_size == 0) { @@ -1273,6 +1296,8 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, pack = (char *) out_buf; pack_size = out_size; + final_payload_buf = pack; + final_payload_size = pack_size; /* Should we compress the payload ? */ if (compress_gzip == FLB_TRUE) { @@ -1284,17 +1309,9 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, } else { compressed = FLB_TRUE; + final_payload_buf = out_buf; + final_payload_size = out_size; } - - /* - * The payload buffer is different than pack, means we must be free it. - */ - if (out_buf != pack) { - flb_free(pack); - } - - pack = (char *) out_buf; - pack_size = out_size; } /* Compose HTTP Client request */ @@ -1304,7 +1321,8 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, } c = flb_http_client(u_conn, FLB_HTTP_POST, uri, - pack, pack_size, NULL, 0, NULL, 0); + final_payload_buf, final_payload_size, + NULL, 0, NULL, 0); if (c == NULL) { goto retry; } @@ -1421,17 +1439,34 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, } if (c->resp.payload_size > 0) { - /* - * Elasticsearch payload should be JSON, we convert it to msgpack - * and lookup the 'error' field. - */ - ret = elasticsearch_error_check(ctx, c); - if (ret & FLB_ES_STATUS_SUCCESS) { + ret = flb_search_bulk_process_response(c->resp.payload, + c->resp.payload_size, + pack, pack_size, + &next_retry_payload); + if (ret == FLB_SEARCH_BULK_COMPLETE) { + flb_output_clear_retry_context(out_flush); flb_plg_debug(ctx->ins, "Elasticsearch response\n%s", c->resp.payload); } else { - /* we got an error */ + if (ret == FLB_SEARCH_BULK_RETRY) { + ret = flb_output_set_retry_context( + out_flush, next_retry_payload, + flb_search_bulk_retry_destroy, + next_retry_payload->records, + next_retry_payload->size); + if (ret != 0) { + flb_search_bulk_retry_destroy(next_retry_payload); + } + else { + next_retry_payload = NULL; + } + } + else { + flb_plg_error(ctx->ins, + "invalid Elasticsearch bulk response"); + } + if (ctx->trace_error) { /* * If trace_error is set, trace the actual @@ -1467,6 +1502,9 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, flb_http_client_destroy(c); } flb_free(pack); + if (final_payload_buf != pack) { + flb_free(final_payload_buf); + } flb_upstream_conn_release(u_conn); if (signature) { flb_sds_destroy(signature); @@ -1483,8 +1521,8 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, if (signature != NULL) { flb_sds_destroy(signature); } - if (out_buf != pack) { - flb_free(out_buf); + if (final_payload_buf != pack) { + flb_free(final_payload_buf); } flb_sds_destroy(uri); From 5a228591923640ce541664e63faf05f44d4fd72d Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 21 Aug 2026 18:34:30 +0900 Subject: [PATCH 05/14] out_opensearch: Add a state of retrying bulk Signed-off-by: Hiroshi Hatake --- plugins/out_opensearch/opensearch.c | 70 ++++++++++++++++++++++++----- 1 file changed, 58 insertions(+), 12 deletions(-) diff --git a/plugins/out_opensearch/opensearch.c b/plugins/out_opensearch/opensearch.c index 5541deea0da..a47f15bf4b2 100644 --- a/plugins/out_opensearch/opensearch.c +++ b/plugins/out_opensearch/opensearch.c @@ -29,6 +29,7 @@ #include #include #include +#include #include #include @@ -939,11 +940,13 @@ static void cb_opensearch_flush(struct flb_event_chunk *event_chunk, size_t b_sent; struct flb_opensearch *ctx = out_context; struct flb_connection *u_conn; - struct flb_http_client *c; + struct flb_http_client *c = NULL; flb_sds_t signature = NULL; int compressed = FLB_FALSE; void *final_payload_buf = NULL; size_t final_payload_size = 0; + struct flb_search_bulk_retry *retry_payload; + struct flb_search_bulk_retry *next_retry_payload = NULL; /* Get upstream connection */ u_conn = flb_upstream_conn_get(ctx->u); @@ -951,8 +954,19 @@ static void cb_opensearch_flush(struct flb_event_chunk *event_chunk, FLB_OUTPUT_RETURN(FLB_RETRY); } - /* Convert format */ - if (event_chunk->type == FLB_EVENT_TYPE_TRACES) { + retry_payload = flb_output_get_retry_context(out_flush, NULL, NULL); + if (retry_payload != NULL) { + pack = flb_sds_create_len(retry_payload->payload, + retry_payload->size); + if (pack == NULL) { + flb_upstream_conn_release(u_conn); + FLB_OUTPUT_RETURN(FLB_RETRY); + } + ret = 0; + out_buf = pack; + out_size = flb_sds_len(pack); + } + else if (event_chunk->type == FLB_EVENT_TYPE_TRACES) { pack = flb_msgpack_raw_to_json_sds(event_chunk->data, event_chunk->size, config->json_escape_unicode); if (pack) { @@ -1008,6 +1022,9 @@ static void cb_opensearch_flush(struct flb_event_chunk *event_chunk, /* Compose HTTP Client request */ c = flb_http_client(u_conn, FLB_HTTP_POST, ctx->uri, final_payload_buf, final_payload_size, NULL, 0, NULL, 0); + if (c == NULL) { + goto retry; + } flb_http_buffer_size(c, ctx->buffer_size); @@ -1072,13 +1089,37 @@ static void cb_opensearch_flush(struct flb_event_chunk *event_chunk, } if (c->resp.payload_size > 0) { - /* - * OpenSearch payload should be JSON, we convert it to msgpack - * and lookup the 'error' field. - */ - ret = opensearch_error_check(ctx, c); - if (ret == FLB_TRUE) { - /* we got an error */ + if (event_chunk->type == FLB_EVENT_TYPE_LOGS) { + ret = flb_search_bulk_process_response(c->resp.payload, + c->resp.payload_size, + pack, pack_size, + &next_retry_payload); + } + else if (opensearch_error_check(ctx, c) == FLB_TRUE) { + ret = FLB_SEARCH_BULK_INVALID; + } + else { + ret = FLB_SEARCH_BULK_COMPLETE; + } + if (ret != FLB_SEARCH_BULK_COMPLETE) { + if (ret == FLB_SEARCH_BULK_RETRY) { + ret = flb_output_set_retry_context( + out_flush, next_retry_payload, + flb_search_bulk_retry_destroy, + next_retry_payload->records, + next_retry_payload->size); + if (ret != 0) { + flb_search_bulk_retry_destroy(next_retry_payload); + } + else { + next_retry_payload = NULL; + } + } + else { + flb_plg_error(ctx->ins, + "invalid OpenSearch bulk response"); + } + if (ctx->trace_error) { /* * If trace_error is set, trace the actual @@ -1108,6 +1149,7 @@ static void cb_opensearch_flush(struct flb_event_chunk *event_chunk, goto retry; } else { + flb_output_clear_retry_context(out_flush); flb_plg_debug(ctx->ins, "OpenSearch response\n%s", c->resp.payload); } @@ -1122,7 +1164,9 @@ static void cb_opensearch_flush(struct flb_event_chunk *event_chunk, } /* Cleanup */ - flb_http_client_destroy(c); + if (c != NULL) { + flb_http_client_destroy(c); + } if (final_payload_buf != pack) { flb_free(final_payload_buf); @@ -1137,7 +1181,9 @@ static void cb_opensearch_flush(struct flb_event_chunk *event_chunk, /* Issue a retry */ retry: - flb_http_client_destroy(c); + if (c != NULL) { + flb_http_client_destroy(c); + } flb_sds_destroy(pack); if (final_payload_buf != pack) { From 210d9825631d0efe181e4efe12cd1a6893e657ce Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 21 Aug 2026 18:35:48 +0900 Subject: [PATCH 06/14] tests: internal: Add test cases for retrying contexts Signed-off-by: Hiroshi Hatake --- tests/internal/task_map.c | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/tests/internal/task_map.c b/tests/internal/task_map.c index 05ff5d107ae..26fbf50d2ef 100644 --- a/tests/internal/task_map.c +++ b/tests/internal/task_map.c @@ -18,6 +18,14 @@ struct test_ctx { struct mk_event_loop *evl; }; +static int retry_context_destroy_count; + +static void retry_context_destroy(void *data) +{ + retry_context_destroy_count++; + flb_free(data); +} + struct test_ctx* test_ctx_create() { struct test_ctx *ret_ctx = NULL; @@ -126,6 +134,8 @@ void test_task_route_data_preserved_across_retry() struct flb_task_route *route_a; struct flb_task_route *route_b; struct flb_task_retry *retry; + void *retry_context_a; + void *retry_context_b; ctx = test_ctx_create(); if (!TEST_CHECK(ctx != NULL)) { @@ -169,6 +179,18 @@ void test_task_route_data_preserved_across_retry() mk_list_add(&route_a->_head, &task->routes); mk_list_add(&route_b->_head, &task->routes); + retry_context_destroy_count = 0; + retry_context_a = flb_malloc(1); + retry_context_b = flb_malloc(1); + TEST_CHECK(retry_context_a != NULL); + TEST_CHECK(retry_context_b != NULL); + flb_task_set_route_retry_context(task, &out_a, + retry_context_a, + retry_context_destroy, 1, 111); + flb_task_set_route_retry_context(task, &out_b, + retry_context_b, + retry_context_destroy, 2, 222); + flb_task_set_route_data(task, &out_a, 1, 111); ret = flb_task_get_route_data(task, &out_a, &records, &bytes); TEST_CHECK(ret == 0); @@ -193,7 +215,13 @@ void test_task_route_data_preserved_across_retry() TEST_CHECK(bytes == 111); flb_task_retry_clean(task, &out_a); + TEST_CHECK(retry_context_destroy_count == 1); + TEST_CHECK(flb_task_get_route_retry_context(task, &out_a, + NULL, NULL) == NULL); + TEST_CHECK(flb_task_get_route_retry_context(task, &out_b, + NULL, NULL) == retry_context_b); flb_task_destroy(task, FLB_TRUE); + TEST_CHECK(retry_context_destroy_count == 2); test_ctx_destroy(ctx); } From e995feaf7a9d96ab2c0ad059e4a4720847adec26 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 21 Aug 2026 18:36:20 +0900 Subject: [PATCH 07/14] tests: internal: Add test cases for retrying state of bulk operations Signed-off-by: Hiroshi Hatake --- tests/internal/CMakeLists.txt | 1 + tests/internal/search_bulk.c | 108 ++++++++++++++++++++++++++++++++++ 2 files changed, 109 insertions(+) create mode 100644 tests/internal/search_bulk.c diff --git a/tests/internal/CMakeLists.txt b/tests/internal/CMakeLists.txt index b66e561505c..94dce331d68 100644 --- a/tests/internal/CMakeLists.txt +++ b/tests/internal/CMakeLists.txt @@ -65,6 +65,7 @@ set(UNIT_TESTS_FILES storage_dlq.c engine_adaptive_flush.c engine_dispatch.c + search_bulk.c ) if(FLB_OUT_AZURE_BLOB) diff --git a/tests/internal/search_bulk.c b/tests/internal/search_bulk.c new file mode 100644 index 00000000000..2162b55114a --- /dev/null +++ b/tests/internal/search_bulk.c @@ -0,0 +1,108 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +#include + +#include + +#include "flb_tests_internal.h" + +#define BULK_PAYLOAD \ + "{\"create\":{\"_index\":\"logs\",\"_id\":\"one\"}}\n" \ + "{\"message\":\"one\"}\n" \ + "{\"create\":{\"_index\":\"logs\",\"_id\":\"two\"}}\n" \ + "{\"message\":\"two\"}\n" \ + "{\"create\":{\"_index\":\"logs\",\"_id\":\"three\"}}\n" \ + "{\"message\":\"three\"}\n" + +#define SECOND_ENTRY \ + "{\"create\":{\"_index\":\"logs\",\"_id\":\"two\"}}\n" \ + "{\"message\":\"two\"}\n" + +static void test_mixed_response_keeps_only_unresolved(void) +{ + int result; + const char *response; + struct flb_search_bulk_retry *retry; + + response = "{\"errors\":true,\"items\":[" + "{\"create\":{\"status\":201}}," + "{\"create\":{\"status\":429}}," + "{\"create\":{\"status\":409}}]}"; + + result = flb_search_bulk_process_response(response, strlen(response), + BULK_PAYLOAD, + strlen(BULK_PAYLOAD), + &retry); + TEST_CHECK(result == FLB_SEARCH_BULK_RETRY); + TEST_CHECK(retry != NULL); + TEST_CHECK(retry->records == 1); + TEST_CHECK(retry->size == strlen(SECOND_ENTRY)); + TEST_CHECK(memcmp(retry->payload, SECOND_ENTRY, retry->size) == 0); + flb_search_bulk_retry_destroy(retry); +} + +static void test_create_conflicts_are_complete(void) +{ + int result; + const char *response; + struct flb_search_bulk_retry *retry; + + response = "{\"errors\":true,\"items\":[" + "{\"create\":{\"status\":409}}," + "{\"create\":{\"status\":409}}," + "{\"create\":{\"status\":409}}]}"; + + result = flb_search_bulk_process_response(response, strlen(response), + BULK_PAYLOAD, + strlen(BULK_PAYLOAD), + &retry); + TEST_CHECK(result == FLB_SEARCH_BULK_COMPLETE); + TEST_CHECK(retry == NULL); +} + +static void test_update_conflict_is_retried(void) +{ + int result; + const char *payload; + const char *response; + struct flb_search_bulk_retry *retry; + + payload = "{\"update\":{\"_index\":\"logs\",\"_id\":\"one\"}}\n" + "{\"doc\":{\"message\":\"one\"}}\n"; + response = "{\"errors\":true,\"items\":[" + "{\"update\":{\"status\":409}}]}"; + + result = flb_search_bulk_process_response(response, strlen(response), + payload, strlen(payload), + &retry); + TEST_CHECK(result == FLB_SEARCH_BULK_RETRY); + TEST_CHECK(retry != NULL); + TEST_CHECK(retry->records == 1); + TEST_CHECK(retry->size == strlen(payload)); + flb_search_bulk_retry_destroy(retry); +} + +static void test_item_count_mismatch_is_invalid(void) +{ + int result; + const char *response; + struct flb_search_bulk_retry *retry; + + response = "{\"errors\":true,\"items\":[" + "{\"create\":{\"status\":429}}]}"; + + result = flb_search_bulk_process_response(response, strlen(response), + BULK_PAYLOAD, + strlen(BULK_PAYLOAD), + &retry); + TEST_CHECK(result == FLB_SEARCH_BULK_INVALID); + TEST_CHECK(retry == NULL); +} + +TEST_LIST = { + {"mixed_response_keeps_only_unresolved", test_mixed_response_keeps_only_unresolved}, + {"create_conflicts_are_complete", test_create_conflicts_are_complete}, + {"update_conflict_is_retried", test_update_conflict_is_retried}, + {"item_count_mismatch_is_invalid", test_item_count_mismatch_is_invalid}, + {NULL, NULL} +}; From b3acc36a9bfaeb048fe9d44e779d3d1442ec6433 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 21 Aug 2026 18:37:06 +0900 Subject: [PATCH 08/14] tests: integration: Add senarios for retying bulk operations Signed-off-by: Hiroshi Hatake --- .../config/out_es_partial_bulk_retry.yaml | 25 +++++++++ .../out_opensearch_partial_bulk_retry.yaml | 25 +++++++++ .../test_out_es_ndjson_action_line_001.py | 56 +++++++++++++++++-- 3 files changed, 102 insertions(+), 4 deletions(-) create mode 100644 tests/integration/scenarios/out_es/config/out_es_partial_bulk_retry.yaml create mode 100644 tests/integration/scenarios/out_es/config/out_opensearch_partial_bulk_retry.yaml diff --git a/tests/integration/scenarios/out_es/config/out_es_partial_bulk_retry.yaml b/tests/integration/scenarios/out_es/config/out_es_partial_bulk_retry.yaml new file mode 100644 index 00000000000..c2493c886f6 --- /dev/null +++ b/tests/integration/scenarios/out_es/config/out_es_partial_bulk_retry.yaml @@ -0,0 +1,25 @@ +service: + flush: 1 + grace: 1 + log_level: info + scheduler.base: 1 + scheduler.cap: 1 + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: dummy + tag: out_es_partial_bulk_retry + dummy: '{"message":"partial retry"}' + copies: 3 + samples: 1 + + outputs: + - name: es + match: out_es_partial_bulk_retry + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + index: fluent-bit + suppress_type_name: on + retry_limit: 2 diff --git a/tests/integration/scenarios/out_es/config/out_opensearch_partial_bulk_retry.yaml b/tests/integration/scenarios/out_es/config/out_opensearch_partial_bulk_retry.yaml new file mode 100644 index 00000000000..59c567a5ae8 --- /dev/null +++ b/tests/integration/scenarios/out_es/config/out_opensearch_partial_bulk_retry.yaml @@ -0,0 +1,25 @@ +service: + flush: 1 + grace: 1 + log_level: info + scheduler.base: 1 + scheduler.cap: 1 + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: dummy + tag: out_opensearch_partial_bulk_retry + dummy: '{"message":"partial retry"}' + copies: 3 + samples: 1 + + outputs: + - name: opensearch + match: out_opensearch_partial_bulk_retry + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + index: fluent-bit + suppress_type_name: on + retry_limit: 2 diff --git a/tests/integration/scenarios/out_es/tests/test_out_es_ndjson_action_line_001.py b/tests/integration/scenarios/out_es/tests/test_out_es_ndjson_action_line_001.py index f773cc1611a..0b140750918 100644 --- a/tests/integration/scenarios/out_es/tests/test_out_es_ndjson_action_line_001.py +++ b/tests/integration/scenarios/out_es/tests/test_out_es_ndjson_action_line_001.py @@ -14,6 +14,8 @@ class _BulkCaptureHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + def log_message(self, fmt, *args): return @@ -29,7 +31,12 @@ def do_POST(self): } ) - response = b'{"errors":false,"items":[{"create":{"status":201}}]}' + if self.server.response_factory is None: + response = b'{"errors":false,"items":[{"create":{"status":201}}]}' + else: + response = self.server.response_factory( + len(self.server.requests), self.server.requests[-1] + ) self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(response))) @@ -41,18 +48,20 @@ class _BulkCaptureServer(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True - def __init__(self, address): + def __init__(self, address, response_factory=None): super().__init__(address, _BulkCaptureHandler) self.requests = [] + self.response_factory = response_factory class Service: - def __init__(self, config_file): + def __init__(self, config_file, response_factory=None): self.config_file = os.path.abspath( os.path.join(os.path.dirname(__file__), "../config", config_file) ) self.bulk_server = None self.bulk_server_thread = None + self.response_factory = response_factory self.service = FluentBitTestService( self.config_file, pre_start=self._start_receiver, @@ -60,7 +69,10 @@ def __init__(self, config_file): ) def _start_receiver(self, service): - self.bulk_server = _BulkCaptureServer(("127.0.0.1", service.test_suite_http_port)) + self.bulk_server = _BulkCaptureServer( + ("127.0.0.1", service.test_suite_http_port), + self.response_factory, + ) self.bulk_server_thread = threading.Thread( target=self.bulk_server.serve_forever, daemon=True, @@ -152,6 +164,22 @@ def _bulk_actions(requests): return actions +def _partial_bulk_response(request_number, request): + action_count = len(_bulk_action_lines(request["body"])) + + if request_number == 1: + assert action_count == 3 + return ( + b'{"errors":true,"items":[' + b'{"create":{"status":201}},' + b'{"create":{"status":429}},' + b'{"create":{"status":409}}]}' + ) + + assert action_count == 1 + return b'{"errors":false,"items":[{"create":{"status":201}}]}' + + @pytest.mark.parametrize( "config_file", [ @@ -198,3 +226,23 @@ def test_unsafe_required_id_key_does_not_emit_idless_update(config_file): assert all(request["path"].startswith("/_bulk") for request in requests_seen) assert len(actions) == 1 assert updates == [{"_index": "fluent-bit", "_id": SAFE_UPDATE_ID}] + + +@pytest.mark.parametrize( + "config_file", + [ + "out_es_partial_bulk_retry.yaml", + "out_opensearch_partial_bulk_retry.yaml", + ], +) +def test_partial_bulk_retry_sends_only_unresolved_records(config_file): + service = Service(config_file, response_factory=_partial_bulk_response) + + try: + service.start() + requests_seen = service.wait_for_requests(2) + finally: + service.stop() + + assert len(_bulk_action_lines(requests_seen[0]["body"])) == 3 + assert len(_bulk_action_lines(requests_seen[1]["body"])) == 1 From 1040cb373c26d20176e1dbc862a65d3db33a8504 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 24 Aug 2026 19:30:30 +0900 Subject: [PATCH 09/14] search_bulk: Add acknowledgment handler Signed-off-by: Hiroshi Hatake --- include/fluent-bit/flb_search_bulk.h | 4 +++ src/flb_search_bulk.c | 42 +++++++++++++++++++++++++--- 2 files changed, 42 insertions(+), 4 deletions(-) diff --git a/include/fluent-bit/flb_search_bulk.h b/include/fluent-bit/flb_search_bulk.h index 602a7dafcd0..76695f0f647 100644 --- a/include/fluent-bit/flb_search_bulk.h +++ b/include/fluent-bit/flb_search_bulk.h @@ -20,6 +20,9 @@ #define FLB_SEARCH_BULK_RETRY 1 #define FLB_SEARCH_BULK_INVALID -1 +#define FLB_SEARCH_BULK_ACK_CREATE_CONFLICTS 0 +#define FLB_SEARCH_BULK_ACK_ALL_CONFLICTS 1 + struct flb_search_bulk_retry { char *payload; size_t size; @@ -30,6 +33,7 @@ int flb_search_bulk_process_response(const char *response, size_t response_size, const char *payload, size_t payload_size, + int acknowledge_all_conflicts, struct flb_search_bulk_retry **retry); void flb_search_bulk_retry_destroy(void *data); diff --git a/src/flb_search_bulk.c b/src/flb_search_bulk.c index 75d234b12cf..7135dee2eaa 100644 --- a/src/flb_search_bulk.c +++ b/src/flb_search_bulk.c @@ -27,7 +27,26 @@ static int object_key_equals(msgpack_object key, const char *value, size_t lengt return strncmp(key.via.str.ptr, value, length) == 0; } -static int item_is_acknowledged(msgpack_object item) +static int response_contains(const char *response, size_t response_size, + const char *value, size_t value_size) +{ + size_t index; + + if (value_size > response_size) { + return FLB_FALSE; + } + + for (index = 0; index <= response_size - value_size; index++) { + if (memcmp(response + index, value, value_size) == 0) { + return FLB_TRUE; + } + } + + return FLB_FALSE; +} + +static int item_is_acknowledged(msgpack_object item, + int acknowledge_all_conflicts) { int index; int status; @@ -64,8 +83,11 @@ static int item_is_acknowledged(msgpack_object item) if (status >= 200 && status < 300) { return FLB_TRUE; } - if (status == 409 && object_key_equals(key, "create", 6) == FLB_TRUE) { - return FLB_TRUE; + if (status == 409) { + if (acknowledge_all_conflicts == FLB_TRUE || + object_key_equals(key, "create", 6) == FLB_TRUE) { + return FLB_TRUE; + } } return FLB_FALSE; @@ -118,6 +140,7 @@ int flb_search_bulk_process_response(const char *response, size_t response_size, const char *payload, size_t payload_size, + int acknowledge_all_conflicts, struct flb_search_bulk_retry **out_retry) { int index; @@ -150,6 +173,16 @@ int flb_search_bulk_process_response(const char *response, &packed_response, &packed_size, &root_type, NULL); if (result != 0) { + /* + * A successful bulk response can exceed the configured HTTP response + * buffer. Preserve the success marker available at the start of the + * bounded response instead of retrying an already accepted batch. + */ + if (response_contains(response, response_size, + "\"errors\":false,\"items\":[", + sizeof("\"errors\":false,\"items\":[") - 1) == FLB_TRUE) { + return FLB_SEARCH_BULK_COMPLETE; + } return FLB_SEARCH_BULK_INVALID; } @@ -221,7 +254,8 @@ int flb_search_bulk_process_response(const char *response, goto done; } - acknowledged = item_is_acknowledged(items.via.array.ptr[index]); + acknowledged = item_is_acknowledged(items.via.array.ptr[index], + acknowledge_all_conflicts); if (acknowledged < 0) { result = FLB_SEARCH_BULK_INVALID; goto done; From abbe359011907cb8fd2a41be9e4784525aa0dac9 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 24 Aug 2026 19:38:42 +0900 Subject: [PATCH 10/14] out_es: Preserve acknowledgement behavior Signed-off-by: Hiroshi Hatake --- plugins/out_es/es.c | 2 ++ 1 file changed, 2 insertions(+) diff --git a/plugins/out_es/es.c b/plugins/out_es/es.c index aecee3bee5a..0d6316c8a35 100644 --- a/plugins/out_es/es.c +++ b/plugins/out_es/es.c @@ -1439,9 +1439,11 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, } if (c->resp.payload_size > 0) { + /* Only create conflicts confirm that the document already exists. */ ret = flb_search_bulk_process_response(c->resp.payload, c->resp.payload_size, pack, pack_size, + FLB_SEARCH_BULK_ACK_CREATE_CONFLICTS, &next_retry_payload); if (ret == FLB_SEARCH_BULK_COMPLETE) { flb_output_clear_retry_context(out_flush); From 952864d2e3b5753120f50265bb070acfa3ebc594 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 24 Aug 2026 19:39:05 +0900 Subject: [PATCH 11/14] out_opensearch: Preserve acknowledgement behavior Signed-off-by: Hiroshi Hatake --- plugins/out_opensearch/opensearch.c | 2 ++ 1 file changed, 2 insertions(+) diff --git a/plugins/out_opensearch/opensearch.c b/plugins/out_opensearch/opensearch.c index a47f15bf4b2..b2848bd22b4 100644 --- a/plugins/out_opensearch/opensearch.c +++ b/plugins/out_opensearch/opensearch.c @@ -1090,9 +1090,11 @@ static void cb_opensearch_flush(struct flb_event_chunk *event_chunk, if (c->resp.payload_size > 0) { if (event_chunk->type == FLB_EVENT_TYPE_LOGS) { + /* Preserve the existing behavior that acknowledges all conflicts. */ ret = flb_search_bulk_process_response(c->resp.payload, c->resp.payload_size, pack, pack_size, + FLB_SEARCH_BULK_ACK_ALL_CONFLICTS, &next_retry_payload); } else if (opensearch_error_check(ctx, c) == FLB_TRUE) { From 842a39723990f87b9637baf930f5d0306884b17c Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 24 Aug 2026 19:32:14 +0900 Subject: [PATCH 12/14] tests: internal: Add test cases for preserving acknowledgement behavior Signed-off-by: Hiroshi Hatake --- tests/internal/search_bulk.c | 46 ++++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/tests/internal/search_bulk.c b/tests/internal/search_bulk.c index 2162b55114a..971a685ca16 100644 --- a/tests/internal/search_bulk.c +++ b/tests/internal/search_bulk.c @@ -32,6 +32,7 @@ static void test_mixed_response_keeps_only_unresolved(void) result = flb_search_bulk_process_response(response, strlen(response), BULK_PAYLOAD, strlen(BULK_PAYLOAD), + FLB_SEARCH_BULK_ACK_CREATE_CONFLICTS, &retry); TEST_CHECK(result == FLB_SEARCH_BULK_RETRY); TEST_CHECK(retry != NULL); @@ -55,6 +56,7 @@ static void test_create_conflicts_are_complete(void) result = flb_search_bulk_process_response(response, strlen(response), BULK_PAYLOAD, strlen(BULK_PAYLOAD), + FLB_SEARCH_BULK_ACK_CREATE_CONFLICTS, &retry); TEST_CHECK(result == FLB_SEARCH_BULK_COMPLETE); TEST_CHECK(retry == NULL); @@ -74,6 +76,7 @@ static void test_update_conflict_is_retried(void) result = flb_search_bulk_process_response(response, strlen(response), payload, strlen(payload), + FLB_SEARCH_BULK_ACK_CREATE_CONFLICTS, &retry); TEST_CHECK(result == FLB_SEARCH_BULK_RETRY); TEST_CHECK(retry != NULL); @@ -82,6 +85,44 @@ static void test_update_conflict_is_retried(void) flb_search_bulk_retry_destroy(retry); } +static void test_update_conflict_is_complete_when_all_conflicts_are_acknowledged(void) +{ + int result; + const char *payload; + const char *response; + struct flb_search_bulk_retry *retry; + + payload = "{\"update\":{\"_index\":\"logs\",\"_id\":\"one\"}}\n" + "{\"doc\":{\"message\":\"one\"}}\n"; + response = "{\"errors\":true,\"items\":[" + "{\"update\":{\"status\":409}}]}"; + + result = flb_search_bulk_process_response(response, strlen(response), + payload, strlen(payload), + FLB_SEARCH_BULK_ACK_ALL_CONFLICTS, + &retry); + TEST_CHECK(result == FLB_SEARCH_BULK_COMPLETE); + TEST_CHECK(retry == NULL); +} + +static void test_truncated_success_response_is_complete(void) +{ + int result; + const char *response; + struct flb_search_bulk_retry *retry; + + response = "{\"took\":1,\"errors\":false,\"items\":[" + "{\"create\":{\"status\":201}}"; + + result = flb_search_bulk_process_response(response, strlen(response), + BULK_PAYLOAD, + strlen(BULK_PAYLOAD), + FLB_SEARCH_BULK_ACK_CREATE_CONFLICTS, + &retry); + TEST_CHECK(result == FLB_SEARCH_BULK_COMPLETE); + TEST_CHECK(retry == NULL); +} + static void test_item_count_mismatch_is_invalid(void) { int result; @@ -94,6 +135,7 @@ static void test_item_count_mismatch_is_invalid(void) result = flb_search_bulk_process_response(response, strlen(response), BULK_PAYLOAD, strlen(BULK_PAYLOAD), + FLB_SEARCH_BULK_ACK_CREATE_CONFLICTS, &retry); TEST_CHECK(result == FLB_SEARCH_BULK_INVALID); TEST_CHECK(retry == NULL); @@ -103,6 +145,10 @@ TEST_LIST = { {"mixed_response_keeps_only_unresolved", test_mixed_response_keeps_only_unresolved}, {"create_conflicts_are_complete", test_create_conflicts_are_complete}, {"update_conflict_is_retried", test_update_conflict_is_retried}, + {"update_conflict_is_complete_when_all_conflicts_are_acknowledged", + test_update_conflict_is_complete_when_all_conflicts_are_acknowledged}, + {"truncated_success_response_is_complete", + test_truncated_success_response_is_complete}, {"item_count_mismatch_is_invalid", test_item_count_mismatch_is_invalid}, {NULL, NULL} }; From eed46dad9269445f1f847780ae86b3fdd4a12091 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 24 Aug 2026 19:45:52 +0900 Subject: [PATCH 13/14] search_bulk: Validate the fallback marker at the top level Signed-off-by: Hiroshi Hatake --- src/flb_search_bulk.c | 289 ++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 280 insertions(+), 9 deletions(-) diff --git a/src/flb_search_bulk.c b/src/flb_search_bulk.c index 7135dee2eaa..ab6815ced87 100644 --- a/src/flb_search_bulk.c +++ b/src/flb_search_bulk.c @@ -27,19 +27,292 @@ static int object_key_equals(msgpack_object key, const char *value, size_t lengt return strncmp(key.via.str.ptr, value, length) == 0; } -static int response_contains(const char *response, size_t response_size, - const char *value, size_t value_size) +static void json_skip_whitespace(const char *json, size_t size, size_t *offset) { + while (*offset < size) { + if (json[*offset] != ' ' && json[*offset] != '\t' && + json[*offset] != '\r' && json[*offset] != '\n') { + break; + } + (*offset)++; + } +} + +static int json_is_hexadecimal(char value) +{ + if ((value >= '0' && value <= '9') || + (value >= 'a' && value <= 'f') || + (value >= 'A' && value <= 'F')) { + return FLB_TRUE; + } + + return FLB_FALSE; +} + +static int json_scan_string(const char *json, size_t size, size_t *offset, + size_t *content_start, size_t *content_size) +{ + char value; size_t index; + size_t start; + + if (*offset >= size || json[*offset] != '"') { + return -1; + } + + start = *offset + 1; + index = start; + while (index < size) { + value = json[index]; + if (value == '"') { + if (content_start != NULL) { + *content_start = start; + } + if (content_size != NULL) { + *content_size = index - start; + } + *offset = index + 1; + return 0; + } + if ((unsigned char) value < 0x20) { + return -1; + } + if (value == '\\') { + index++; + if (index >= size || strchr("\"\\/bfnrtu", json[index]) == NULL) { + return -1; + } + if (json[index] == 'u') { + if (index + 4 >= size) { + return -1; + } + if (json_is_hexadecimal(json[index + 1]) == FLB_FALSE || + json_is_hexadecimal(json[index + 2]) == FLB_FALSE || + json_is_hexadecimal(json[index + 3]) == FLB_FALSE || + json_is_hexadecimal(json[index + 4]) == FLB_FALSE) { + return -1; + } + index += 4; + } + } + index++; + } - if (value_size > response_size) { + return -1; +} + +static int json_skip_value(const char *json, size_t size, + size_t *offset, int depth); + +static int json_skip_object(const char *json, size_t size, + size_t *offset, int depth) +{ + if (depth > 64 || *offset >= size || json[*offset] != '{') { + return -1; + } + + (*offset)++; + json_skip_whitespace(json, size, offset); + if (*offset < size && json[*offset] == '}') { + (*offset)++; + return 0; + } + + while (*offset < size) { + if (json_scan_string(json, size, offset, NULL, NULL) != 0) { + return -1; + } + json_skip_whitespace(json, size, offset); + if (*offset >= size || json[*offset] != ':') { + return -1; + } + (*offset)++; + if (json_skip_value(json, size, offset, depth + 1) != 0) { + return -1; + } + json_skip_whitespace(json, size, offset); + if (*offset >= size) { + return -1; + } + if (json[*offset] == '}') { + (*offset)++; + return 0; + } + if (json[*offset] != ',') { + return -1; + } + (*offset)++; + json_skip_whitespace(json, size, offset); + } + + return -1; +} + +static int json_skip_array(const char *json, size_t size, + size_t *offset, int depth) +{ + if (depth > 64 || *offset >= size || json[*offset] != '[') { + return -1; + } + + (*offset)++; + json_skip_whitespace(json, size, offset); + if (*offset < size && json[*offset] == ']') { + (*offset)++; + return 0; + } + + while (*offset < size) { + if (json_skip_value(json, size, offset, depth + 1) != 0) { + return -1; + } + json_skip_whitespace(json, size, offset); + if (*offset >= size) { + return -1; + } + if (json[*offset] == ']') { + (*offset)++; + return 0; + } + if (json[*offset] != ',') { + return -1; + } + (*offset)++; + json_skip_whitespace(json, size, offset); + } + + return -1; +} + +static int json_skip_number(const char *json, size_t size, size_t *offset) +{ + size_t index; + + index = *offset; + if (index < size && json[index] == '-') { + index++; + } + if (index >= size) { + return -1; + } + if (json[index] == '0') { + index++; + } + else { + if (json[index] < '1' || json[index] > '9') { + return -1; + } + while (index < size && json[index] >= '0' && json[index] <= '9') { + index++; + } + } + if (index < size && json[index] == '.') { + index++; + if (index >= size || json[index] < '0' || json[index] > '9') { + return -1; + } + while (index < size && json[index] >= '0' && json[index] <= '9') { + index++; + } + } + if (index < size && (json[index] == 'e' || json[index] == 'E')) { + index++; + if (index < size && (json[index] == '+' || json[index] == '-')) { + index++; + } + if (index >= size || json[index] < '0' || json[index] > '9') { + return -1; + } + while (index < size && json[index] >= '0' && json[index] <= '9') { + index++; + } + } + + *offset = index; + return 0; +} + +static int json_skip_value(const char *json, size_t size, + size_t *offset, int depth) +{ + json_skip_whitespace(json, size, offset); + if (*offset >= size) { + return -1; + } + + if (json[*offset] == '"') { + return json_scan_string(json, size, offset, NULL, NULL); + } + if (json[*offset] == '{') { + return json_skip_object(json, size, offset, depth); + } + if (json[*offset] == '[') { + return json_skip_array(json, size, offset, depth); + } + if (size - *offset >= 4 && + (memcmp(json + *offset, "true", 4) == 0 || + memcmp(json + *offset, "null", 4) == 0)) { + *offset += 4; + return 0; + } + if (size - *offset >= 5 && memcmp(json + *offset, "false", 5) == 0) { + *offset += 5; + return 0; + } + + return json_skip_number(json, size, offset); +} + +static int top_level_errors_is_false(const char *json, size_t size) +{ + size_t offset; + size_t key_start; + size_t key_size; + size_t value_end; + + offset = 0; + json_skip_whitespace(json, size, &offset); + if (offset >= size || json[offset] != '{') { return FLB_FALSE; } - for (index = 0; index <= response_size - value_size; index++) { - if (memcmp(response + index, value, value_size) == 0) { - return FLB_TRUE; + offset++; + json_skip_whitespace(json, size, &offset); + while (offset < size && json[offset] != '}') { + if (json_scan_string(json, size, &offset, + &key_start, &key_size) != 0) { + return FLB_FALSE; + } + json_skip_whitespace(json, size, &offset); + if (offset >= size || json[offset] != ':') { + return FLB_FALSE; + } + offset++; + json_skip_whitespace(json, size, &offset); + + if (key_size == 6 && memcmp(json + key_start, "errors", 6) == 0) { + if (size - offset < 5 || memcmp(json + offset, "false", 5) != 0) { + return FLB_FALSE; + } + + value_end = offset + 5; + json_skip_whitespace(json, size, &value_end); + if (value_end == size || json[value_end] == ',' || + json[value_end] == '}') { + return FLB_TRUE; + } + return FLB_FALSE; + } + + if (json_skip_value(json, size, &offset, 0) != 0) { + return FLB_FALSE; + } + json_skip_whitespace(json, size, &offset); + if (offset >= size || json[offset] != ',') { + return FLB_FALSE; } + offset++; + json_skip_whitespace(json, size, &offset); } return FLB_FALSE; @@ -178,9 +451,7 @@ int flb_search_bulk_process_response(const char *response, * buffer. Preserve the success marker available at the start of the * bounded response instead of retrying an already accepted batch. */ - if (response_contains(response, response_size, - "\"errors\":false,\"items\":[", - sizeof("\"errors\":false,\"items\":[") - 1) == FLB_TRUE) { + if (top_level_errors_is_false(response, response_size) == FLB_TRUE) { return FLB_SEARCH_BULK_COMPLETE; } return FLB_SEARCH_BULK_INVALID; From 739411b9c666bcb3dd333c29301de34304e9a298 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Mon, 24 Aug 2026 19:46:12 +0900 Subject: [PATCH 14/14] tests: internal: Add a test case for validating ifallback at the top level Signed-off-by: Hiroshi Hatake --- tests/internal/search_bulk.c | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/internal/search_bulk.c b/tests/internal/search_bulk.c index 971a685ca16..bf71f7550de 100644 --- a/tests/internal/search_bulk.c +++ b/tests/internal/search_bulk.c @@ -123,6 +123,24 @@ static void test_truncated_success_response_is_complete(void) TEST_CHECK(retry == NULL); } +static void test_nested_success_marker_with_top_level_errors_is_invalid(void) +{ + int result; + const char *response; + struct flb_search_bulk_retry *retry; + + response = "{\"metadata\":{\"errors\":false,\"items\":[1]}," + "\"errors\":true,\"items\":[{\"create\":{\"status\":429}}"; + + result = flb_search_bulk_process_response(response, strlen(response), + BULK_PAYLOAD, + strlen(BULK_PAYLOAD), + FLB_SEARCH_BULK_ACK_CREATE_CONFLICTS, + &retry); + TEST_CHECK(result == FLB_SEARCH_BULK_INVALID); + TEST_CHECK(retry == NULL); +} + static void test_item_count_mismatch_is_invalid(void) { int result; @@ -149,6 +167,8 @@ TEST_LIST = { test_update_conflict_is_complete_when_all_conflicts_are_acknowledged}, {"truncated_success_response_is_complete", test_truncated_success_response_is_complete}, + {"nested_success_marker_with_top_level_errors_is_invalid", + test_nested_success_marker_with_top_level_errors_is_invalid}, {"item_count_mismatch_is_invalid", test_item_count_mismatch_is_invalid}, {NULL, NULL} };