in_opentelemetry: implement pause and resume callbacks

Previously, the in_opentelemetry input plugin did not respond to engine pause or resume
signals—meaning it continued to accept and handle connections even when the engine had
paused the plugin (e.g: it should stop ingesting data).

This patch implements the missing pause and resume callbacks. When the plugin is paused, all
active client connections are forcefully closed, and new incoming connections are rejected. The
plugin tracks its paused state to ensure correct behavior until it is resumed.

Signed-off-by: Eduardo Silva <eduardo@chronosphere.io>

Author: edsiper

plugins/in_opentelemetry/opentelemetry.c

@@ -50,6 +50,13 @@ static int in_opentelemetry_collect(struct flb_input_instance *ins,
         return -1;
     }
 
+    if (ctx->is_paused) {
+        flb_plg_trace(ctx->ins, "TCP connection will be closed FD=%i",
+                      connection->fd);
+        flb_downstream_conn_release(connection);
+        return -1;
+    }
+
     flb_plg_trace(ctx->ins, "new TCP connection arrived FD=%i", connection->fd);
 
     conn = opentelemetry_conn_add(connection, ctx);
@@ -76,6 +83,7 @@ static int in_opentelemetry_init(struct flb_input_instance *ins,
         return -1;
     }
     ctx->collector_id = -1;
+    ctx->is_paused = FLB_FALSE;
 
     /* Populate context with config map defaults and incoming properties */
     ret = flb_input_config_map_set(ins, (void *) ctx);
@@ -195,6 +203,27 @@ static int in_opentelemetry_exit(void *data, struct flb_config *config)
     return 0;
 }
 
+static void in_opentelemetry_pause(void *data, struct flb_config *config)
+{
+    struct flb_opentelemetry *ctx = data;
+
+    if (config->is_running == FLB_TRUE) {
+        flb_input_collector_pause(ctx->collector_id, ctx->ins);
+        opentelemetry_conn_release_all(ctx);
+        ctx->is_paused = FLB_TRUE;
+    }
+}
+
+static void in_opentelemetry_resume(void *data, struct flb_config *config)
+{
+    struct flb_opentelemetry *ctx = data;
+
+    if (config->is_running == FLB_TRUE) {
+        flb_input_collector_resume(ctx->collector_id, ctx->ins);
+        ctx->is_paused = FLB_FALSE;
+    }
+}
+
 /* Configuration properties map */
 static struct flb_config_map config_map[] = {
     {
@@ -272,8 +301,8 @@ struct flb_input_plugin in_opentelemetry_plugin = {
     .cb_pre_run   = NULL,
     .cb_collect   = in_opentelemetry_collect,
     .cb_flush_buf = NULL,
-    .cb_pause     = NULL,
-    .cb_resume    = NULL,
+    .cb_pause     = in_opentelemetry_pause,
+    .cb_resume    = in_opentelemetry_resume,
     .cb_exit      = in_opentelemetry_exit,
     .config_map   = config_map,
     .flags        = FLB_INPUT_NET_SERVER | FLB_IO_OPT_TLS

plugins/in_opentelemetry/opentelemetry.h

@@ -57,6 +57,7 @@ struct flb_opentelemetry {
     struct mk_list connections;        /* linked list of connections */
 
     struct mk_server *server;
+    int is_paused;                     /* Plugin is paused */
 };