es: opensearch: output: task: handle partial succeeded requests

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

include/fluent-bit/flb_search_bulk.h

@@ -0,0 +1,40 @@
+/* -*- 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 <stddef.h>
+
+#define FLB_SEARCH_BULK_COMPLETE  0
+#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;
+    int records;
+};
+
+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);
+
+#endif

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,

plugins/out_es/es.c

@@ -31,6 +31,7 @@
 #include <fluent-bit/flb_log_event_decoder.h>
 #include <fluent-bit/flb_log.h>
 #include <fluent-bit/flb_sds.h>
+#include <fluent-bit/flb_search_bulk.h>
 #include <msgpack.h>
 
 #include <time.h>
@@ -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,36 @@ 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) {
+            /* 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);
                 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 +1504,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 +1523,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);