http_server: add settings for ingress queue

AGENTS.md

@@ -71,6 +71,10 @@ Keep changes scoped: plugin logic in its plugin directory, shared behavior in `s
 - Run broader test coverage when changing shared lifecycle, routing, storage, or accounting code.
 - Validate both success and failure paths (invalid payloads, boundary sizes, null/missing fields).
 - You can also run specific binaries from `build/bin` (e.g., `./bin/flb-it-opentelemetry`).
+- When changing code covered by `tests/integration`, agents must verify the
+  affected scenarios are valgrind-clean. Run the focused integration tests with
+  `tests/integration/run_tests.py --valgrind --valgrind-strict ...` and do not
+  stop at functional pass/fail if memory errors or leaks remain.
 - Keep generated integration artifacts out of git. Do not commit
   `.venv/`, `.pytest_cache/`, `results/`, or `__pycache__/` under
   `tests/integration`.

include/fluent-bit/flb_input.h

@@ -436,6 +436,9 @@ struct flb_input_instance {
     struct cmt_counter *cmt_ring_buffer_writes;
     struct cmt_counter *cmt_ring_buffer_retries;
     struct cmt_counter *cmt_ring_buffer_retry_failures;
+    struct cmt_counter *cmt_ingress_queue_busy;
+    struct cmt_gauge   *cmt_ingress_queue_pending_events;
+    struct cmt_gauge   *cmt_ingress_queue_pending_bytes;
 
     /*
      * Indexes for generated chunks: simple hash tables that keeps the latest

include/fluent-bit/http_server/flb_http_server.h

@@ -65,12 +65,17 @@ typedef int (*flb_http_server_worker_callback)(struct flb_http_server *server,
 
 struct flb_input_instance;
 
+#define FLB_HTTP_SERVER_INGRESS_QUEUE_EVENT_LIMIT 8192
+#define FLB_HTTP_SERVER_INGRESS_QUEUE_BYTE_LIMIT  (256 * 1024 * 1024)
+
 struct flb_http_server_config {
     int    http2;
     size_t buffer_max_size;
     size_t buffer_chunk_size;
     size_t max_connections;
     int    workers;
+    size_t ingress_queue_event_limit;
+    size_t ingress_queue_byte_limit;
 };
 
 struct flb_http_server_options {

src/flb_input.c

@@ -466,8 +466,8 @@ struct flb_input_instance *flb_input_new(struct flb_config *config,
         instance->ingress_queue_signal_pending = FLB_FALSE;
         instance->ingress_queue_pending_events = 0;
         instance->ingress_queue_pending_bytes = 0;
-        instance->ingress_queue_event_limit = 8192;
-        instance->ingress_queue_byte_limit = 256 * 1024 * 1024;
+        instance->ingress_queue_event_limit = FLB_HTTP_SERVER_INGRESS_QUEUE_EVENT_LIMIT;
+        instance->ingress_queue_byte_limit = FLB_HTTP_SERVER_INGRESS_QUEUE_BYTE_LIMIT;
 
         /* Plugin use networking */
         if (plugin->flags & (FLB_INPUT_NET | FLB_INPUT_NET_SERVER)) {
@@ -975,6 +975,8 @@ void flb_input_instance_destroy(struct flb_input_instance *ins)
     struct mk_list *head;
     struct flb_input_collector *collector;
 
+    flb_input_ingress_destroy(ins);
+
     if (ins->alias) {
         flb_sds_destroy(ins->alias);
     }
@@ -1134,8 +1136,6 @@ void flb_input_instance_destroy(struct flb_input_instance *ins)
         mk_event_closesocket(ins->ingress_queue_channels[1]);
     }
 
-    flb_input_ingress_destroy(ins);
-
     /* Collectors */
     mk_list_foreach_safe(head, tmp, &ins->collectors) {
         collector = mk_list_entry(head, struct flb_input_collector, _head);
@@ -1552,6 +1552,35 @@ int flb_input_instance_init(struct flb_input_instance *ins,
                             1, (char *[]) {"name"});
     cmt_counter_set(ins->cmt_ring_buffer_retry_failures, ts, 0, 1, (char *[]) {name});
 
+    if ((p->flags & FLB_INPUT_HTTP_SERVER) != 0) {
+        /* fluentbit_input_http_server_ingress_queue_busy_total */
+        ins->cmt_ingress_queue_busy = \
+            cmt_counter_create(ins->cmt,
+                               "fluentbit", "input",
+                               "http_server_ingress_queue_busy_total",
+                               "Number of deferred HTTP server ingress queue busy events.",
+                               1, (char *[]) {"name"});
+        cmt_counter_set(ins->cmt_ingress_queue_busy, ts, 0, 1, (char *[]) {name});
+
+        /* fluentbit_input_http_server_ingress_queue_pending_events */
+        ins->cmt_ingress_queue_pending_events = \
+            cmt_gauge_create(ins->cmt,
+                             "fluentbit", "input",
+                             "http_server_ingress_queue_pending_events",
+                             "Current number of deferred HTTP server ingress queue events.",
+                             1, (char *[]) {"name"});
+        cmt_gauge_set(ins->cmt_ingress_queue_pending_events, ts, 0, 1, (char *[]) {name});
+
+        /* fluentbit_input_http_server_ingress_queue_pending_bytes */
+        ins->cmt_ingress_queue_pending_bytes = \
+            cmt_gauge_create(ins->cmt,
+                             "fluentbit", "input",
+                             "http_server_ingress_queue_pending_bytes",
+                             "Current number of deferred HTTP server ingress queue bytes.",
+                             1, (char *[]) {"name"});
+        cmt_gauge_set(ins->cmt_ingress_queue_pending_bytes, ts, 0, 1, (char *[]) {name});
+    }
+
     /* OLD Metrics */
     ins->metrics = flb_metrics_create(name);
     if (ins->metrics) {

src/flb_input_ingest.c

@@ -31,6 +31,7 @@
 #include <fluent-bit/flb_pipe.h>
 #include <fluent-bit/flb_ring_buffer.h>
 #include <fluent-bit/flb_sds.h>
+#include <fluent-bit/http_server/flb_http_server.h>
 
 #include <cprofiles/cprof_encode_msgpack.h>
 
@@ -157,6 +158,33 @@ static void flb_input_ingress_signal(struct flb_input_instance *ins)
     }
 }
 
+static void flb_input_ingress_update_metrics(struct flb_input_instance *ins)
+{
+    uint64_t ts;
+    char *name;
+
+    if (ins == NULL || ins->cmt == NULL) {
+        return;
+    }
+
+    ts = cfl_time_now();
+    name = (char *) flb_input_name(ins);
+
+    if (ins->cmt_ingress_queue_pending_events != NULL) {
+        cmt_gauge_set(ins->cmt_ingress_queue_pending_events,
+                      ts,
+                      ins->ingress_queue_pending_events,
+                      1, (char *[]) {name});
+    }
+
+    if (ins->cmt_ingress_queue_pending_bytes != NULL) {
+        cmt_gauge_set(ins->cmt_ingress_queue_pending_bytes,
+                      ts,
+                      ins->ingress_queue_pending_bytes,
+                      1, (char *[]) {name});
+    }
+}
+
 static int flb_input_ingress_enqueue(struct flb_input_instance *ins,
                                      struct flb_input_ingress_event *event)
 {
@@ -165,6 +193,8 @@ static int flb_input_ingress_enqueue(struct flb_input_instance *ins,
     int should_signal;
     int wait_result;
     struct timespec deadline;
+    struct cmt_counter *cmt_ingress_queue_busy;
+    flb_sds_t input_name;
 
     if (ins == NULL || event == NULL || ins->ingress_queue_enabled != FLB_TRUE) {
         flb_input_ingress_event_destroy(event);
@@ -210,8 +240,23 @@ static int flb_input_ingress_enqueue(struct flb_input_instance *ins,
                                              &ins->ingress_queue_lock,
                                              &deadline);
         if (wait_result == ETIMEDOUT) {
+            cmt_ingress_queue_busy = ins->cmt_ingress_queue_busy;
+            input_name = NULL;
+
+            if (cmt_ingress_queue_busy != NULL) {
+                input_name = flb_sds_create(flb_input_name(ins));
+            }
+
             pthread_mutex_unlock(&ins->ingress_queue_lock);
 
+            if (cmt_ingress_queue_busy != NULL && input_name != NULL) {
+                cmt_counter_add(cmt_ingress_queue_busy,
+                                cfl_time_now(),
+                                1,
+                                1, (char *[]) {input_name});
+                flb_sds_destroy(input_name);
+            }
+
             flb_input_ingress_event_destroy(event);
 
             return FLB_INPUT_INGRESS_BUSY;
@@ -229,6 +274,7 @@ static int flb_input_ingress_enqueue(struct flb_input_instance *ins,
     mk_list_add(&event->_head, &ins->ingress_queue);
     ins->ingress_queue_pending_events++;
     ins->ingress_queue_pending_bytes += event_size;
+    flb_input_ingress_update_metrics(ins);
 
     if (ins->ingress_queue_signal_pending == FLB_FALSE) {
         ins->ingress_queue_signal_pending = FLB_TRUE;
@@ -331,6 +377,7 @@ static int flb_input_ingress_collector(struct flb_input_instance *ins,
         ins->ingress_queue_pending_events = 0;
         ins->ingress_queue_pending_bytes = 0;
         ins->ingress_queue_signal_pending = FLB_FALSE;
+        flb_input_ingress_update_metrics(ins);
 
         if (queue_was_full == FLB_TRUE) {
             pthread_cond_broadcast(&ins->ingress_queue_space_available);
@@ -409,8 +456,22 @@ int flb_input_ingress_enable(struct flb_input_instance *ins)
         return 0;
     }
 
+    if (ins->http_server_config != NULL) {
+        ins->ingress_queue_event_limit =
+            ins->http_server_config->ingress_queue_event_limit;
+
+        if (ins->http_server_config->ingress_queue_byte_limit > 0) {
+            ins->ingress_queue_byte_limit =
+                ins->http_server_config->ingress_queue_byte_limit;
+        }
+        else {
+            ins->ingress_queue_byte_limit = ins->mem_buf_limit;
+        }
+    }
+
     if (ins->mem_buf_limit > 0 &&
-        ins->mem_buf_limit < ins->ingress_queue_byte_limit) {
+        (ins->ingress_queue_byte_limit == 0 ||
+         ins->mem_buf_limit < ins->ingress_queue_byte_limit)) {
         ins->ingress_queue_byte_limit = ins->mem_buf_limit;
     }
 
@@ -459,6 +520,7 @@ void flb_input_ingress_destroy(struct flb_input_instance *ins)
     }
     ins->ingress_queue_pending_events = 0;
     ins->ingress_queue_pending_bytes = 0;
+    flb_input_ingress_update_metrics(ins);
     pthread_mutex_unlock(&ins->ingress_queue_lock);
 }
 

src/http_server/flb_http_server.c

@@ -749,6 +749,8 @@ void flb_http_server_config_init(struct flb_http_server_config *config)
     config->buffer_chunk_size = HTTP_SERVER_INITIAL_BUFFER_SIZE;
     config->max_connections = 0;
     config->workers = 1;
+    config->ingress_queue_event_limit = FLB_HTTP_SERVER_INGRESS_QUEUE_EVENT_LIMIT;
+    config->ingress_queue_byte_limit = FLB_HTTP_SERVER_INGRESS_QUEUE_BYTE_LIMIT;
 }
 
 int flb_http_server_options_init_from_input(struct flb_http_server_options *options,

src/http_server/flb_http_server_config_map.c

@@ -47,6 +47,16 @@ struct flb_config_map flb_http_server_config_map[] = {
      0, FLB_TRUE, offsetof(struct flb_http_server_config, workers),
      "Set the number of HTTP listener workers"
     },
+    {
+     FLB_CONFIG_MAP_SIZE, "http_server.ingress_queue_event_limit", "8192",
+     0, FLB_TRUE, offsetof(struct flb_http_server_config, ingress_queue_event_limit),
+     "Set the maximum number of deferred ingress queue events. Applies only when http_server.workers > 1."
+    },
+    {
+     FLB_CONFIG_MAP_SIZE, "http_server.ingress_queue_byte_limit", "256M",
+     0, FLB_TRUE, offsetof(struct flb_http_server_config, ingress_queue_byte_limit),
+     "Set the maximum number of deferred ingress queue bytes. Applies only when http_server.workers > 1."
+    },
     {
      FLB_CONFIG_MAP_BOOL, "http2", "true",
      0, FLB_TRUE, offsetof(struct flb_http_server_config, http2),

tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue.yaml

@@ -0,0 +1,24 @@
+service:
+  flush: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: opentelemetry
+      port: ${FLUENT_BIT_TEST_LISTENER_PORT}
+      http2: off
+      tls: off
+      http_server.workers: 4
+      http_server.ingress_queue_event_limit: 0
+      http_server.ingress_queue_byte_limit: 1
+
+  outputs:
+    - name: stdout
+      match: '*'
+
+    - name: opentelemetry
+      match: '*'
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}

tests/integration/scenarios/in_opentelemetry/config/otlp_http1_cleartext_workers_tiny_ingress_queue_byte_limit.yaml

@@ -0,0 +1,24 @@
+service:
+  flush: 1
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: opentelemetry
+      port: ${FLUENT_BIT_TEST_LISTENER_PORT}
+      http2: off
+      tls: off
+      http_server.workers: 4
+      http_server.ingress_queue_event_limit: 1024
+      http_server.ingress_queue_byte_limit: 2000
+
+  outputs:
+    - name: stdout
+      match: '*'
+
+    - name: opentelemetry
+      match: '*'
+      host: 127.0.0.1
+      port: ${TEST_SUITE_HTTP_PORT}