feat(cli): Handle reload based on referenced file change (vectordotdev#22539)

* feat(watch tls files) Add http sink tls cert/key to config::watcher

* Update changelog.d/22386_extend_watcher_paths.enhancement.md

Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

* feat(watch tls file) Add ComponentConfig to let watcher find components

* feat(watch tls file) Let watcher send ReloadComponent signal

* feat(watch tls file) handle ReloadComponent signal

* feat(watch tls file) Ensure signal is sent once in case of ReloadComponent

* (feat reload components) Handle Vec<ComponentKey> instead of single one

* cargo fmt

* Update changelog.d/22386_extend_watcher_paths.enhancement.md

Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

* Update src/config/mod.rs

Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

* Fix missing enclosed delimiter

* solve clippy issues

* cargo fmt

* Fix testing

* run cargo fmt

* add dedicated watcher test

---------

Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

Author: gllb

changelog.d/22386_extend_watcher_paths.enhancement.md

@@ -0,0 +1,3 @@
+The TLS `crt_file` and `key_file` from `http` sinks are now watched when `--watch_config` is enabled and therefore changes to those files will trigger a config reload without the need to restart Vector.
+
+authors: gllb

lib/vector-common/src/config.rs

@@ -39,6 +39,12 @@ impl ComponentKey {
     }
 }
 
+impl AsRef<ComponentKey> for ComponentKey {
+    fn as_ref(&self) -> &ComponentKey {
+        self
+    }
+}
+
 impl From<String> for ComponentKey {
     fn from(id: String) -> Self {
         Self { id }

src/app.rs

@@ -18,7 +18,7 @@ use crate::extra_context::ExtraContext;
 use crate::{api, internal_events::ApiStarted};
 use crate::{
     cli::{handle_config_errors, LogFormat, Opts, RootOpts, WatchConfigMethod},
-    config::{self, Config, ConfigPath},
+    config::{self, ComponentConfig, ComponentKey, Config, ConfigPath},
     heartbeat,
     internal_events::{VectorConfigLoadError, VectorQuit, VectorStarted, VectorStopped},
     signal::{SignalHandler, SignalPair, SignalRx, SignalTo},
@@ -336,9 +336,32 @@ async fn handle_signal(
     allow_empty_config: bool,
 ) -> Option<SignalTo> {
     match signal {
+        Ok(SignalTo::ReloadComponents(component_keys)) => {
+            let mut topology_controller = topology_controller.lock().await;
+
+            // Reload paths
+            if let Some(paths) = config::process_paths(config_paths) {
+                topology_controller.config_paths = paths;
+            }
+
+            // Reload config
+            let new_config = config::load_from_paths_with_provider_and_secrets(
+                &topology_controller.config_paths,
+                signal_handler,
+                allow_empty_config,
+            )
+            .await;
+
+            reload_config_from_result(
+                topology_controller,
+                new_config,
+                Some(component_keys.iter().map(AsRef::as_ref).collect()),
+            )
+            .await
+        }
         Ok(SignalTo::ReloadFromConfigBuilder(config_builder)) => {
             let topology_controller = topology_controller.lock().await;
-            reload_config_from_result(topology_controller, config_builder.build()).await
+            reload_config_from_result(topology_controller, config_builder.build(), None).await
         }
         Ok(SignalTo::ReloadFromDisk) => {
             let mut topology_controller = topology_controller.lock().await;
@@ -356,7 +379,7 @@ async fn handle_signal(
             )
             .await;
 
-            reload_config_from_result(topology_controller, new_config).await
+            reload_config_from_result(topology_controller, new_config, None).await
         }
         Err(RecvError::Lagged(amt)) => {
             warn!("Overflow, dropped {} signals.", amt);
@@ -370,9 +393,13 @@ async fn handle_signal(
 async fn reload_config_from_result(
     mut topology_controller: MutexGuard<'_, TopologyController>,
     config: Result<Config, Vec<String>>,
+    components_to_reload: Option<Vec<&ComponentKey>>,
 ) -> Option<SignalTo> {
     match config {
-        Ok(new_config) => match topology_controller.reload(new_config).await {
+        Ok(new_config) => match topology_controller
+            .reload(new_config, components_to_reload)
+            .await
+        {
             ReloadOutcome::FatalError(error) => Some(SignalTo::Shutdown(Some(error))),
             _ => None,
         },
@@ -491,23 +518,14 @@ pub async fn load_configs(
 ) -> Result<Config, ExitCode> {
     let config_paths = config::process_paths(config_paths).ok_or(exitcode::CONFIG)?;
 
-    if let Some(watcher_conf) = watcher_conf {
-        // Start listening for config changes immediately.
-        config::watcher::spawn_thread(
-            watcher_conf,
-            signal_handler.clone_tx(),
-            config_paths.iter().map(Into::into),
-            None,
-        )
-        .map_err(|error| {
-            error!(message = "Unable to start config watcher.", %error);
-            exitcode::CONFIG
-        })?;
-    }
+    let watched_paths = config_paths
+        .iter()
+        .map(<&PathBuf>::from)
+        .collect::<Vec<_>>();
 
     info!(
         message = "Loading configs.",
-        paths = ?config_paths.iter().map(<&PathBuf>::from).collect::<Vec<_>>()
+        paths = ?watched_paths
     );
 
     let mut config = config::load_from_paths_with_provider_and_secrets(
@@ -518,6 +536,35 @@ pub async fn load_configs(
     .await
     .map_err(handle_config_errors)?;
 
+    let mut watched_component_paths = Vec::new();
+
+    if let Some(watcher_conf) = watcher_conf {
+        for (name, sink) in config.sinks() {
+            let files = sink.inner.files_to_watch();
+            let component_config =
+                ComponentConfig::new(files.into_iter().cloned().collect(), name.clone());
+            watched_component_paths.push(component_config);
+        }
+
+        info!(
+            message = "Starting watcher.",
+            paths = ?watched_paths
+        );
+
+        // Start listening for config changes.
+        config::watcher::spawn_thread(
+            watcher_conf,
+            signal_handler.clone_tx(),
+            watched_paths,
+            watched_component_paths,
+            None,
+        )
+        .map_err(|error| {
+            error!(message = "Unable to start config watcher.", %error);
+            exitcode::CONFIG
+        })?;
+    }
+
     config::init_log_schema(config.global.log_schema.clone(), true);
     config::init_telemetry(config.global.telemetry.clone(), true);
 

src/config/mod.rs

@@ -76,6 +76,28 @@ pub use vector_lib::{
     id::Inputs,
 };
 
+#[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq)]
+pub struct ComponentConfig {
+    pub config_paths: Vec<PathBuf>,
+    pub component_key: ComponentKey,
+}
+
+impl ComponentConfig {
+    pub const fn new(config_paths: Vec<PathBuf>, component_key: ComponentKey) -> Self {
+        Self {
+            config_paths,
+            component_key,
+        }
+    }
+
+    pub fn contains(&self, config_paths: &[PathBuf]) -> Option<ComponentKey> {
+        if config_paths.iter().any(|p| self.config_paths.contains(p)) {
+            return Some(self.component_key.clone());
+        }
+        None
+    }
+}
+
 #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq)]
 pub enum ConfigPath {
     File(PathBuf, FormatHint),

src/config/sink.rs

@@ -3,6 +3,7 @@ use std::cell::RefCell;
 use async_trait::async_trait;
 use dyn_clone::DynClone;
 use serde::Serialize;
+use std::path::PathBuf;
 use vector_lib::buffers::{BufferConfig, BufferType};
 use vector_lib::configurable::attributes::CustomAttribute;
 use vector_lib::configurable::schema::{SchemaGenerator, SchemaObject};
@@ -220,6 +221,11 @@ pub trait SinkConfig: DynClone + NamedComponent + core::fmt::Debug + Send + Sync
     /// Gets the input configuration for this sink.
     fn input(&self) -> Input;
 
+    /// Gets the files to watch to trigger reload
+    fn files_to_watch(&self) -> Vec<&PathBuf> {
+        Vec::new()
+    }
+
     /// Gets the list of resources, if any, used by this sink.
     ///
     /// Resources represent dependencies -- network ports, file descriptors, and so on -- that

src/config/watcher.rs

@@ -7,6 +7,8 @@ use std::{
     thread,
 };
 
+use crate::config::ComponentConfig;
+
 use notify::{recommended_watcher, EventKind, RecursiveMode};
 
 use crate::Error;
@@ -67,9 +69,18 @@ pub fn spawn_thread<'a>(
     watcher_conf: WatcherConfig,
     signal_tx: crate::signal::SignalTx,
     config_paths: impl IntoIterator<Item = &'a PathBuf> + 'a,
+    component_configs: Vec<ComponentConfig>,
     delay: impl Into<Option<Duration>>,
 ) -> Result<(), Error> {
-    let config_paths: Vec<_> = config_paths.into_iter().cloned().collect();
+    let mut config_paths: Vec<_> = config_paths.into_iter().cloned().collect();
+    let mut component_config_paths: Vec<_> = component_configs
+        .clone()
+        .into_iter()
+        .flat_map(|p| p.config_paths.clone())
+        .collect();
+
+    config_paths.append(&mut component_config_paths);
+
     let delay = delay.into().unwrap_or(CONFIG_WATCH_DELAY);
 
     // Create watcher now so not to miss any changes happening between
@@ -92,6 +103,12 @@ pub fn spawn_thread<'a>(
 
                     debug!(message = "Consumed file change events for delay.", delay = ?delay);
 
+                    let component_keys: Vec<_> = component_configs
+                        .clone()
+                        .into_iter()
+                        .flat_map(|p| p.contains(&event.paths))
+                        .collect();
+
                     // We need to read paths to resolve any inode changes that may have happened.
                     // And we need to do it before raising sighup to avoid missing any change.
                     if let Err(error) = watcher.add_paths(&config_paths) {
@@ -102,9 +119,17 @@ pub fn spawn_thread<'a>(
                     debug!(message = "Reloaded paths.");
 
                     info!("Configuration file changed.");
-                    _ = signal_tx.send(crate::signal::SignalTo::ReloadFromDisk).map_err(|error| {
-                        error!(message = "Unable to reload configuration file. Restart Vector to reload it.", cause = %error)
-                    });
+                    if !component_keys.is_empty() {
+                        info!("Component {:?} configuration changed.", component_keys);
+                        _ = signal_tx.send(crate::signal::SignalTo::ReloadComponents(component_keys)).map_err(|error| {
+                            error!(message = "Unable to reload component configuration. Restart Vector to reload it.", cause = %error)
+                        });
+                    } else {
+                        _ = signal_tx.send(crate::signal::SignalTo::ReloadFromDisk)
+                            .map_err(|error| {
+                                error!(message = "Unable to reload configuration file. Restart Vector to reload it.", cause = %error)
+                            });
+                    }
                 } else {
                     debug!(message = "Ignoring event.", event = ?event)
                 }
@@ -158,6 +183,7 @@ fn create_watcher(
 mod tests {
     use super::*;
     use crate::{
+        config::ComponentKey,
         signal::SignalRx,
         test_util::{temp_dir, temp_file, trace_init},
     };
@@ -174,6 +200,75 @@ mod tests {
         )
     }
 
+    async fn test_component_reload(
+        file: &mut File,
+        expected_component: &ComponentKey,
+        timeout: Duration,
+        mut receiver: SignalRx,
+    ) -> bool {
+        file.write_all(&[0]).unwrap();
+        file.sync_all().unwrap();
+
+        matches!(
+            tokio::time::timeout(timeout, receiver.recv()).await,
+            Ok(Ok(crate::signal::SignalTo::ReloadComponents(components))) if components.contains(expected_component)
+        )
+    }
+
+    #[tokio::test]
+    async fn component_update() {
+        trace_init();
+
+        let delay = Duration::from_secs(3);
+        let dir = temp_dir().to_path_buf();
+        let watcher_conf = WatcherConfig::RecommendedWatcher;
+        let component_file_path = vec![dir.join("tls.cert"), dir.join("tls.key")];
+        let http_component = ComponentKey::from("http");
+
+        std::fs::create_dir(&dir).unwrap();
+
+        let mut component_files: Vec<std::fs::File> = component_file_path
+            .iter()
+            .map(|file| File::create(file).unwrap())
+            .collect();
+        let component_config =
+            ComponentConfig::new(component_file_path.clone(), http_component.clone());
+
+        let (signal_tx, signal_rx) = broadcast::channel(128);
+        spawn_thread(
+            watcher_conf,
+            signal_tx,
+            &[dir],
+            vec![component_config],
+            delay,
+        )
+        .unwrap();
+
+        let signal_rx = signal_rx.resubscribe();
+        let signal_rx2 = signal_rx.resubscribe();
+
+        if !test_component_reload(
+            &mut component_files[0],
+            &http_component,
+            delay * 5,
+            signal_rx,
+        )
+        .await
+        {
+            panic!("Test timed out");
+        }
+
+        if !test_component_reload(
+            &mut component_files[1],
+            &http_component,
+            delay * 5,
+            signal_rx2,
+        )
+        .await
+        {
+            panic!("Test timed out");
+        }
+    }
     #[tokio::test]
     async fn file_directory_update() {
         trace_init();
@@ -187,7 +282,7 @@ mod tests {
         let mut file = File::create(&file_path).unwrap();
 
         let (signal_tx, signal_rx) = broadcast::channel(128);
-        spawn_thread(watcher_conf, signal_tx, &[dir], delay).unwrap();
+        spawn_thread(watcher_conf, signal_tx, &[dir], vec![], delay).unwrap();
 
         if !test(&mut file, delay * 5, signal_rx).await {
             panic!("Test timed out");
@@ -204,7 +299,7 @@ mod tests {
         let watcher_conf = WatcherConfig::RecommendedWatcher;
 
         let (signal_tx, signal_rx) = broadcast::channel(128);
-        spawn_thread(watcher_conf, signal_tx, &[file_path], delay).unwrap();
+        spawn_thread(watcher_conf, signal_tx, &[file_path], vec![], delay).unwrap();
 
         if !test(&mut file, delay * 5, signal_rx).await {
             panic!("Test timed out");
@@ -225,7 +320,7 @@ mod tests {
         let watcher_conf = WatcherConfig::RecommendedWatcher;
 
         let (signal_tx, signal_rx) = broadcast::channel(128);
-        spawn_thread(watcher_conf, signal_tx, &[sym_file], delay).unwrap();
+        spawn_thread(watcher_conf, signal_tx, &[sym_file], vec![], delay).unwrap();
 
         if !test(&mut file, delay * 5, signal_rx).await {
             panic!("Test timed out");
@@ -246,7 +341,7 @@ mod tests {
         let mut file = File::create(&file_path).unwrap();
 
         let (signal_tx, signal_rx) = broadcast::channel(128);
-        spawn_thread(watcher_conf, signal_tx, &[sub_dir], delay).unwrap();
+        spawn_thread(watcher_conf, signal_tx, &[sub_dir], vec![], delay).unwrap();
 
         if !test(&mut file, delay * 5, signal_rx).await {
             panic!("Test timed out");

src/signal.rs

@@ -14,6 +14,8 @@ pub type SignalRx = broadcast::Receiver<SignalTo>;
 /// Control messages used by Vector to drive topology and shutdown events.
 #[allow(clippy::large_enum_variant)] // discovered during Rust upgrade to 1.57; just allowing for now since we did previously
 pub enum SignalTo {
+    /// Signal to reload given components.
+    ReloadComponents(Vec<ComponentKey>),
     /// Signal to reload config from a string.
     ReloadFromConfigBuilder(ConfigBuilder),
     /// Signal to reload config from the filesystem.

src/sinks/http/config.rs

@@ -3,6 +3,7 @@
 use http::{header::AUTHORIZATION, HeaderName, HeaderValue, Method, Request, StatusCode};
 use hyper::Body;
 use indexmap::IndexMap;
+use std::path::PathBuf;
 use vector_lib::codecs::{
     encoding::{Framer, Serializer},
     CharacterDelimitedEncoder,
@@ -301,6 +302,19 @@ impl SinkConfig for HttpSinkConfig {
         Input::new(self.encoding.config().1.input_type())
     }
 
+    fn files_to_watch(&self) -> Vec<&PathBuf> {
+        let mut files = Vec::new();
+        if let Some(tls) = &self.tls {
+            if let Some(crt_file) = &tls.crt_file {
+                files.push(crt_file)
+            }
+            if let Some(key_file) = &tls.key_file {
+                files.push(key_file)
+            }
+        };
+        files
+    }
+
     fn acknowledgements(&self) -> &AcknowledgementsConfig {
         &self.acknowledgements
     }

src/topology/controller.rs

@@ -59,7 +59,11 @@ pub enum ReloadOutcome {
 }
 
 impl TopologyController {
-    pub async fn reload(&mut self, mut new_config: config::Config) -> ReloadOutcome {
+    pub async fn reload(
+        &mut self,
+        mut new_config: config::Config,
+        components_to_reload: Option<Vec<&config::ComponentKey>>,
+    ) -> ReloadOutcome {
         new_config
             .healthchecks
             .set_require_healthy(self.require_healthy);
@@ -103,7 +107,7 @@ impl TopologyController {
 
         match self
             .topology
-            .reload_config_and_respawn(new_config, self.extra_context.clone())
+            .reload_config_and_respawn(new_config, self.extra_context.clone(), components_to_reload)
             .await
         {
             Ok(true) => {

src/topology/running.rs

@@ -227,6 +227,7 @@ impl RunningTopology {
         &mut self,
         new_config: Config,
         extra_context: ExtraContext,
+        components_to_reload: Option<Vec<&ComponentKey>>,
     ) -> Result<bool, ()> {
         info!("Reloading running topology with new configuration.");
 
@@ -244,7 +245,9 @@ impl RunningTopology {
         //
         // We also shutdown any component that is simply being removed entirely.
         let diff = ConfigDiff::new(&self.config, &new_config);
-        let buffers = self.shutdown_diff(&diff, &new_config).await;
+        let buffers = self
+            .shutdown_diff(&diff, &new_config, components_to_reload)
+            .await;
 
         // Gives windows some time to make available any port
         // released by shutdown components.
@@ -349,6 +352,7 @@ impl RunningTopology {
         &mut self,
         diff: &ConfigDiff,
         new_config: &Config,
+        components_to_reload: Option<Vec<&ComponentKey>>,
     ) -> HashMap<ComponentKey, BuiltBuffer> {
         // First, we shutdown any changed/removed sources. This ensures that we can allow downstream
         // components to terminate naturally by virtue of the flow of events stopping.
@@ -532,7 +536,7 @@ impl RunningTopology {
         // they can naturally shutdown and allow us to recover their buffers if possible.
         let mut buffer_tx = HashMap::new();
 
-        let sinks_to_change = diff
+        let mut sinks_to_change = diff
             .sinks
             .to_change
             .iter()
@@ -543,6 +547,11 @@ impl RunningTopology {
                     .is_some()
             }))
             .collect::<Vec<_>>();
+
+        if let Some(mut components) = components_to_reload {
+            sinks_to_change.append(&mut components)
+        }
+
         for key in &sinks_to_change {
             debug!(component = %key, "Changing sink.");
             if reuse_buffers.contains(key) {

src/topology/test/doesnt_reload.rs

@@ -23,7 +23,7 @@ async fn topology_doesnt_reload_new_data_dir() {
     new_config.global.data_dir = Some(Path::new("/qwerty").to_path_buf());
 
     topology
-        .reload_config_and_respawn(new_config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(new_config.build().unwrap(), Default::default(), None)
         .await
         .unwrap();
 

src/topology/test/mod.rs

@@ -301,7 +301,7 @@ async fn topology_remove_one_source() {
     config.add_sink("out1", &["in1"], sink1);
 
     assert!(topology
-        .reload_config_and_respawn(config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -350,7 +350,7 @@ async fn topology_remove_one_sink() {
     config.add_sink("out1", &["in1"], basic_sink(10).1);
 
     assert!(topology
-        .reload_config_and_respawn(config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -403,7 +403,7 @@ async fn topology_remove_one_transform() {
     config.add_sink("out1", &["t2"], sink2);
 
     assert!(topology
-        .reload_config_and_respawn(config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -452,7 +452,7 @@ async fn topology_swap_source() {
     config.add_sink("out1", &["in2"], sink2);
 
     assert!(topology
-        .reload_config_and_respawn(config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -517,7 +517,7 @@ async fn topology_swap_transform() {
     config.add_sink("out1", &["t1"], sink2);
 
     assert!(topology
-        .reload_config_and_respawn(config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -569,7 +569,7 @@ async fn topology_swap_sink() {
     config.add_sink("out1", &["in1"], sink2);
 
     assert!(topology
-        .reload_config_and_respawn(config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -657,7 +657,7 @@ async fn topology_swap_transform_is_atomic() {
     config.add_sink("out1", &["t1"], basic_sink(10).1);
 
     assert!(topology
-        .reload_config_and_respawn(config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -693,7 +693,7 @@ async fn topology_rebuild_connected() {
     config.add_sink("out1", &["in1"], sink1);
 
     assert!(topology
-        .reload_config_and_respawn(config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -752,7 +752,7 @@ async fn topology_rebuild_connected_transform() {
     config.add_sink("out1", &["t2"], sink2);
 
     assert!(topology
-        .reload_config_and_respawn(config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -807,7 +807,7 @@ async fn topology_optional_healthcheck_does_not_fail_reload() {
     let (mut topology, _) = start_topology(config, false).await;
     let config = basic_config_with_sink_failing_healthcheck();
     assert!(topology
-        .reload_config_and_respawn(config, Default::default())
+        .reload_config_and_respawn(config, Default::default(), None)
         .await
         .unwrap());
 }
@@ -820,7 +820,7 @@ async fn topology_healthcheck_not_run_on_unchanged_reload() {
     let mut config = basic_config_with_sink_failing_healthcheck();
     config.healthchecks.require_healthy = true;
     assert!(topology
-        .reload_config_and_respawn(config, Default::default())
+        .reload_config_and_respawn(config, Default::default(), None)
         .await
         .unwrap());
 }
@@ -846,7 +846,7 @@ async fn topology_healthcheck_run_for_changes_on_reload() {
     let mut config = config.build().unwrap();
     config.healthchecks.require_healthy = true;
     assert!(!topology
-        .reload_config_and_respawn(config, Default::default())
+        .reload_config_and_respawn(config, Default::default(), None)
         .await
         .unwrap());
 }

src/topology/test/reload.rs

@@ -65,7 +65,7 @@ async fn topology_reuse_old_port() {
 
     let (mut topology, _) = start_topology(old_config.build().unwrap(), false).await;
     assert!(topology
-        .reload_config_and_respawn(new_config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(new_config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 }
@@ -90,7 +90,7 @@ async fn topology_rebuild_old() {
 
     let (mut topology, _) = start_topology(old_config.build().unwrap(), false).await;
     assert!(!topology
-        .reload_config_and_respawn(new_config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(new_config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 }
@@ -107,7 +107,7 @@ async fn topology_old() {
 
     let (mut topology, _) = start_topology(old_config.clone().build().unwrap(), false).await;
     assert!(topology
-        .reload_config_and_respawn(old_config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(old_config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 }
@@ -258,7 +258,7 @@ async fn topology_readd_input() {
     new_config.add_source("in2", internal_metrics_source());
     new_config.add_sink("out", &["in1"], prom_exporter_sink(address_0, 1));
     assert!(topology
-        .reload_config_and_respawn(new_config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(new_config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -268,7 +268,7 @@ async fn topology_readd_input() {
     new_config.add_source("in2", internal_metrics_source());
     new_config.add_sink("out", &["in1", "in2"], prom_exporter_sink(address_0, 1));
     assert!(topology
-        .reload_config_and_respawn(new_config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(new_config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 
@@ -299,7 +299,7 @@ async fn reload_sink_test(
 
     // Now reload the topology with the "new" configuration, and make sure that a component is now listening on `new_address`.
     assert!(topology
-        .reload_config_and_respawn(new_config, Default::default())
+        .reload_config_and_respawn(new_config, Default::default(), None)
         .await
         .unwrap());
 

src/topology/test/transient_state.rs

@@ -30,7 +30,7 @@ async fn closed_source() {
     topology.sources_finished().await;
 
     assert!(topology
-        .reload_config_and_respawn(new_config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(new_config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 }
@@ -52,7 +52,7 @@ async fn remove_sink() {
 
     let (mut topology, _) = start_topology(old_config.build().unwrap(), false).await;
     assert!(topology
-        .reload_config_and_respawn(new_config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(new_config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 }
@@ -75,7 +75,7 @@ async fn remove_transform() {
 
     let (mut topology, _) = start_topology(old_config.build().unwrap(), false).await;
     assert!(topology
-        .reload_config_and_respawn(new_config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(new_config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 }
@@ -99,7 +99,7 @@ async fn replace_transform() {
 
     let (mut topology, _) = start_topology(old_config.build().unwrap(), false).await;
     assert!(topology
-        .reload_config_and_respawn(new_config.build().unwrap(), Default::default())
+        .reload_config_and_respawn(new_config.build().unwrap(), Default::default(), None)
         .await
         .unwrap());
 }