http: add workers support

include/fluent-bit/flb_downstream.h

@@ -85,8 +85,10 @@ void flb_downstream_resume(struct flb_downstream *stream);
 int flb_downstream_conn_release(struct flb_connection *connection);
 
 int flb_downstream_conn_pending_destroy_list(struct mk_list *list);
+int flb_downstream_conn_pending_destroy(struct flb_downstream *stream);
 
 int flb_downstream_conn_timeouts(struct mk_list *list);
+int flb_downstream_conn_timeouts_stream(struct flb_downstream *stream);
 
 int flb_downstream_is_async(struct flb_downstream *downstream);
 

include/fluent-bit/flb_input.h

@@ -79,6 +79,9 @@
 #define FLB_INPUT_RUNNING     1
 #define FLB_INPUT_PAUSED      0
 
+/* Owner-side ingress queue status */
+#define FLB_INPUT_INGRESS_BUSY -2
+
 struct flb_input_instance;
 struct flb_http_server_config;
 
@@ -476,6 +479,19 @@ struct flb_input_instance {
     struct flb_http_server_config *http_server_config;
     struct mk_list http_server_properties;
 
+    /* Owner-side ingress queue for foreign worker threads */
+    struct mk_list ingress_queue;
+    pthread_mutex_t ingress_queue_lock;
+    pthread_cond_t ingress_queue_space_available;
+    flb_pipefd_t ingress_queue_channels[2];
+    int ingress_queue_enabled;
+    int ingress_queue_collector_id;
+    int ingress_queue_signal_pending;
+    size_t ingress_queue_pending_events;
+    size_t ingress_queue_pending_bytes;
+    size_t ingress_queue_event_limit;
+    size_t ingress_queue_byte_limit;
+
     struct mk_list *oauth2_jwt_config_map;
     struct mk_list oauth2_jwt_properties;
 
@@ -846,6 +862,24 @@ struct mk_event_loop *flb_input_event_loop_get(struct flb_input_instance *ins);
 int flb_input_upstream_set(struct flb_upstream *u, struct flb_input_instance *ins);
 int flb_input_downstream_set(struct flb_downstream *stream,
                              struct flb_input_instance *ins);
+void flb_input_ingress_destroy(struct flb_input_instance *ins);
+int flb_input_ingress_enable(struct flb_input_instance *ins);
+int flb_input_ingress_queue_log(struct flb_input_instance *ins,
+                                const char *tag, size_t tag_len,
+                                const void *buf, size_t buf_size);
+int flb_input_ingress_queue_log_take(struct flb_input_instance *ins,
+                                     const char *tag, size_t tag_len,
+                                     void *buf, size_t buf_size,
+                                     size_t allocation_size);
+int flb_input_ingress_queue_metrics(struct flb_input_instance *ins,
+                                    const char *tag, size_t tag_len,
+                                    struct cmt *cmt);
+int flb_input_ingress_queue_traces(struct flb_input_instance *ins,
+                                   const char *tag, size_t tag_len,
+                                   struct ctrace *ctr);
+int flb_input_ingress_queue_profiles(struct flb_input_instance *ins,
+                                     const char *tag, size_t tag_len,
+                                     struct cprof *profile);
 
 
 /* processors */

include/fluent-bit/flb_input_chunk.h

@@ -124,6 +124,11 @@ int flb_input_chunk_append_raw(struct flb_input_instance *in,
                                size_t records,
                                const char *tag, size_t tag_len,
                                const void *buf, size_t buf_size);
+int flb_input_chunk_ring_buffer_enqueue(struct flb_input_instance *in,
+                                        int event_type,
+                                        size_t records,
+                                        const char *tag, size_t tag_len,
+                                        const void *buf, size_t buf_size);
 
 const void *flb_input_chunk_flush(struct flb_input_chunk *ic, size_t *size);
 int flb_input_chunk_release_lock(struct flb_input_chunk *ic);

include/fluent-bit/flb_ring_buffer.h

@@ -21,8 +21,10 @@
 #define FLB_RING_BUFFER_H
 
 #include <fluent-bit/flb_pipe.h>
+#include <fluent-bit/flb_pthread.h>
 
 struct flb_ring_buffer {
+    pthread_mutex_t lock;            /* protects backend context and signaling state */
     void *ctx;                        /* pointer to backend context */
     void *event_loop;                 /* event loop where this ring buffer emits flush request signals */
     int flush_pending;                /* flag meant to prevent flush request signal flood */
@@ -40,5 +42,6 @@ int flb_ring_buffer_add_event_loop(struct flb_ring_buffer *rb, void *evl, uint8_
 
 int flb_ring_buffer_write(struct flb_ring_buffer *rb, void *ptr, size_t size);
 int flb_ring_buffer_read(struct flb_ring_buffer *rb, void *ptr, size_t size);
+void flb_ring_buffer_mark_flushed(struct flb_ring_buffer *rb);
 
 #endif

include/fluent-bit/http_server/flb_http_server.h

@@ -70,6 +70,7 @@ struct flb_http_server_config {
     size_t buffer_max_size;
     size_t buffer_chunk_size;
     size_t max_connections;
+    int    workers;
 };
 
 struct flb_http_server_options {
@@ -87,6 +88,7 @@ struct flb_http_server_options {
     struct flb_config                   *system_context;
 
     size_t                               buffer_max_size;
+    size_t                               buffer_chunk_size;
     size_t                               max_connections;
 
     /* Total number of worker listeners to spawn. */
@@ -129,11 +131,13 @@ struct flb_http_server {
     flb_http_server_request_processor_callback request_callback;
     void                               *user_data;
     size_t                              buffer_max_size;
+    size_t                              buffer_chunk_size;
     size_t                              max_connections;
     int                                 workers;
     int                                 worker_id;
     int                                 use_caller_event_loop;
     int                                 reuse_port;
+    int                                 tls_alpn_configured;
     flb_http_server_worker_callback     cb_worker_init;
     flb_http_server_worker_callback     cb_worker_exit;
     struct flb_http_server_runtime     *runtime;
@@ -148,6 +152,8 @@ struct flb_http_server_session {
 
     cfl_sds_t                       incoming_data;
     cfl_sds_t                       outgoing_data;
+    unsigned char                  *read_buffer;
+    size_t                          read_buffer_size;
 
     int                             releasable;
 

plugins/in_elasticsearch/in_elasticsearch.c

@@ -101,6 +101,11 @@ static int in_elasticsearch_bulk_init(struct flb_input_instance *ins,
             (FLB_HTTP_SERVER_FLAG_KEEPALIVE | FLB_HTTP_SERVER_FLAG_AUTO_INFLATE),
             in_elasticsearch_bulk_prot_handle_ng,
             ctx);
+    if (ret == 0) {
+        if (http_server_options.workers > 1) {
+            ret = flb_input_ingress_enable(ins);
+        }
+    }
     if (ret == 0) {
         ret = flb_http_server_init_with_options(&ctx->http_server,
                                                 &http_server_options);
@@ -109,7 +114,7 @@ static int in_elasticsearch_bulk_init(struct flb_input_instance *ins,
             ret = flb_http_server_start(&ctx->http_server);
         }
 
-        if (ret == 0) {
+        if (ret == 0 && ctx->http_server.downstream != NULL) {
             ret = flb_input_downstream_set(ctx->http_server.downstream, ins);
         }
     }

plugins/in_elasticsearch/in_elasticsearch.h

@@ -50,5 +50,24 @@ struct flb_in_elasticsearch {
     struct flb_http_server http_server;
 };
 
+static inline int in_elasticsearch_uses_worker_ingress_queue(
+    struct flb_in_elasticsearch *ctx)
+{
+    return ctx->http_server.workers > 1;
+}
+
+static inline int in_elasticsearch_ingest_logs(struct flb_in_elasticsearch *ctx,
+                                               const char *tag,
+                                               size_t tag_len,
+                                               const void *buf,
+                                               size_t buf_size)
+{
+    if (in_elasticsearch_uses_worker_ingress_queue(ctx)) {
+        return flb_input_ingress_queue_log(ctx->ins, tag, tag_len, buf, buf_size);
+    }
+
+    return flb_input_log_append(ctx->ins, tag, tag_len, buf, buf_size);
+}
+
 
 #endif