feat(connectors): restart connector with new config without runtime restart

core/connectors/runtime/src/api/sink.rs

@@ -30,7 +30,7 @@ use axum::{
     extract::{Path, Query, State},
     http::{HeaderMap, StatusCode, header},
     response::IntoResponse,
-    routing::get,
+    routing::{get, post},
 };
 use serde::Deserialize;
 use std::sync::Arc;
@@ -52,6 +52,7 @@ pub fn router(state: Arc<RuntimeContext>) -> Router {
             "/sinks/{key}/configs/active",
             get(get_sink_active_config).put(update_sink_active_config),
         )
+        .route("/sinks/{key}/restart", post(restart_sink))
         .with_state(state)
 }
 
@@ -246,3 +247,20 @@ async fn delete_sink_config(
         .await?;
     Ok(StatusCode::NO_CONTENT)
 }
+
+async fn restart_sink(
+    State(context): State<Arc<RuntimeContext>>,
+    Path(key): Path<String>,
+) -> Result<StatusCode, ApiError> {
+    context
+        .sinks
+        .restart_connector(
+            &key,
+            context.config_provider.as_ref(),
+            &context.iggy_clients.consumer,
+            &context.metrics,
+            &context,
+        )
+        .await?;
+    Ok(StatusCode::NO_CONTENT)
+}

core/connectors/runtime/src/api/source.rs

@@ -30,7 +30,7 @@ use axum::{
     extract::{Path, Query, State},
     http::{HeaderMap, StatusCode, header},
     response::IntoResponse,
-    routing::get,
+    routing::{get, post},
 };
 use serde::Deserialize;
 use std::sync::Arc;
@@ -55,6 +55,7 @@ pub fn router(state: Arc<RuntimeContext>) -> Router {
             "/sources/{key}/configs/active",
             get(get_source_active_config).put(update_source_active_config),
         )
+        .route("/sources/{key}/restart", post(restart_source))
         .with_state(state)
 }
 
@@ -249,3 +250,21 @@ async fn delete_source_config(
         .await?;
     Ok(StatusCode::NO_CONTENT)
 }
+
+async fn restart_source(
+    State(context): State<Arc<RuntimeContext>>,
+    Path(key): Path<String>,
+) -> Result<StatusCode, ApiError> {
+    context
+        .sources
+        .restart_connector(
+            &key,
+            context.config_provider.as_ref(),
+            &context.iggy_clients.producer,
+            &context.metrics,
+            &context.state_path,
+            &context,
+        )
+        .await?;
+    Ok(StatusCode::NO_CONTENT)
+}

core/connectors/runtime/src/context.rs

@@ -19,6 +19,7 @@
 use crate::configs::connectors::{ConnectorsConfigProvider, SinkConfig, SourceConfig};
 use crate::configs::runtime::ConnectorsRuntimeConfig;
 use crate::metrics::Metrics;
+use crate::stream::IggyClients;
 use crate::{
     SinkConnectorWrapper, SourceConnectorWrapper,
     manager::{
@@ -31,6 +32,7 @@ use iggy_connector_sdk::api::ConnectorError;
 use iggy_connector_sdk::api::ConnectorStatus;
 use std::collections::HashMap;
 use std::sync::Arc;
+use tokio::sync::Mutex;
 use tracing::error;
 
 pub struct RuntimeContext {
@@ -40,15 +42,20 @@ pub struct RuntimeContext {
     pub config_provider: Arc<dyn ConnectorsConfigProvider>,
     pub metrics: Arc<Metrics>,
     pub start_time: IggyTimestamp,
+    pub iggy_clients: Arc<IggyClients>,
+    pub state_path: String,
 }
 
+#[allow(clippy::too_many_arguments)]
 pub fn init(
     config: &ConnectorsRuntimeConfig,
     sinks_config: &HashMap<String, SinkConfig>,
     sources_config: &HashMap<String, SourceConfig>,
     sink_wrappers: &[SinkConnectorWrapper],
     source_wrappers: &[SourceConnectorWrapper],
     config_provider: Box<dyn ConnectorsConfigProvider>,
+    iggy_clients: Arc<IggyClients>,
+    state_path: String,
 ) -> RuntimeContext {
     let metrics = Arc::new(Metrics::init());
     let sinks = SinkManager::new(map_sinks(sinks_config, sink_wrappers));
@@ -64,6 +71,8 @@ pub fn init(
         config_provider: Arc::from(config_provider),
         metrics,
         start_time: IggyTimestamp::now(),
+        iggy_clients,
+        state_path,
     }
 }
 
@@ -103,6 +112,10 @@ fn map_sinks(
                     plugin_config_format: sink_plugin.config_format,
                 },
                 config: sink_config.clone(),
+                shutdown_tx: None,
+                task_handles: vec![],
+                container: None,
+                restart_guard: Arc::new(Mutex::new(())),
             });
         }
     }
@@ -145,6 +158,9 @@ fn map_sources(
                     plugin_config_format: source_plugin.config_format,
                 },
                 config: source_config.clone(),
+                handler_tasks: vec![],
+                container: None,
+                restart_guard: Arc::new(Mutex::new(())),
             });
         }
     }

core/connectors/runtime/src/main.rs

@@ -29,6 +29,7 @@ use figlet_rs::FIGfont;
 use iggy::prelude::{Client, IggyConsumer, IggyProducer};
 use iggy_connector_sdk::{
     StreamDecoder, StreamEncoder,
+    api::ConnectorStatus,
     sink::ConsumeCallback,
     source::{HandleCallback, SendCallback},
     transforms::Transform,
@@ -40,7 +41,7 @@ use std::{
     env,
     sync::{Arc, atomic::AtomicU32},
 };
-use tracing::info;
+use tracing::{error, info};
 
 mod api;
 pub(crate) mod configs;
@@ -69,8 +70,8 @@ static PLUGIN_ID: AtomicU32 = AtomicU32::new(1);
 const ALLOWED_PLUGIN_EXTENSIONS: [&str; 3] = ["so", "dylib", "dll"];
 const DEFAULT_CONFIG_PATH: &str = "core/connectors/runtime/config.toml";
 
-#[derive(WrapperApi)]
-struct SourceApi {
+#[derive(WrapperApi, Debug)]
+pub(crate) struct SourceApi {
     iggy_source_open: extern "C" fn(
         id: u32,
         config_ptr: *const u8,
@@ -84,8 +85,8 @@ struct SourceApi {
     iggy_source_version: extern "C" fn() -> *const std::ffi::c_char,
 }
 
-#[derive(WrapperApi)]
-struct SinkApi {
+#[derive(WrapperApi, Debug)]
+pub(crate) struct SinkApi {
     iggy_sink_open: extern "C" fn(
         id: u32,
         config_ptr: *const u8,
@@ -143,7 +144,7 @@ async fn main() -> Result<(), RuntimeError> {
 
     info!("State will be stored in: {}", config.state.path);
 
-    let iggy_clients = stream::init(config.iggy.clone()).await?;
+    let iggy_clients = Arc::new(stream::init(config.iggy.clone()).await?);
 
     let connectors_config_provider: Box<dyn ConnectorsConfigProvider> =
         create_connectors_config_provider(&config.connectors).await?;
@@ -166,47 +167,31 @@ async fn main() -> Result<(), RuntimeError> {
     let sinks = sink::init(sinks_config.clone(), &iggy_clients.consumer).await?;
 
     let mut sink_wrappers = vec![];
-    let mut sink_with_plugins = HashMap::new();
-    for (key, sink) in sinks {
-        let plugin_ids = sink
-            .plugins
-            .iter()
-            .filter(|plugin| plugin.error.is_none())
-            .map(|plugin| plugin.id)
-            .collect();
+    let mut sink_containers_by_key: HashMap<String, Arc<Container<SinkApi>>> = HashMap::new();
+    for (_path, sink) in sinks {
+        let container = Arc::new(sink.container);
+        let callback = container.iggy_sink_consume;
+        for plugin in &sink.plugins {
+            sink_containers_by_key.insert(plugin.key.clone(), container.clone());
+        }
         sink_wrappers.push(SinkConnectorWrapper {
-            callback: sink.container.iggy_sink_consume,
+            callback,
             plugins: sink.plugins,
         });
-        sink_with_plugins.insert(
-            key,
-            SinkWithPlugins {
-                container: sink.container,
-                plugin_ids,
-            },
-        );
     }
 
     let mut source_wrappers = vec![];
-    let mut source_with_plugins = HashMap::new();
-    for (key, source) in sources {
-        let plugin_ids = source
-            .plugins
-            .iter()
-            .filter(|plugin| plugin.error.is_none())
-            .map(|plugin| plugin.id)
-            .collect();
+    let mut source_containers_by_key: HashMap<String, Arc<Container<SourceApi>>> = HashMap::new();
+    for (_path, source) in sources {
+        let container = Arc::new(source.container);
+        let callback = container.iggy_source_handle;
+        for plugin in &source.plugins {
+            source_containers_by_key.insert(plugin.key.clone(), container.clone());
+        }
         source_wrappers.push(SourceConnectorWrapper {
-            callback: source.container.iggy_source_handle,
+            callback,
             plugins: source.plugins,
         });
-        source_with_plugins.insert(
-            key,
-            SourceWithPlugins {
-                container: source.container,
-                plugin_ids,
-            },
-        );
     }
 
     let context = context::init(
@@ -216,13 +201,47 @@ async fn main() -> Result<(), RuntimeError> {
         &sink_wrappers,
         &source_wrappers,
         connectors_config_provider,
+        iggy_clients.clone(),
+        config.state.path.clone(),
     );
+    for (key, container) in sink_containers_by_key {
+        if let Some(details) = context.sinks.get(&key).await {
+            let mut details = details.lock().await;
+            details.container = Some(container);
+        }
+    }
+    for (key, container) in source_containers_by_key {
+        if let Some(details) = context.sources.get(&key).await {
+            let mut details = details.lock().await;
+            details.container = Some(container);
+        }
+    }
+
     let context = Arc::new(context);
-    api::init(&config.http, context.clone()).await;
 
-    let source_handler_tasks = source::handle(source_wrappers, context.clone());
-    sink::consume(sink_wrappers, context.clone());
+    let source_handles = source::handle(source_wrappers, context.clone());
+    for (key, handler_tasks) in source_handles {
+        if let Some(details) = context.sources.get(&key).await {
+            let mut details = details.lock().await;
+            details.handler_tasks = handler_tasks;
+        }
+    }
+
+    let sink_handles = sink::consume(sink_wrappers, context.clone());
+    for (key, shutdown_tx, task_handles) in sink_handles {
+        if let Some(details) = context.sinks.get(&key).await {
+            let mut details = details.lock().await;
+            details.shutdown_tx = Some(shutdown_tx);
+            details.task_handles = task_handles;
+        }
+        context
+            .sinks
+            .update_status(&key, ConnectorStatus::Running, Some(&context.metrics))
+            .await;
+    }
+
     info!("All sources and sinks spawned.");
+    api::init(&config.http, context.clone()).await;
 
     #[cfg(unix)]
     let (mut ctrl_c, mut sigterm) = {
@@ -243,26 +262,37 @@ async fn main() -> Result<(), RuntimeError> {
         }
     }
 
-    for (key, source) in source_with_plugins {
-        for id in source.plugin_ids {
-            info!("Closing source connector with ID: {id} for plugin: {key}");
-            source.container.iggy_source_close(id);
-            source::cleanup_sender(id);
-            info!("Closed source connector with ID: {id} for plugin: {key}");
+    let source_keys: Vec<String> = context
+        .sources
+        .get_all()
+        .await
+        .into_iter()
+        .map(|s| s.key)
+        .collect();
+    for key in &source_keys {
+        if let Err(err) = context
+            .sources
+            .stop_connector_with_guard(key, &context.metrics)
+            .await
+        {
+            error!("Failed to stop source connector: {key}. {err}");
         }
     }
 
-    // Wait for source handler tasks to drain remaining messages and persist state
-    // before shutting down the Iggy clients they depend on.
-    for handle in source_handler_tasks {
-        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle).await;
-    }
-
-    for (key, sink) in sink_with_plugins {
-        for id in sink.plugin_ids {
-            info!("Closing sink connector with ID: {id} for plugin: {key}");
-            sink.container.iggy_sink_close(id);
-            info!("Closed sink connector with ID: {id} for plugin: {key}");
+    let sink_keys: Vec<String> = context
+        .sinks
+        .get_all()
+        .await
+        .into_iter()
+        .map(|s| s.key)
+        .collect();
+    for key in &sink_keys {
+        if let Err(err) = context
+            .sinks
+            .stop_connector_with_guard(key, &context.metrics)
+            .await
+        {
+            error!("Failed to stop sink connector: {key}. {err}");
         }
     }
 
@@ -400,11 +430,6 @@ struct SinkConnectorWrapper {
     plugins: Vec<SinkConnectorPlugin>,
 }
 
-struct SinkWithPlugins {
-    container: Container<SinkApi>,
-    plugin_ids: Vec<u32>,
-}
-
 struct SourceConnector {
     container: Container<SourceApi>,
     plugins: Vec<SourceConnectorPlugin>,
@@ -429,11 +454,6 @@ struct SourceConnectorProducer {
     producer: IggyProducer,
 }
 
-struct SourceWithPlugins {
-    container: Container<SourceApi>,
-    plugin_ids: Vec<u32>,
-}
-
 struct SourceConnectorWrapper {
     callback: HandleCallback,
     plugins: Vec<SourceConnectorPlugin>,