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

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},
     };
@@ -182,12 +208,21 @@ mod tests {
         let dir = temp_dir().to_path_buf();
         let file_path = dir.join("vector.toml");
         let watcher_conf = WatcherConfig::RecommendedWatcher;
-
+        let component_file_path = vec![dir.join("tls.cert"), dir.join("tls.key")];
+        let component_config =
+            ComponentConfig::new(component_file_path, ComponentKey::from("http"));
         std::fs::create_dir(&dir).unwrap();
         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![component_config],
+            delay,
+        )
+        .unwrap();
 
         if !test(&mut file, delay * 5, signal_rx).await {
             panic!("Test timed out");
@@ -204,7 +239,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 +260,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 +281,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) => {