input: add pause/resume callbacks

.github/workflows/input-http-pause-tests.yaml

@@ -0,0 +1,113 @@
+name: Input HTTP pause and lifecycle tests
+
+on:
+  pull_request:
+    branches:
+      - master
+    paths:
+      - '.github/workflows/input-http-pause-tests.yaml'
+      - 'include/fluent-bit/flb_connection.h'
+      - 'include/fluent-bit/flb_downstream.h'
+      - 'include/fluent-bit/http_server/**'
+      - 'plugins/in_elasticsearch/**'
+      - 'plugins/in_http/**'
+      - 'plugins/in_opentelemetry/**'
+      - 'plugins/in_prometheus_remote_write/**'
+      - 'plugins/in_splunk/**'
+      - 'src/flb_downstream.c'
+      - 'src/flb_network.c'
+      - 'src/http_server/**'
+      - 'tests/integration/scenarios/in_elasticsearch/**'
+      - 'tests/integration/scenarios/in_http/**'
+      - 'tests/integration/scenarios/in_http_max_connections/**'
+      - 'tests/integration/scenarios/in_opentelemetry/**'
+      - 'tests/integration/scenarios/in_prometheus_remote_write/**'
+      - 'tests/integration/scenarios/in_splunk/**'
+      - 'tests/integration/src/**'
+      - 'tests/internal/http_server.c'
+  push:
+    branches:
+      - master
+    paths:
+      - 'include/fluent-bit/flb_connection.h'
+      - 'include/fluent-bit/flb_downstream.h'
+      - 'include/fluent-bit/http_server/**'
+      - 'plugins/in_elasticsearch/**'
+      - 'plugins/in_http/**'
+      - 'plugins/in_opentelemetry/**'
+      - 'plugins/in_prometheus_remote_write/**'
+      - 'plugins/in_splunk/**'
+      - 'src/flb_downstream.c'
+      - 'src/flb_network.c'
+      - 'src/http_server/**'
+      - 'tests/integration/scenarios/in_elasticsearch/**'
+      - 'tests/integration/scenarios/in_http/**'
+      - 'tests/integration/scenarios/in_http_max_connections/**'
+      - 'tests/integration/scenarios/in_opentelemetry/**'
+      - 'tests/integration/scenarios/in_prometheus_remote_write/**'
+      - 'tests/integration/scenarios/in_splunk/**'
+      - 'tests/integration/src/**'
+      - 'tests/internal/http_server.c'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  focused-linux:
+    name: Focused Linux integration and Valgrind
+    runs-on: ubuntu-22.04
+    timeout-minutes: 30
+
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+
+      - name: Install build and test dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y \
+            bison \
+            build-essential \
+            cmake \
+            flex \
+            libbpf-dev \
+            libssl-dev \
+            libsystemd-dev \
+            libyaml-dev \
+            python3-venv \
+            valgrind
+
+      - name: Configure and build
+        run: |
+          tests/integration/setup-venv.sh
+          cmake -S . -B build \
+            -DFLB_TESTS_RUNTIME=On \
+            -DFLB_TESTS_INTERNAL=On
+          cmake --build build -j8
+
+      - name: Run focused integration tests
+        run: |
+          tests/integration/.venv/bin/python -m pytest -q \
+            tests/integration/scenarios/in_http/tests/test_in_http_001.py \
+            tests/integration/scenarios/in_http_max_connections/tests/test_in_http_max_connections_001.py \
+            tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py \
+            tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py \
+            tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py \
+            tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py \
+            -k 'pause_resume or shutdown_while or async_tls_accept_timeout or max_connections or idle_timeout'
+
+      - name: Run focused integration tests with strict Valgrind
+        env:
+          VALGRIND: 1
+          VALGRIND_STRICT: 1
+        run: |
+          tests/integration/.venv/bin/python -m pytest -q \
+            tests/integration/scenarios/in_http/tests/test_in_http_001.py \
+            tests/integration/scenarios/in_http_max_connections/tests/test_in_http_max_connections_001.py \
+            tests/integration/scenarios/in_elasticsearch/tests/test_in_elasticsearch_001.py \
+            tests/integration/scenarios/in_opentelemetry/tests/test_in_opentelemetry_001.py \
+            tests/integration/scenarios/in_prometheus_remote_write/tests/test_in_prometheus_remote_write_001.py \
+            tests/integration/scenarios/in_splunk/tests/test_in_splunk_001.py \
+            -k 'pause_resume or shutdown_while or async_tls_accept_timeout or max_connections or idle_timeout'

include/fluent-bit/flb_connection.h

@@ -62,6 +62,9 @@ struct flb_connection;
 
 typedef void (*flb_connection_drop_notification_callback)(
                  struct flb_connection *connection);
+typedef int (*flb_connection_accept_callback)(
+                 struct flb_connection *connection,
+                 void *data);
 typedef int (*flb_connection_event_callback)(void *data);
 
 /* Base network connection */
@@ -79,6 +82,9 @@ struct flb_connection {
      * teardown immediately after the callback returns.
      */
     flb_connection_drop_notification_callback drop_notification_callback;
+    flb_connection_accept_callback accept_callback;
+    void *accept_callback_data;
+    int event_registration_mask;
 
     /* Socket */
     flb_sockfd_t fd;
@@ -156,6 +162,9 @@ struct flb_connection {
     /* Downstream-owned event callback coroutine */
     struct flb_coro *event_coroutine;
     flb_connection_event_callback event_callback;
+    flb_connection_event_callback event_parent_callback;
+    void *event_parent_callback_data;
+    int event_parent_callback_result;
     int event_wakeup_pending;
     int event_release_pending;
 

include/fluent-bit/flb_downstream.h

@@ -86,14 +86,30 @@ void flb_downstream_pause(struct flb_downstream *stream);
 void flb_downstream_resume(struct flb_downstream *stream);
 
 int flb_downstream_conn_release(struct flb_connection *connection);
+int flb_downstream_conn_release_all(struct flb_downstream *stream);
 
 /*
  * The callback and any ingestion it invokes run on config->coro_stack_size.
  * Callers must size that stack for their complete callback path.
  */
+int flb_downstream_conn_event_accept(
+        struct flb_downstream *stream,
+        flb_connection_accept_callback accept_callback,
+        void *accept_callback_data,
+        flb_connection_event_callback event_callback,
+        int mask);
 int flb_downstream_conn_event_register(struct flb_connection *connection,
                                        int (*callback)(void *data),
                                        int mask);
+
+/*
+ * Suspend an event coroutine while callback runs on its parent stack. This is
+ * required for code which depends on native thread stack bounds, such as WAMR.
+ */
+int flb_downstream_conn_event_call_parent(
+        struct flb_connection *connection,
+        flb_connection_event_callback callback,
+        void *callback_data);
 void flb_downstream_conn_event_resume(struct flb_connection *connection);
 
 int flb_downstream_conn_pending_destroy_list(struct mk_list *list);

include/fluent-bit/flb_input.h

@@ -203,6 +203,8 @@ struct flb_input_plugin {
      */
     void (*cb_pause) (void *, struct flb_config *);
     void (*cb_resume) (void *, struct flb_config *);
+    int (*cb_pause_checked) (void *, struct flb_config *);
+    int (*cb_resume_checked) (void *, struct flb_config *);
 
     /*
      * Optional callback that can be used from a parent caller to ingest
@@ -903,6 +905,8 @@ void *flb_input_flush(struct flb_input_instance *ins, size_t *size);
 int flb_input_test_pause_resume(struct flb_input_instance *ins, int sleep_seconds);
 int flb_input_pause(struct flb_input_instance *ins);
 int flb_input_pause_all(struct flb_config *config);
+int flb_input_plugin_pause(struct flb_input_instance *ins);
+int flb_input_plugin_resume(struct flb_input_instance *ins);
 int flb_input_resume(struct flb_input_instance *ins);
 #ifdef FLB_HAVE_METRICS
 void flb_input_rate_update(struct flb_input_instance *ins,

include/fluent-bit/http_server/flb_http_server.h

@@ -169,6 +169,7 @@ struct flb_http_server_session {
 
     int                             releasable;
     int                             drop_pending;
+    int                             destroying;
     int                             connection_slot_reserved;
 
     struct flb_connection          *connection;
@@ -219,6 +220,10 @@ int flb_http_server_init_with_options(struct flb_http_server *session,
 
 int flb_http_server_start(struct flb_http_server *session);
 
+int flb_http_server_pause(struct flb_http_server *session);
+
+int flb_http_server_resume(struct flb_http_server *session);
+
 int flb_http_server_stop(struct flb_http_server *session);
 
 int flb_http_server_destroy(struct flb_http_server *session);

plugins/in_elasticsearch/in_elasticsearch.c

@@ -153,6 +153,36 @@ static int in_elasticsearch_bulk_exit(void *data, struct flb_config *config)
     return 0;
 }
 
+static int in_elasticsearch_bulk_pause(void *data, struct flb_config *config)
+{
+    struct flb_in_elasticsearch *ctx;
+
+    (void) config;
+
+    ctx = data;
+    if (flb_http_server_pause(&ctx->http_server) != 0) {
+        flb_plg_error(ctx->ins, "could not pause HTTP server");
+        return -1;
+    }
+
+    return 0;
+}
+
+static int in_elasticsearch_bulk_resume(void *data, struct flb_config *config)
+{
+    struct flb_in_elasticsearch *ctx;
+
+    (void) config;
+
+    ctx = data;
+    if (flb_http_server_resume(&ctx->http_server) != 0) {
+        flb_plg_error(ctx->ins, "could not resume HTTP server");
+        return -1;
+    }
+
+    return 0;
+}
+
 /* Configuration properties map */
 static struct flb_config_map config_map[] = {
     {
@@ -191,8 +221,8 @@ struct flb_input_plugin in_elasticsearch_plugin = {
     .cb_pre_run   = NULL,
     .cb_collect   = NULL,
     .cb_flush_buf = NULL,
-    .cb_pause     = NULL,
-    .cb_resume    = NULL,
+    .cb_pause_checked = in_elasticsearch_bulk_pause,
+    .cb_resume_checked = in_elasticsearch_bulk_resume,
     .cb_exit      = in_elasticsearch_bulk_exit,
     .config_map   = config_map,
     .flags        = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER | FLB_IO_OPT_TLS

plugins/in_http/http.c

@@ -138,6 +138,38 @@ static int in_http_exit(void *data, struct flb_config *config)
     return 0;
 }
 
+static int in_http_pause(void *data, struct flb_config *config)
+{
+    struct flb_http *ctx;
+
+    (void) config;
+
+    ctx = data;
+
+    if (flb_http_server_pause(&ctx->http_server) != 0) {
+        flb_plg_error(ctx->ins, "could not pause HTTP server");
+        return -1;
+    }
+
+    return 0;
+}
+
+static int in_http_resume(void *data, struct flb_config *config)
+{
+    struct flb_http *ctx;
+
+    (void) config;
+
+    ctx = data;
+
+    if (flb_http_server_resume(&ctx->http_server) != 0) {
+        flb_plg_error(ctx->ins, "could not resume HTTP server");
+        return -1;
+    }
+
+    return 0;
+}
+
 /* Configuration properties map */
 static struct flb_config_map config_map[] = {
     {
@@ -188,8 +220,8 @@ struct flb_input_plugin in_http_plugin = {
     .cb_pre_run   = NULL,
     .cb_collect   = NULL,
     .cb_flush_buf = NULL,
-    .cb_pause     = NULL,
-    .cb_resume    = NULL,
+    .cb_pause_checked = in_http_pause,
+    .cb_resume_checked = in_http_resume,
     .cb_exit      = in_http_exit,
     .config_map   = config_map,
     .flags        = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER | FLB_IO_OPT_TLS

plugins/in_opentelemetry/opentelemetry.c

@@ -137,6 +137,38 @@ static int in_opentelemetry_exit(void *data, struct flb_config *config)
     return 0;
 }
 
+static int in_opentelemetry_pause(void *data, struct flb_config *config)
+{
+    struct flb_opentelemetry *ctx;
+
+    (void) config;
+
+    ctx = data;
+
+    if (flb_http_server_pause(&ctx->http_server) != 0) {
+        flb_plg_error(ctx->ins, "could not pause HTTP server");
+        return -1;
+    }
+
+    return 0;
+}
+
+static int in_opentelemetry_resume(void *data, struct flb_config *config)
+{
+    struct flb_opentelemetry *ctx;
+
+    (void) config;
+
+    ctx = data;
+
+    if (flb_http_server_resume(&ctx->http_server) != 0) {
+        flb_plg_error(ctx->ins, "could not resume HTTP server");
+        return -1;
+    }
+
+    return 0;
+}
+
 /* Configuration properties map */
 static struct flb_config_map config_map[] = {
     {
@@ -197,8 +229,8 @@ struct flb_input_plugin in_opentelemetry_plugin = {
     .cb_pre_run   = NULL,
     .cb_collect   = NULL,
     .cb_flush_buf = NULL,
-    .cb_pause     = NULL,
-    .cb_resume    = NULL,
+    .cb_pause_checked = in_opentelemetry_pause,
+    .cb_resume_checked = in_opentelemetry_resume,
     .cb_exit      = in_opentelemetry_exit,
     .config_map   = config_map,
     .flags        = FLB_INPUT_NET_SERVER | FLB_INPUT_HTTP_SERVER | FLB_IO_OPT_TLS