feat(cli): Handle reload based on referenced file change (#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 <[email protected]>

* 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 <[email protected]>

* Update src/config/mod.rs

Co-authored-by: Pavlos Rontidis <[email protected]>

* Fix missing enclosed delimiter

* solve clippy issues

* cargo fmt

* Fix testing

* run cargo fmt

* add dedicated watcher test

---------

Co-authored-by: Pavlos Rontidis <[email protected]>

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.