From 26f201b236353fffee95e998af97bfd63a6af46b Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Mon, 23 Oct 2023 11:07:20 -0600 Subject: [PATCH 01/10] Rename `Pieces` to `TopologyPieces` --- src/config/unit_test/mod.rs | 4 ++-- src/topology/builder.rs | 8 ++++---- src/topology/mod.rs | 16 ++++++++-------- src/topology/running.rs | 26 +++++++++++++++++--------- src/validate.rs | 6 +++--- 5 files changed, 34 insertions(+), 26 deletions(-) diff --git a/src/config/unit_test/mod.rs b/src/config/unit_test/mod.rs index f61917d336835..b706a1d8d654c 100644 --- a/src/config/unit_test/mod.rs +++ b/src/config/unit_test/mod.rs @@ -31,14 +31,14 @@ use crate::{ signal, topology::{ self, - builder::{self, Pieces}, + builder::{self, TopologyPieces}, }, }; pub struct UnitTest { pub name: String, config: Config, - pieces: Pieces, + pieces: TopologyPieces, test_result_rxs: Vec>, } diff --git a/src/topology/builder.rs b/src/topology/builder.rs index 9215848d6c224..5fa84ccef4ac8 100644 --- a/src/topology/builder.rs +++ b/src/topology/builder.rs @@ -78,7 +78,7 @@ pub async fn build_pieces( config: &super::Config, diff: &ConfigDiff, buffers: HashMap, -) -> Result> { +) -> Result> { Builder::new(config, diff, buffers).build().await } @@ -116,7 +116,7 @@ impl<'a> Builder<'a> { } /// Builds the new pieces of the topology found in `self.diff`. - async fn build(mut self) -> Result> { + async fn build(mut self) -> Result> { let enrichment_tables = self.load_enrichment_tables().await; let source_tasks = self.build_sources().await; self.build_transforms(enrichment_tables).await; @@ -127,7 +127,7 @@ impl<'a> Builder<'a> { enrichment_tables.finish_load(); if self.errors.is_empty() { - Ok(Pieces { + Ok(TopologyPieces { inputs: self.inputs, outputs: Self::finalize_outputs(self.outputs), tasks: self.tasks, @@ -677,7 +677,7 @@ impl<'a> Builder<'a> { } } -pub struct Pieces { +pub struct TopologyPieces { pub(super) inputs: HashMap, Inputs)>, pub(crate) outputs: HashMap, fanout::ControlChannel>>, pub(super) tasks: HashMap, diff --git a/src/topology/mod.rs b/src/topology/mod.rs index 2b3991914cd8a..98dea2f631d88 100644 --- a/src/topology/mod.rs +++ b/src/topology/mod.rs @@ -25,21 +25,21 @@ use std::{ sync::{Arc, Mutex}, }; -pub use controller::{ReloadOutcome, SharedTopologyController, TopologyController}; use futures::{Future, FutureExt}; -pub(super) use running::RunningTopology; use tokio::sync::{mpsc, watch}; use vector_buffers::topology::channel::{BufferReceiverStream, BufferSender}; +pub use self::controller::{ReloadOutcome, SharedTopologyController, TopologyController}; +pub(super) use self::running::RunningTopology; + +use self::builder::TopologyPieces; +use self::task::{Task, TaskError, TaskResult}; use crate::{ config::{ComponentKey, Config, ConfigDiff, Inputs, OutputId}, event::EventArray, signal::ShutdownError, - topology::{builder::Pieces, task::Task}, }; -use self::task::{TaskError, TaskResult}; - type TaskHandle = tokio::task::JoinHandle; type BuiltBuffer = ( @@ -78,7 +78,7 @@ pub type WatchRx = watch::Receiver; pub async fn start_validated( config: Config, diff: ConfigDiff, - mut pieces: Pieces, + mut pieces: TopologyPieces, ) -> Option<( RunningTopology, ( @@ -131,7 +131,7 @@ pub async fn build_or_log_errors( config: &Config, diff: &ConfigDiff, buffers: HashMap, -) -> Option { +) -> Option { match builder::build_pieces(config, diff, buffers).await { Err(errors) => { for error in errors { @@ -145,7 +145,7 @@ pub async fn build_or_log_errors( pub(super) fn take_healthchecks( diff: &ConfigDiff, - pieces: &mut Pieces, + pieces: &mut TopologyPieces, ) -> Vec<(ComponentKey, Task)> { (&diff.sinks.to_change | &diff.sinks.to_add) .into_iter() diff --git a/src/topology/running.rs b/src/topology/running.rs index bb300dd7e5652..e6306b0be3b78 100644 --- a/src/topology/running.rs +++ b/src/topology/running.rs @@ -24,7 +24,7 @@ use crate::{ spawn_named, topology::{ build_or_log_errors, builder, - builder::Pieces, + builder::TopologyPieces, fanout::{ControlChannel, ControlMessage}, handle_errors, retain, take_healthchecks, task::TaskOutput, @@ -295,7 +295,7 @@ impl RunningTopology { pub(crate) async fn run_healthchecks( &mut self, diff: &ConfigDiff, - pieces: &mut Pieces, + pieces: &mut TopologyPieces, options: HealthcheckOptions, ) -> bool { if options.enabled { @@ -527,7 +527,11 @@ impl RunningTopology { } /// Connects all changed/added components in the given configuration diff. - pub(crate) async fn connect_diff(&mut self, diff: &ConfigDiff, new_pieces: &mut Pieces) { + pub(crate) async fn connect_diff( + &mut self, + diff: &ConfigDiff, + new_pieces: &mut TopologyPieces, + ) { debug!("Connecting changed/added component(s)."); // Update tap metadata @@ -654,7 +658,11 @@ impl RunningTopology { } } - async fn setup_outputs(&mut self, key: &ComponentKey, new_pieces: &mut builder::Pieces) { + async fn setup_outputs( + &mut self, + key: &ComponentKey, + new_pieces: &mut builder::TopologyPieces, + ) { let outputs = new_pieces.outputs.remove(key).unwrap(); for (port, output) in outputs { debug!(component = %key, output_id = ?port, "Configuring output for component."); @@ -672,7 +680,7 @@ impl RunningTopology { &mut self, key: &ComponentKey, diff: &ConfigDiff, - new_pieces: &mut builder::Pieces, + new_pieces: &mut builder::TopologyPieces, ) { let (tx, inputs) = new_pieces.inputs.remove(key).unwrap(); @@ -794,7 +802,7 @@ impl RunningTopology { } /// Starts any new or changed components in the given configuration diff. - pub(crate) fn spawn_diff(&mut self, diff: &ConfigDiff, mut new_pieces: Pieces) { + pub(crate) fn spawn_diff(&mut self, diff: &ConfigDiff, mut new_pieces: TopologyPieces) { for key in &diff.sources.to_change { debug!(message = "Spawning changed source.", key = %key); self.spawn_source(key, &mut new_pieces); @@ -826,7 +834,7 @@ impl RunningTopology { } } - fn spawn_sink(&mut self, key: &ComponentKey, new_pieces: &mut builder::Pieces) { + fn spawn_sink(&mut self, key: &ComponentKey, new_pieces: &mut builder::TopologyPieces) { let task = new_pieces.tasks.remove(key).unwrap(); let span = error_span!( "sink", @@ -869,7 +877,7 @@ impl RunningTopology { } } - fn spawn_transform(&mut self, key: &ComponentKey, new_pieces: &mut builder::Pieces) { + fn spawn_transform(&mut self, key: &ComponentKey, new_pieces: &mut builder::TopologyPieces) { let task = new_pieces.tasks.remove(key).unwrap(); let span = error_span!( "transform", @@ -912,7 +920,7 @@ impl RunningTopology { } } - fn spawn_source(&mut self, key: &ComponentKey, new_pieces: &mut builder::Pieces) { + fn spawn_source(&mut self, key: &ComponentKey, new_pieces: &mut builder::TopologyPieces) { let task = new_pieces.tasks.remove(key).unwrap(); let span = error_span!( "source", diff --git a/src/validate.rs b/src/validate.rs index bd35909e47822..df466306b5222 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -8,7 +8,7 @@ use exitcode::ExitCode; use crate::{ config::{self, Config, ConfigDiff}, - topology::{self, builder::Pieces}, + topology::{self, builder::TopologyPieces}, }; const TEMPORARY_DIRECTORY: &str = "validate_tmp"; @@ -185,7 +185,7 @@ async fn validate_components( config: &Config, diff: &ConfigDiff, fmt: &mut Formatter, -) -> Option { +) -> Option { match topology::builder::build_pieces(config, diff, HashMap::new()).await { Ok(pieces) => { fmt.success("Component configuration"); @@ -203,7 +203,7 @@ async fn validate_healthchecks( opts: &Opts, config: &Config, diff: &ConfigDiff, - pieces: &mut Pieces, + pieces: &mut TopologyPieces, fmt: &mut Formatter, ) -> bool { if !config.healthchecks.enabled { From 38bb830b3fa8a1e03836a71f79cc93cbaa0ed3b7 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Mon, 23 Oct 2023 11:30:48 -0600 Subject: [PATCH 02/10] Move `topology::start_validated` into `impl RunningTopology` --- src/app.rs | 2 +- src/components/validation/runner/mod.rs | 4 +- src/config/unit_test/mod.rs | 4 +- src/test_util/mod.rs | 2 +- src/topology/mod.rs | 54 +------------------ src/topology/running.rs | 69 +++++++++++++++++++++---- src/topology/test/end_to_end.rs | 12 +++-- src/topology/test/mod.rs | 6 +-- 8 files changed, 77 insertions(+), 76 deletions(-) diff --git a/src/app.rs b/src/app.rs index 18ac0c1798571..700ae48621cba 100644 --- a/src/app.rs +++ b/src/app.rs @@ -107,7 +107,7 @@ impl ApplicationConfig { #[cfg(feature = "api")] let api = config.api; - let result = topology::start_validated(config, diff, pieces).await; + let result = RunningTopology::start_validated(config, diff, pieces).await; let (topology, (graceful_crash_sender, graceful_crash_receiver)) = result.ok_or(exitcode::CONFIG)?; diff --git a/src/components/validation/runner/mod.rs b/src/components/validation/runner/mod.rs index f38bdc75a3f41..d1e062c32cc41 100644 --- a/src/components/validation/runner/mod.rs +++ b/src/components/validation/runner/mod.rs @@ -25,7 +25,7 @@ use crate::{ codecs::Encoder, components::validation::{RunnerMetrics, TestCase}, config::{ConfigBuilder, ConfigDiff}, - topology, + topology::{self, RunningTopology}, }; use super::{ @@ -486,7 +486,7 @@ fn spawn_component_topology( .await .unwrap(); let (topology, (_, mut crash_rx)) = - topology::start_validated(config, config_diff, pieces) + RunningTopology::start_validated(config, config_diff, pieces) .await .unwrap(); diff --git a/src/config/unit_test/mod.rs b/src/config/unit_test/mod.rs index b706a1d8d654c..0b573f3a68792 100644 --- a/src/config/unit_test/mod.rs +++ b/src/config/unit_test/mod.rs @@ -30,8 +30,8 @@ use crate::{ event::{Event, LogEvent, Value}, signal, topology::{ - self, builder::{self, TopologyPieces}, + RunningTopology, }, }; @@ -49,7 +49,7 @@ pub struct UnitTestResult { impl UnitTest { pub async fn run(self) -> UnitTestResult { let diff = config::ConfigDiff::initial(&self.config); - let (topology, _) = topology::start_validated(self.config, diff, self.pieces) + let (topology, _) = RunningTopology::start_validated(self.config, diff, self.pieces) .await .unwrap(); topology.sources_finished().await; diff --git a/src/test_util/mod.rs b/src/test_util/mod.rs index 655e872f9fe52..804e363188e5c 100644 --- a/src/test_util/mod.rs +++ b/src/test_util/mod.rs @@ -693,7 +693,7 @@ pub async fn start_topology( let pieces = topology::build_or_log_errors(&config, &diff, HashMap::new()) .await .unwrap(); - topology::start_validated(config, diff, pieces) + RunningTopology::start_validated(config, diff, pieces) .await .unwrap() } diff --git a/src/topology/mod.rs b/src/topology/mod.rs index 98dea2f631d88..97084eb1f1d52 100644 --- a/src/topology/mod.rs +++ b/src/topology/mod.rs @@ -30,7 +30,7 @@ use tokio::sync::{mpsc, watch}; use vector_buffers::topology::channel::{BufferReceiverStream, BufferSender}; pub use self::controller::{ReloadOutcome, SharedTopologyController, TopologyController}; -pub(super) use self::running::RunningTopology; +pub use self::running::RunningTopology; use self::builder::TopologyPieces; use self::task::{Task, TaskError, TaskResult}; @@ -75,58 +75,6 @@ pub struct TapResource { type WatchTx = watch::Sender; pub type WatchRx = watch::Receiver; -pub async fn start_validated( - config: Config, - diff: ConfigDiff, - mut pieces: TopologyPieces, -) -> Option<( - RunningTopology, - ( - mpsc::UnboundedSender, - mpsc::UnboundedReceiver, - ), -)> { - let (abort_tx, abort_rx) = mpsc::unbounded_channel(); - - let expire_metrics = match ( - config.global.expire_metrics, - config.global.expire_metrics_secs, - ) { - (Some(e), None) => { - warn!( - "DEPRECATED: `expire_metrics` setting is deprecated and will be removed in a future version. Use `expire_metrics_secs` instead." - ); - Some(e.as_secs_f64()) - } - (Some(_), Some(_)) => { - error!("Cannot set both `expire_metrics` and `expire_metrics_secs`."); - return None; - } - (None, e) => e, - }; - - if let Err(error) = crate::metrics::Controller::get() - .expect("Metrics must be initialized") - .set_expiry(expire_metrics) - { - error!(message = "Invalid metrics expiry.", %error); - return None; - } - - let mut running_topology = RunningTopology::new(config, abort_tx.clone()); - - if !running_topology - .run_healthchecks(&diff, &mut pieces, running_topology.config.healthchecks) - .await - { - return None; - } - running_topology.connect_diff(&diff, &mut pieces).await; - running_topology.spawn_diff(&diff, pieces); - - Some((running_topology, (abort_tx, abort_rx))) -} - pub async fn build_or_log_errors( config: &Config, diff: &ConfigDiff, diff --git a/src/topology/running.rs b/src/topology/running.rs index e6306b0be3b78..7fe2558fe8e7e 100644 --- a/src/topology/running.rs +++ b/src/topology/running.rs @@ -15,21 +15,20 @@ use tracing::Instrument; use vector_buffers::topology::channel::BufferSender; use vector_common::trigger::DisabledTrigger; -use super::{TapOutput, TapResource}; +use super::{ + build_or_log_errors, builder, + builder::TopologyPieces, + fanout::{ControlChannel, ControlMessage}, + handle_errors, retain, take_healthchecks, + task::TaskOutput, + BuiltBuffer, TapOutput, TapResource, TaskHandle, WatchRx, WatchTx, +}; use crate::{ config::{ComponentKey, Config, ConfigDiff, HealthcheckOptions, Inputs, OutputId, Resource}, event::EventArray, shutdown::SourceShutdownCoordinator, signal::ShutdownError, spawn_named, - topology::{ - build_or_log_errors, builder, - builder::TopologyPieces, - fanout::{ControlChannel, ControlMessage}, - handle_errors, retain, take_healthchecks, - task::TaskOutput, - BuiltBuffer, TaskHandle, WatchRx, WatchTx, - }, }; #[allow(dead_code)] @@ -978,6 +977,58 @@ impl RunningTopology { self.source_tasks .insert(key.clone(), spawn_named(source_task, task_name.as_ref())); } + + pub async fn start_validated( + config: Config, + diff: ConfigDiff, + mut pieces: TopologyPieces, + ) -> Option<( + Self, + ( + mpsc::UnboundedSender, + mpsc::UnboundedReceiver, + ), + )> { + let (abort_tx, abort_rx) = mpsc::unbounded_channel(); + + let expire_metrics = match ( + config.global.expire_metrics, + config.global.expire_metrics_secs, + ) { + (Some(e), None) => { + warn!( + "DEPRECATED: `expire_metrics` setting is deprecated and will be removed in a future version. Use `expire_metrics_secs` instead." + ); + Some(e.as_secs_f64()) + } + (Some(_), Some(_)) => { + error!("Cannot set both `expire_metrics` and `expire_metrics_secs`."); + return None; + } + (None, e) => e, + }; + + if let Err(error) = crate::metrics::Controller::get() + .expect("Metrics must be initialized") + .set_expiry(expire_metrics) + { + error!(message = "Invalid metrics expiry.", %error); + return None; + } + + let mut running_topology = Self::new(config, abort_tx.clone()); + + if !running_topology + .run_healthchecks(&diff, &mut pieces, running_topology.config.healthchecks) + .await + { + return None; + } + running_topology.connect_diff(&diff, &mut pieces).await; + running_topology.spawn_diff(&diff, pieces); + + Some((running_topology, (abort_tx, abort_rx))) + } } fn get_changed_outputs(diff: &ConfigDiff, output_ids: Inputs) -> Vec { diff --git a/src/topology/test/end_to_end.rs b/src/topology/test/end_to_end.rs index 75580479ed58c..065413c246833 100644 --- a/src/topology/test/end_to_end.rs +++ b/src/topology/test/end_to_end.rs @@ -1,9 +1,5 @@ use std::{collections::HashMap, net::SocketAddr, sync::Arc}; -use crate::{ - config::{self, ConfigDiff, Format}, - test_util, topology, Error, -}; use hyper::{ service::{make_service_fn, service_fn}, Body, Request, Response, Server, StatusCode, @@ -14,6 +10,12 @@ use tokio::{ time::{timeout, Duration}, }; +use super::RunningTopology; +use crate::{ + config::{self, ConfigDiff, Format}, + test_util, topology, Error, +}; + type Lock = Arc>; pub async fn respond( @@ -104,7 +106,7 @@ uri = "http://{address2}/" let pieces = topology::build_or_log_errors(&config, &diff, HashMap::new()) .await .unwrap(); - let (_topology, _) = topology::start_validated(config, diff, pieces) + let (_topology, _) = RunningTopology::start_validated(config, diff, pieces) .await .unwrap(); diff --git a/src/topology/test/mod.rs b/src/topology/test/mod.rs index 5519c173b3345..886d7e9a37db1 100644 --- a/src/topology/test/mod.rs +++ b/src/topology/test/mod.rs @@ -19,7 +19,7 @@ use crate::{ }, start_topology, trace_init, }, - topology::{self, builder}, + topology::{self, builder, RunningTopology}, }; use futures::{future, stream, StreamExt}; use tokio::{ @@ -789,7 +789,7 @@ async fn topology_required_healthcheck_fails_start() { .await .unwrap(); - assert!(topology::start_validated(config, diff, pieces) + assert!(RunningTopology::start_validated(config, diff, pieces) .await .is_none()); } @@ -801,7 +801,7 @@ async fn topology_optional_healthcheck_does_not_fail_start() { let pieces = topology::build_or_log_errors(&config, &diff, HashMap::new()) .await .unwrap(); - assert!(topology::start_validated(config, diff, pieces) + assert!(RunningTopology::start_validated(config, diff, pieces) .await .is_some()); } From 207dcaa26def3a2f6d9b264adb302140f781cffb Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Mon, 23 Oct 2023 11:40:46 -0600 Subject: [PATCH 03/10] Move `topology::builder::build_pieces` into `impl TopologyPieces` --- src/config/mod.rs | 2 +- src/config/unit_test/mod.rs | 7 ++----- src/topology/builder.rs | 20 +++++++++++--------- src/topology/mod.rs | 4 ++-- src/topology/test/mod.rs | 4 ++-- src/validate.rs | 2 +- 6 files changed, 19 insertions(+), 20 deletions(-) diff --git a/src/config/mod.rs b/src/config/mod.rs index c65de18da77d4..23b6bf45bafb4 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -566,7 +566,7 @@ mod tests { let c2 = config::load_from_str(config, format).unwrap(); match ( config::warnings(&c2), - topology::builder::build_pieces(&c, &diff, HashMap::new()).await, + topology::TopologyPieces::build(&c, &diff, HashMap::new()).await, ) { (warnings, Ok(_pieces)) => Ok(warnings), (_, Err(errors)) => Err(errors), diff --git a/src/config/unit_test/mod.rs b/src/config/unit_test/mod.rs index 0b573f3a68792..4727ee6479615 100644 --- a/src/config/unit_test/mod.rs +++ b/src/config/unit_test/mod.rs @@ -29,10 +29,7 @@ use crate::{ }, event::{Event, LogEvent, Value}, signal, - topology::{ - builder::{self, TopologyPieces}, - RunningTopology, - }, + topology::{builder::TopologyPieces, RunningTopology}, }; pub struct UnitTest { @@ -422,7 +419,7 @@ async fn build_unit_test( } let config = config_builder.build()?; let diff = config::ConfigDiff::initial(&config); - let pieces = builder::build_pieces(&config, &diff, HashMap::new()).await?; + let pieces = TopologyPieces::build(&config, &diff, HashMap::new()).await?; Ok(UnitTest { name: test.name, diff --git a/src/topology/builder.rs b/src/topology/builder.rs index 5fa84ccef4ac8..8af2139d32b72 100644 --- a/src/topology/builder.rs +++ b/src/topology/builder.rs @@ -73,15 +73,6 @@ static TRANSFORM_CONCURRENCY_LIMIT: Lazy = Lazy::new(|| { const INTERNAL_SOURCES: [&str; 2] = ["internal_logs", "internal_metrics"]; -/// Builds only the new pieces, and doesn't check their topology. -pub async fn build_pieces( - config: &super::Config, - diff: &ConfigDiff, - buffers: HashMap, -) -> Result> { - Builder::new(config, diff, buffers).build().await -} - struct Builder<'a> { config: &'a super::Config, diff: &'a ConfigDiff, @@ -687,6 +678,17 @@ pub struct TopologyPieces { pub(crate) detach_triggers: HashMap, } +impl TopologyPieces { + /// Builds only the new pieces, and doesn't check their topology. + pub async fn build( + config: &super::Config, + diff: &ConfigDiff, + buffers: HashMap, + ) -> Result> { + Builder::new(config, diff, buffers).build().await + } +} + const fn filter_events_type(events: &EventArray, data_type: DataType) -> bool { match events { EventArray::Logs(_) => data_type.contains(DataType::Log), diff --git a/src/topology/mod.rs b/src/topology/mod.rs index 97084eb1f1d52..2d225888a845f 100644 --- a/src/topology/mod.rs +++ b/src/topology/mod.rs @@ -29,10 +29,10 @@ use futures::{Future, FutureExt}; use tokio::sync::{mpsc, watch}; use vector_buffers::topology::channel::{BufferReceiverStream, BufferSender}; +pub use self::builder::TopologyPieces; pub use self::controller::{ReloadOutcome, SharedTopologyController, TopologyController}; pub use self::running::RunningTopology; -use self::builder::TopologyPieces; use self::task::{Task, TaskError, TaskResult}; use crate::{ config::{ComponentKey, Config, ConfigDiff, Inputs, OutputId}, @@ -80,7 +80,7 @@ pub async fn build_or_log_errors( diff: &ConfigDiff, buffers: HashMap, ) -> Option { - match builder::build_pieces(config, diff, buffers).await { + match TopologyPieces::build(config, diff, buffers).await { Err(errors) => { for error in errors { error!(message = "Configuration error.", %error); diff --git a/src/topology/test/mod.rs b/src/topology/test/mod.rs index 886d7e9a37db1..5a744d39637fd 100644 --- a/src/topology/test/mod.rs +++ b/src/topology/test/mod.rs @@ -19,7 +19,7 @@ use crate::{ }, start_topology, trace_init, }, - topology::{self, builder, RunningTopology}, + topology::{self, RunningTopology, TopologyPieces}, }; use futures::{future, stream, StreamExt}; use tokio::{ @@ -915,7 +915,7 @@ async fn topology_transform_error_definition() { let config = config.build().unwrap(); let diff = ConfigDiff::initial(&config); - let errors = match builder::build_pieces(&config, &diff, HashMap::new()).await { + let errors = match TopologyPieces::build(&config, &diff, HashMap::new()).await { Ok(_) => panic!("build pieces should not succeed"), Err(err) => err, }; diff --git a/src/validate.rs b/src/validate.rs index df466306b5222..7fe876c168857 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -186,7 +186,7 @@ async fn validate_components( diff: &ConfigDiff, fmt: &mut Formatter, ) -> Option { - match topology::builder::build_pieces(config, diff, HashMap::new()).await { + match topology::TopologyPieces::build(config, diff, HashMap::new()).await { Ok(pieces) => { fmt.success("Component configuration"); Some(pieces) From f480dfd0c5c72df515f2e1627c9953d6c41e1bb0 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Mon, 23 Oct 2023 11:44:50 -0600 Subject: [PATCH 04/10] Move `topology::build_or_log_errors` into `TopologyPieces` --- src/app.rs | 5 +++-- src/components/validation/runner/mod.rs | 4 ++-- src/test_util/mod.rs | 4 ++-- src/topology/builder.rs | 20 ++++++++++++++++++-- src/topology/mod.rs | 16 ---------------- src/topology/running.rs | 9 ++++++--- src/topology/test/end_to_end.rs | 6 +++--- src/topology/test/mod.rs | 6 +++--- 8 files changed, 37 insertions(+), 33 deletions(-) diff --git a/src/app.rs b/src/app.rs index 700ae48621cba..b25215b50585d 100644 --- a/src/app.rs +++ b/src/app.rs @@ -30,7 +30,8 @@ use crate::{ heartbeat, signal::{ShutdownError, SignalHandler, SignalPair, SignalRx, SignalTo}, topology::{ - self, ReloadOutcome, RunningTopology, SharedTopologyController, TopologyController, + ReloadOutcome, RunningTopology, SharedTopologyController, TopologyController, + TopologyPieces, }, trace, }; @@ -100,7 +101,7 @@ impl ApplicationConfig { let enterprise = build_enterprise(&mut config, config_paths.clone())?; let diff = config::ConfigDiff::initial(&config); - let pieces = topology::build_or_log_errors(&config, &diff, HashMap::new()) + let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()) .await .ok_or(exitcode::CONFIG)?; diff --git a/src/components/validation/runner/mod.rs b/src/components/validation/runner/mod.rs index d1e062c32cc41..f2885d3ff1f23 100644 --- a/src/components/validation/runner/mod.rs +++ b/src/components/validation/runner/mod.rs @@ -25,7 +25,7 @@ use crate::{ codecs::Encoder, components::validation::{RunnerMetrics, TestCase}, config::{ConfigBuilder, ConfigDiff}, - topology::{self, RunningTopology}, + topology::{RunningTopology, TopologyPieces}, }; use super::{ @@ -482,7 +482,7 @@ fn spawn_component_topology( test_runtime.block_on(async move { debug!("Building component topology..."); - let pieces = topology::build_or_log_errors(&config, &config_diff, HashMap::new()) + let pieces = TopologyPieces::build_or_log_errors(&config, &config_diff, HashMap::new()) .await .unwrap(); let (topology, (_, mut crash_rx)) = diff --git a/src/test_util/mod.rs b/src/test_util/mod.rs index 804e363188e5c..3b04ea133cf31 100644 --- a/src/test_util/mod.rs +++ b/src/test_util/mod.rs @@ -42,7 +42,7 @@ use zstd::Decoder as ZstdDecoder; use crate::{ config::{Config, ConfigDiff, GenerateConfig}, signal::ShutdownError, - topology::{self, RunningTopology}, + topology::{RunningTopology, TopologyPieces}, trace, }; @@ -690,7 +690,7 @@ pub async fn start_topology( ) { config.healthchecks.set_require_healthy(require_healthy); let diff = ConfigDiff::initial(&config); - let pieces = topology::build_or_log_errors(&config, &diff, HashMap::new()) + let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()) .await .unwrap(); RunningTopology::start_validated(config, diff, pieces) diff --git a/src/topology/builder.rs b/src/topology/builder.rs index 8af2139d32b72..4f74b202325e2 100644 --- a/src/topology/builder.rs +++ b/src/topology/builder.rs @@ -41,8 +41,8 @@ use super::{ }; use crate::{ config::{ - ComponentKey, DataType, EnrichmentTableConfig, Input, Inputs, OutputId, ProxyConfig, - SinkContext, SourceContext, TransformContext, TransformOuter, TransformOutput, + ComponentKey, Config, DataType, EnrichmentTableConfig, Input, Inputs, OutputId, + ProxyConfig, SinkContext, SourceContext, TransformContext, TransformOuter, TransformOutput, }, event::{EventArray, EventContainer}, internal_events::EventsReceived, @@ -679,6 +679,22 @@ pub struct TopologyPieces { } impl TopologyPieces { + pub async fn build_or_log_errors( + config: &Config, + diff: &ConfigDiff, + buffers: HashMap, + ) -> Option { + match TopologyPieces::build(config, diff, buffers).await { + Err(errors) => { + for error in errors { + error!(message = "Configuration error.", %error); + } + None + } + Ok(new_pieces) => Some(new_pieces), + } + } + /// Builds only the new pieces, and doesn't check their topology. pub async fn build( config: &super::Config, diff --git a/src/topology/mod.rs b/src/topology/mod.rs index 2d225888a845f..ccb619de0b932 100644 --- a/src/topology/mod.rs +++ b/src/topology/mod.rs @@ -75,22 +75,6 @@ pub struct TapResource { type WatchTx = watch::Sender; pub type WatchRx = watch::Receiver; -pub async fn build_or_log_errors( - config: &Config, - diff: &ConfigDiff, - buffers: HashMap, -) -> Option { - match TopologyPieces::build(config, diff, buffers).await { - Err(errors) => { - for error in errors { - error!(message = "Configuration error.", %error); - } - None - } - Ok(new_pieces) => Some(new_pieces), - } -} - pub(super) fn take_healthchecks( diff: &ConfigDiff, pieces: &mut TopologyPieces, diff --git a/src/topology/running.rs b/src/topology/running.rs index 7fe2558fe8e7e..084a87279fa66 100644 --- a/src/topology/running.rs +++ b/src/topology/running.rs @@ -16,7 +16,7 @@ use vector_buffers::topology::channel::BufferSender; use vector_common::trigger::DisabledTrigger; use super::{ - build_or_log_errors, builder, + builder, builder::TopologyPieces, fanout::{ControlChannel, ControlMessage}, handle_errors, retain, take_healthchecks, @@ -247,7 +247,8 @@ impl RunningTopology { // Try to build all of the new components coming from the new configuration. If we can // successfully build them, we'll attempt to connect them up to the topology and spawn their // respective component tasks. - if let Some(mut new_pieces) = build_or_log_errors(&new_config, &diff, buffers.clone()).await + if let Some(mut new_pieces) = + TopologyPieces::build_or_log_errors(&new_config, &diff, buffers.clone()).await { // If healthchecks are configured for any of the changing/new components, try running // them before moving forward with connecting and spawning. In some cases, healthchecks @@ -272,7 +273,9 @@ impl RunningTopology { warn!("Failed to completely load new configuration. Restoring old configuration."); let diff = diff.flip(); - if let Some(mut new_pieces) = build_or_log_errors(&self.config, &diff, buffers).await { + if let Some(mut new_pieces) = + TopologyPieces::build_or_log_errors(&self.config, &diff, buffers).await + { if self .run_healthchecks(&diff, &mut new_pieces, self.config.healthchecks) .await diff --git a/src/topology/test/end_to_end.rs b/src/topology/test/end_to_end.rs index 065413c246833..e5e46fc87ac89 100644 --- a/src/topology/test/end_to_end.rs +++ b/src/topology/test/end_to_end.rs @@ -10,10 +10,10 @@ use tokio::{ time::{timeout, Duration}, }; -use super::RunningTopology; +use super::{RunningTopology, TopologyPieces}; use crate::{ config::{self, ConfigDiff, Format}, - test_util, topology, Error, + test_util, Error, }; type Lock = Arc>; @@ -103,7 +103,7 @@ uri = "http://{address2}/" ) .unwrap(); let diff = ConfigDiff::initial(&config); - let pieces = topology::build_or_log_errors(&config, &diff, HashMap::new()) + let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()) .await .unwrap(); let (_topology, _) = RunningTopology::start_validated(config, diff, pieces) diff --git a/src/topology/test/mod.rs b/src/topology/test/mod.rs index 5a744d39637fd..22854cc340ddf 100644 --- a/src/topology/test/mod.rs +++ b/src/topology/test/mod.rs @@ -19,7 +19,7 @@ use crate::{ }, start_topology, trace_init, }, - topology::{self, RunningTopology, TopologyPieces}, + topology::{RunningTopology, TopologyPieces}, }; use futures::{future, stream, StreamExt}; use tokio::{ @@ -785,7 +785,7 @@ async fn topology_required_healthcheck_fails_start() { let mut config = basic_config_with_sink_failing_healthcheck(); config.healthchecks.require_healthy = true; let diff = ConfigDiff::initial(&config); - let pieces = topology::build_or_log_errors(&config, &diff, HashMap::new()) + let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()) .await .unwrap(); @@ -798,7 +798,7 @@ async fn topology_required_healthcheck_fails_start() { async fn topology_optional_healthcheck_does_not_fail_start() { let config = basic_config_with_sink_failing_healthcheck(); let diff = ConfigDiff::initial(&config); - let pieces = topology::build_or_log_errors(&config, &diff, HashMap::new()) + let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()) .await .unwrap(); assert!(RunningTopology::start_validated(config, diff, pieces) From ed0ffad56ab78509d09c4887084a18e3a7e7aa56 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Mon, 23 Oct 2023 11:50:02 -0600 Subject: [PATCH 05/10] Add helper function to `RunningTopology` for initial startup --- src/app.rs | 19 +++++-------------- src/components/validation/runner/mod.rs | 12 +++--------- src/test_util/mod.rs | 21 ++++----------------- src/topology/mod.rs | 2 +- src/topology/running.rs | 19 ++++++++++++------- src/topology/test/mod.rs | 13 ++----------- 6 files changed, 27 insertions(+), 59 deletions(-) diff --git a/src/app.rs b/src/app.rs index b25215b50585d..c653c3ba9eeb5 100644 --- a/src/app.rs +++ b/src/app.rs @@ -1,7 +1,5 @@ #![allow(missing_docs)] -use std::{ - collections::HashMap, num::NonZeroUsize, path::PathBuf, process::ExitStatus, time::Duration, -}; +use std::{num::NonZeroUsize, path::PathBuf, process::ExitStatus, time::Duration}; use exitcode::ExitCode; use futures::StreamExt; @@ -29,10 +27,7 @@ use crate::{ config::{self, Config, ConfigPath}, heartbeat, signal::{ShutdownError, SignalHandler, SignalPair, SignalRx, SignalTo}, - topology::{ - ReloadOutcome, RunningTopology, SharedTopologyController, TopologyController, - TopologyPieces, - }, + topology::{ReloadOutcome, RunningTopology, SharedTopologyController, TopologyController}, trace, }; @@ -100,17 +95,13 @@ impl ApplicationConfig { #[cfg(feature = "enterprise")] let enterprise = build_enterprise(&mut config, config_paths.clone())?; - let diff = config::ConfigDiff::initial(&config); - let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()) - .await - .ok_or(exitcode::CONFIG)?; - #[cfg(feature = "api")] let api = config.api; - let result = RunningTopology::start_validated(config, diff, pieces).await; let (topology, (graceful_crash_sender, graceful_crash_receiver)) = - result.ok_or(exitcode::CONFIG)?; + RunningTopology::start_init_validated(config) + .await + .ok_or(exitcode::CONFIG)?; Ok(Self { config_paths, diff --git a/src/components/validation/runner/mod.rs b/src/components/validation/runner/mod.rs index f2885d3ff1f23..38c7364d2f233 100644 --- a/src/components/validation/runner/mod.rs +++ b/src/components/validation/runner/mod.rs @@ -24,8 +24,8 @@ use vector_core::{event::Event, EstimatedJsonEncodedSizeOf}; use crate::{ codecs::Encoder, components::validation::{RunnerMetrics, TestCase}, - config::{ConfigBuilder, ConfigDiff}, - topology::{RunningTopology, TopologyPieces}, + config::ConfigBuilder, + topology::RunningTopology, }; use super::{ @@ -471,7 +471,6 @@ fn spawn_component_topology( .build() .expect("config should not have any errors"); config.healthchecks.set_require_healthy(Some(true)); - let config_diff = ConfigDiff::initial(&config); _ = std::thread::spawn(move || { let test_runtime = Builder::new_current_thread() @@ -482,13 +481,8 @@ fn spawn_component_topology( test_runtime.block_on(async move { debug!("Building component topology..."); - let pieces = TopologyPieces::build_or_log_errors(&config, &config_diff, HashMap::new()) - .await - .unwrap(); let (topology, (_, mut crash_rx)) = - RunningTopology::start_validated(config, config_diff, pieces) - .await - .unwrap(); + RunningTopology::start_init_validated(config).await.unwrap(); debug!("Component topology built and spawned."); topology_started.mark_as_done(); diff --git a/src/test_util/mod.rs b/src/test_util/mod.rs index 3b04ea133cf31..2cace746c5921 100644 --- a/src/test_util/mod.rs +++ b/src/test_util/mod.rs @@ -40,9 +40,8 @@ use vector_core::event::{BatchNotifier, Event, EventArray, LogEvent}; use zstd::Decoder as ZstdDecoder; use crate::{ - config::{Config, ConfigDiff, GenerateConfig}, - signal::ShutdownError, - topology::{RunningTopology, TopologyPieces}, + config::{Config, GenerateConfig}, + topology::{RunningTopology, ShutdownErrorPair}, trace, }; @@ -681,21 +680,9 @@ impl CountReceiver { pub async fn start_topology( mut config: Config, require_healthy: impl Into>, -) -> ( - RunningTopology, - ( - tokio::sync::mpsc::UnboundedSender, - tokio::sync::mpsc::UnboundedReceiver, - ), -) { +) -> (RunningTopology, ShutdownErrorPair) { config.healthchecks.set_require_healthy(require_healthy); - let diff = ConfigDiff::initial(&config); - let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()) - .await - .unwrap(); - RunningTopology::start_validated(config, diff, pieces) - .await - .unwrap() + RunningTopology::start_init_validated(config).await.unwrap() } /// Collect the first `n` events from a stream while a future is spawned diff --git a/src/topology/mod.rs b/src/topology/mod.rs index ccb619de0b932..fe7a8e9486d9f 100644 --- a/src/topology/mod.rs +++ b/src/topology/mod.rs @@ -31,7 +31,7 @@ use vector_buffers::topology::channel::{BufferReceiverStream, BufferSender}; pub use self::builder::TopologyPieces; pub use self::controller::{ReloadOutcome, SharedTopologyController, TopologyController}; -pub use self::running::RunningTopology; +pub use self::running::{RunningTopology, ShutdownErrorPair}; use self::task::{Task, TaskError, TaskResult}; use crate::{ diff --git a/src/topology/running.rs b/src/topology/running.rs index 084a87279fa66..eb0e6a7fe1331 100644 --- a/src/topology/running.rs +++ b/src/topology/running.rs @@ -31,6 +31,11 @@ use crate::{ spawn_named, }; +pub type ShutdownErrorPair = ( + mpsc::UnboundedSender, + mpsc::UnboundedReceiver, +); + #[allow(dead_code)] pub struct RunningTopology { inputs: HashMap>, @@ -981,17 +986,17 @@ impl RunningTopology { .insert(key.clone(), spawn_named(source_task, task_name.as_ref())); } + pub async fn start_init_validated(config: Config) -> Option<(Self, ShutdownErrorPair)> { + let diff = ConfigDiff::initial(&config); + let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()).await?; + Self::start_validated(config, diff, pieces).await + } + pub async fn start_validated( config: Config, diff: ConfigDiff, mut pieces: TopologyPieces, - ) -> Option<( - Self, - ( - mpsc::UnboundedSender, - mpsc::UnboundedReceiver, - ), - )> { + ) -> Option<(Self, ShutdownErrorPair)> { let (abort_tx, abort_rx) = mpsc::unbounded_channel(); let expire_metrics = match ( diff --git a/src/topology/test/mod.rs b/src/topology/test/mod.rs index 22854cc340ddf..a17d42a5fce1a 100644 --- a/src/topology/test/mod.rs +++ b/src/topology/test/mod.rs @@ -784,12 +784,7 @@ async fn topology_rebuild_connected_transform() { async fn topology_required_healthcheck_fails_start() { let mut config = basic_config_with_sink_failing_healthcheck(); config.healthchecks.require_healthy = true; - let diff = ConfigDiff::initial(&config); - let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()) - .await - .unwrap(); - - assert!(RunningTopology::start_validated(config, diff, pieces) + assert!(RunningTopology::start_init_validated(config) .await .is_none()); } @@ -797,11 +792,7 @@ async fn topology_required_healthcheck_fails_start() { #[tokio::test] async fn topology_optional_healthcheck_does_not_fail_start() { let config = basic_config_with_sink_failing_healthcheck(); - let diff = ConfigDiff::initial(&config); - let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()) - .await - .unwrap(); - assert!(RunningTopology::start_validated(config, diff, pieces) + assert!(RunningTopology::start_init_validated(config) .await .is_some()); } From 9c9a583949344c7e390a476cdc3b137eafda8391 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Mon, 23 Oct 2023 16:43:37 -0600 Subject: [PATCH 06/10] Simplify topology start return type --- src/app.rs | 30 +++++++++---------- src/components/validation/runner/mod.rs | 2 +- .../traces/apm_stats/integration_tests.rs | 11 ++----- src/test_util/mod.rs | 4 +-- src/topology/mod.rs | 2 +- src/topology/running.rs | 15 ++++------ src/topology/test/crash.rs | 8 ++--- src/topology/test/reload.rs | 4 +-- 8 files changed, 32 insertions(+), 44 deletions(-) diff --git a/src/app.rs b/src/app.rs index c653c3ba9eeb5..424ea99d26dbd 100644 --- a/src/app.rs +++ b/src/app.rs @@ -7,10 +7,7 @@ use futures::StreamExt; use futures_util::future::BoxFuture; use once_cell::race::OnceNonZeroUsize; use openssl::provider::Provider; -use tokio::{ - runtime::{self, Runtime}, - sync::mpsc, -}; +use tokio::runtime::{self, Runtime}; use tokio_stream::wrappers::UnboundedReceiverStream; #[cfg(feature = "enterprise")] @@ -26,8 +23,11 @@ use crate::{ cli::{handle_config_errors, LogFormat, Opts, RootOpts}, config::{self, Config, ConfigPath}, heartbeat, - signal::{ShutdownError, SignalHandler, SignalPair, SignalRx, SignalTo}, - topology::{ReloadOutcome, RunningTopology, SharedTopologyController, TopologyController}, + signal::{SignalHandler, SignalPair, SignalRx, SignalTo}, + topology::{ + ReloadOutcome, RunningTopology, SharedTopologyController, ShutdownErrorReceiver, + TopologyController, + }, trace, }; @@ -46,8 +46,7 @@ use tokio::sync::broadcast::error::RecvError; pub struct ApplicationConfig { pub config_paths: Vec, pub topology: RunningTopology, - pub graceful_crash_sender: mpsc::UnboundedSender, - pub graceful_crash_receiver: mpsc::UnboundedReceiver, + pub graceful_crash_receiver: ShutdownErrorReceiver, #[cfg(feature = "api")] pub api: config::api::Options, #[cfg(feature = "enterprise")] @@ -98,15 +97,13 @@ impl ApplicationConfig { #[cfg(feature = "api")] let api = config.api; - let (topology, (graceful_crash_sender, graceful_crash_receiver)) = - RunningTopology::start_init_validated(config) - .await - .ok_or(exitcode::CONFIG)?; + let (topology, graceful_crash_receiver) = RunningTopology::start_init_validated(config) + .await + .ok_or(exitcode::CONFIG)?; Ok(Self { config_paths, topology, - graceful_crash_sender, graceful_crash_receiver, #[cfg(feature = "api")] api, @@ -137,8 +134,9 @@ impl ApplicationConfig { let error = error.to_string(); error!("An error occurred that Vector couldn't handle: {}.", error); _ = self - .graceful_crash_sender - .send(ShutdownError::ApiFailed { error }); + .topology + .abort_tx + .send(crate::signal::ShutdownError::ApiFailed { error }); None } } @@ -256,7 +254,7 @@ impl Application { pub struct StartedApplication { pub config_paths: Vec, - pub graceful_crash_receiver: mpsc::UnboundedReceiver, + pub graceful_crash_receiver: ShutdownErrorReceiver, pub signals: SignalPair, pub topology_controller: SharedTopologyController, pub openssl_providers: Option>, diff --git a/src/components/validation/runner/mod.rs b/src/components/validation/runner/mod.rs index 38c7364d2f233..cce737e9358be 100644 --- a/src/components/validation/runner/mod.rs +++ b/src/components/validation/runner/mod.rs @@ -481,7 +481,7 @@ fn spawn_component_topology( test_runtime.block_on(async move { debug!("Building component topology..."); - let (topology, (_, mut crash_rx)) = + let (topology, mut crash_rx) = RunningTopology::start_init_validated(config).await.unwrap(); debug!("Component topology built and spawned."); diff --git a/src/sinks/datadog/traces/apm_stats/integration_tests.rs b/src/sinks/datadog/traces/apm_stats/integration_tests.rs index 50c49066d9dfa..fbd2eb6d19b41 100644 --- a/src/sinks/datadog/traces/apm_stats/integration_tests.rs +++ b/src/sinks/datadog/traces/apm_stats/integration_tests.rs @@ -16,11 +16,10 @@ use tokio::time::{sleep, Duration}; use crate::{ config::ConfigBuilder, - signal::ShutdownError, sinks::datadog::traces::{apm_stats::StatsPayload, DatadogTracesConfig}, sources::datadog_agent::DatadogAgentConfig, test_util::{start_topology, trace_init}, - topology::RunningTopology, + topology::{RunningTopology, ShutdownErrorReceiver}, }; /// The port on which the Agent will send traces to vector, and vector `datadog_agent` source will @@ -320,13 +319,7 @@ fn validate_stats(agent_stats: &StatsPayload, vector_stats: &StatsPayload) { /// This creates a scenario where the stats payload that is output by the sink after processing the /// *second* batch of events (the second event) *should* contain the aggregated statistics of both /// of the trace events. i.e , the hit count for that bucket should be equal to "2" , not "1". -async fn start_vector() -> ( - RunningTopology, - ( - tokio::sync::mpsc::UnboundedSender, - tokio::sync::mpsc::UnboundedReceiver, - ), -) { +async fn start_vector() -> (RunningTopology, ShutdownErrorReceiver) { let dd_agent_address = format!("0.0.0.0:{}", vector_receive_port()); let source_config = toml::from_str::(&format!( diff --git a/src/test_util/mod.rs b/src/test_util/mod.rs index 2cace746c5921..b91c2ffb0dd97 100644 --- a/src/test_util/mod.rs +++ b/src/test_util/mod.rs @@ -41,7 +41,7 @@ use zstd::Decoder as ZstdDecoder; use crate::{ config::{Config, GenerateConfig}, - topology::{RunningTopology, ShutdownErrorPair}, + topology::{RunningTopology, ShutdownErrorReceiver}, trace, }; @@ -680,7 +680,7 @@ impl CountReceiver { pub async fn start_topology( mut config: Config, require_healthy: impl Into>, -) -> (RunningTopology, ShutdownErrorPair) { +) -> (RunningTopology, ShutdownErrorReceiver) { config.healthchecks.set_require_healthy(require_healthy); RunningTopology::start_init_validated(config).await.unwrap() } diff --git a/src/topology/mod.rs b/src/topology/mod.rs index fe7a8e9486d9f..06adb8d63bd39 100644 --- a/src/topology/mod.rs +++ b/src/topology/mod.rs @@ -31,7 +31,7 @@ use vector_buffers::topology::channel::{BufferReceiverStream, BufferSender}; pub use self::builder::TopologyPieces; pub use self::controller::{ReloadOutcome, SharedTopologyController, TopologyController}; -pub use self::running::{RunningTopology, ShutdownErrorPair}; +pub use self::running::{RunningTopology, ShutdownErrorReceiver}; use self::task::{Task, TaskError, TaskResult}; use crate::{ diff --git a/src/topology/running.rs b/src/topology/running.rs index eb0e6a7fe1331..c068dfe73664b 100644 --- a/src/topology/running.rs +++ b/src/topology/running.rs @@ -31,10 +31,7 @@ use crate::{ spawn_named, }; -pub type ShutdownErrorPair = ( - mpsc::UnboundedSender, - mpsc::UnboundedReceiver, -); +pub type ShutdownErrorReceiver = mpsc::UnboundedReceiver; #[allow(dead_code)] pub struct RunningTopology { @@ -47,7 +44,7 @@ pub struct RunningTopology { shutdown_coordinator: SourceShutdownCoordinator, detach_triggers: HashMap, pub(crate) config: Config, - abort_tx: mpsc::UnboundedSender, + pub(crate) abort_tx: mpsc::UnboundedSender, watch: (WatchTx, WatchRx), pub(crate) running: Arc, graceful_shutdown_duration: Option, @@ -986,7 +983,7 @@ impl RunningTopology { .insert(key.clone(), spawn_named(source_task, task_name.as_ref())); } - pub async fn start_init_validated(config: Config) -> Option<(Self, ShutdownErrorPair)> { + pub async fn start_init_validated(config: Config) -> Option<(Self, ShutdownErrorReceiver)> { let diff = ConfigDiff::initial(&config); let pieces = TopologyPieces::build_or_log_errors(&config, &diff, HashMap::new()).await?; Self::start_validated(config, diff, pieces).await @@ -996,7 +993,7 @@ impl RunningTopology { config: Config, diff: ConfigDiff, mut pieces: TopologyPieces, - ) -> Option<(Self, ShutdownErrorPair)> { + ) -> Option<(Self, ShutdownErrorReceiver)> { let (abort_tx, abort_rx) = mpsc::unbounded_channel(); let expire_metrics = match ( @@ -1024,7 +1021,7 @@ impl RunningTopology { return None; } - let mut running_topology = Self::new(config, abort_tx.clone()); + let mut running_topology = Self::new(config, abort_tx); if !running_topology .run_healthchecks(&diff, &mut pieces, running_topology.config.healthchecks) @@ -1035,7 +1032,7 @@ impl RunningTopology { running_topology.connect_diff(&diff, &mut pieces).await; running_topology.spawn_diff(&diff, pieces); - Some((running_topology, (abort_tx, abort_rx))) + Some((running_topology, abort_rx)) } } diff --git a/src/topology/test/crash.rs b/src/topology/test/crash.rs index 254dce3662fb8..21ce692d1a89c 100644 --- a/src/topology/test/crash.rs +++ b/src/topology/test/crash.rs @@ -33,7 +33,7 @@ async fn test_source_error() { let mut output_lines = CountReceiver::receive_lines(out_addr); - let (topology, (_, crash)) = start_topology(config.build().unwrap(), false).await; + let (topology, crash) = start_topology(config.build().unwrap(), false).await; // Wait for our source to become ready to accept connections, and likewise, wait for our sink's target server to // receive its connection from the output sink. @@ -78,7 +78,7 @@ async fn test_source_panic() { let mut output_lines = CountReceiver::receive_lines(out_addr); std::panic::set_hook(Box::new(|_| {})); // Suppress panic print on background thread - let (topology, (_, crash)) = start_topology(config.build().unwrap(), false).await; + let (topology, crash) = start_topology(config.build().unwrap(), false).await; // Wait for our source to become ready to accept connections, and likewise, wait for our sink's target server to // receive its connection from the output sink. @@ -125,7 +125,7 @@ async fn test_sink_error() { let mut output_lines = CountReceiver::receive_lines(out_addr); - let (topology, (_, crash)) = start_topology(config.build().unwrap(), false).await; + let (topology, crash) = start_topology(config.build().unwrap(), false).await; // Wait for our sources to become ready to accept connections, and likewise, wait for our sink's target server to // receive its connection from the output sink. @@ -174,7 +174,7 @@ async fn test_sink_panic() { let mut output_lines = CountReceiver::receive_lines(out_addr); std::panic::set_hook(Box::new(|_| {})); // Suppress panic print on background thread - let (topology, (_, crash)) = start_topology(config.build().unwrap(), false).await; + let (topology, crash) = start_topology(config.build().unwrap(), false).await; // Wait for our sources to become ready to accept connections, and likewise, wait for our sink's target server to // receive its connection from the output sink. diff --git a/src/topology/test/reload.rs b/src/topology/test/reload.rs index 91cb9439c6106..165a93b52f1a1 100644 --- a/src/topology/test/reload.rs +++ b/src/topology/test/reload.rs @@ -250,7 +250,7 @@ async fn topology_readd_input() { old_config.add_source("in1", internal_metrics_source()); old_config.add_source("in2", internal_metrics_source()); old_config.add_sink("out", &["in1", "in2"], prom_exporter_sink(address_0, 1)); - let (mut topology, (_, crash)) = start_topology(old_config.build().unwrap(), false).await; + let (mut topology, crash) = start_topology(old_config.build().unwrap(), false).await; // remove in2 let mut new_config = Config::builder(); @@ -289,7 +289,7 @@ async fn reload_sink_test( new_address: SocketAddr, ) { // Start a topology from the "old" configuration, which should result in a component listening on `old_address`. - let (mut topology, (_, crash)) = start_topology(old_config, false).await; + let (mut topology, crash) = start_topology(old_config, false).await; let mut crash_stream = UnboundedReceiverStream::new(crash); wait_for_tcp(old_address).await; From 50b44880294c9896389a397a574400a1739beed4 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Mon, 23 Oct 2023 17:02:51 -0600 Subject: [PATCH 07/10] Add support for internal topologies --- src/app.rs | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/src/app.rs b/src/app.rs index 424ea99d26dbd..259d846ea94eb 100644 --- a/src/app.rs +++ b/src/app.rs @@ -47,6 +47,7 @@ pub struct ApplicationConfig { pub config_paths: Vec, pub topology: RunningTopology, pub graceful_crash_receiver: ShutdownErrorReceiver, + pub internal_topologies: Vec, #[cfg(feature = "api")] pub api: config::api::Options, #[cfg(feature = "enterprise")] @@ -105,6 +106,7 @@ impl ApplicationConfig { config_paths, topology, graceful_crash_receiver, + internal_topologies: Vec::new(), #[cfg(feature = "api")] api, #[cfg(feature = "enterprise")] @@ -112,6 +114,14 @@ impl ApplicationConfig { }) } + pub async fn add_internal_config(&mut self, config: Config) -> Result<(), ExitCode> { + let Some((topology, _)) = RunningTopology::start_init_validated(config).await else { + return Err(exitcode::CONFIG); + }; + self.internal_topologies.push(topology); + Ok(()) + } + /// Configure the API server, if applicable #[cfg(feature = "api")] pub fn setup_api(&self, handle: &Handle) -> Option { @@ -244,6 +254,7 @@ impl Application { Ok(StartedApplication { config_paths: config.config_paths, + internal_topologies: config.internal_topologies, graceful_crash_receiver: config.graceful_crash_receiver, signals, topology_controller, @@ -254,6 +265,7 @@ impl Application { pub struct StartedApplication { pub config_paths: Vec, + pub internal_topologies: Vec, pub graceful_crash_receiver: ShutdownErrorReceiver, pub signals: SignalPair, pub topology_controller: SharedTopologyController, @@ -272,6 +284,7 @@ impl StartedApplication { signals, topology_controller, openssl_providers, + internal_topologies, } = self; let mut graceful_crash = UnboundedReceiverStream::new(graceful_crash_receiver); @@ -304,6 +317,7 @@ impl StartedApplication { signal_rx, topology_controller, openssl_providers, + internal_topologies, } } } @@ -359,6 +373,7 @@ pub struct FinishedApplication { pub signal_rx: SignalRx, pub topology_controller: SharedTopologyController, pub openssl_providers: Option>, + pub internal_topologies: Vec, } impl FinishedApplication { @@ -368,6 +383,7 @@ impl FinishedApplication { signal_rx, topology_controller, openssl_providers, + internal_topologies, } = self; // At this point, we'll have the only reference to the shared topology controller and can @@ -382,6 +398,11 @@ impl FinishedApplication { SignalTo::Quit => Self::quit(), _ => unreachable!(), }; + + for topology in internal_topologies { + topology.stop().await; + } + drop(openssl_providers); status } From 9388c039cd751248b78c4eab0326a60aa6c5c610 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Mon, 23 Oct 2023 17:16:57 -0600 Subject: [PATCH 08/10] Add demo of internal pipeline --- src/app.rs | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/src/app.rs b/src/app.rs index 259d846ea94eb..b612880b7b05b 100644 --- a/src/app.rs +++ b/src/app.rs @@ -157,6 +157,18 @@ impl ApplicationConfig { } } +const INTERNAL_CONFIG: &str = r#" +sources: + _internal_metrics: + type: internal_metrics +sinks: + _internal_console: + type: console + inputs: [_internal_metrics] + encoding: + codec: text +"#; + impl Application { pub fn run() -> ExitStatus { let (runtime, app) = Self::prepare_start().unwrap_or_else(|code| std::process::exit(code)); @@ -165,8 +177,12 @@ impl Application { } pub fn prepare_start() -> Result<(Runtime, StartedApplication), ExitCode> { - Self::prepare() - .and_then(|(runtime, app)| app.start(runtime.handle()).map(|app| (runtime, app))) + Self::prepare().and_then(|(runtime, mut app)| { + let config = config::load_from_str(INTERNAL_CONFIG, config::Format::Yaml) + .expect("Invalid internal config"); + runtime.block_on(app.config.add_internal_config(config))?; + app.start(runtime.handle()).map(|app| (runtime, app)) + }) } pub fn prepare() -> Result<(Runtime, Self), ExitCode> { From 6fd1b9f089596e524baa0bca8480d964789dfa5c Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Tue, 24 Oct 2023 18:36:05 -0600 Subject: [PATCH 09/10] Create the internal config manually --- src/app.rs | 43 +++++++++++++++++++++++++++++-------------- 1 file changed, 29 insertions(+), 14 deletions(-) diff --git a/src/app.rs b/src/app.rs index b612880b7b05b..c55f92a0d5671 100644 --- a/src/app.rs +++ b/src/app.rs @@ -157,17 +157,29 @@ impl ApplicationConfig { } } -const INTERNAL_CONFIG: &str = r#" -sources: - _internal_metrics: - type: internal_metrics -sinks: - _internal_console: - type: console - inputs: [_internal_metrics] - encoding: - codec: text -"#; +#[cfg(all(feature = "sources-internal_metrics", feature = "sinks-console"))] +fn internal_config() -> Config { + use crate::sinks::console; + let mut config = crate::config::ConfigBuilder::default(); + config.add_source( + "_internal_metrics", + crate::sources::internal_metrics::InternalMetricsConfig::default(), + ); + config.add_sink( + "_internal_console", + &["_internal_metrics"], + console::ConsoleSinkConfig { + target: console::Target::Stdout, + encoding: crate::codecs::EncodingConfigWithFraming::new( + None, + codecs::encoding::SerializerConfig::Text(Default::default()), + Default::default(), + ), + acknowledgements: None.into(), + }, + ); + config.build().expect("Could not build internal config") +} impl Application { pub fn run() -> ExitStatus { @@ -177,10 +189,13 @@ impl Application { } pub fn prepare_start() -> Result<(Runtime, StartedApplication), ExitCode> { + #[allow(unused_mut)] Self::prepare().and_then(|(runtime, mut app)| { - let config = config::load_from_str(INTERNAL_CONFIG, config::Format::Yaml) - .expect("Invalid internal config"); - runtime.block_on(app.config.add_internal_config(config))?; + #[cfg(all(feature = "sources-internal_metrics", feature = "sinks-console"))] + { + let config = internal_config(); + runtime.block_on(app.config.add_internal_config(config))?; + } app.start(runtime.handle()).map(|app| (runtime, app)) }) } From c91e35bcc0a66f1cd95ea94e661453c24ae14e08 Mon Sep 17 00:00:00 2001 From: Bruce Guenter Date: Wed, 25 Oct 2023 08:46:57 -0600 Subject: [PATCH 10/10] Revert the sample use of the interface --- src/app.rs | 35 ++--------------------------------- 1 file changed, 2 insertions(+), 33 deletions(-) diff --git a/src/app.rs b/src/app.rs index c55f92a0d5671..259d846ea94eb 100644 --- a/src/app.rs +++ b/src/app.rs @@ -157,30 +157,6 @@ impl ApplicationConfig { } } -#[cfg(all(feature = "sources-internal_metrics", feature = "sinks-console"))] -fn internal_config() -> Config { - use crate::sinks::console; - let mut config = crate::config::ConfigBuilder::default(); - config.add_source( - "_internal_metrics", - crate::sources::internal_metrics::InternalMetricsConfig::default(), - ); - config.add_sink( - "_internal_console", - &["_internal_metrics"], - console::ConsoleSinkConfig { - target: console::Target::Stdout, - encoding: crate::codecs::EncodingConfigWithFraming::new( - None, - codecs::encoding::SerializerConfig::Text(Default::default()), - Default::default(), - ), - acknowledgements: None.into(), - }, - ); - config.build().expect("Could not build internal config") -} - impl Application { pub fn run() -> ExitStatus { let (runtime, app) = Self::prepare_start().unwrap_or_else(|code| std::process::exit(code)); @@ -189,15 +165,8 @@ impl Application { } pub fn prepare_start() -> Result<(Runtime, StartedApplication), ExitCode> { - #[allow(unused_mut)] - Self::prepare().and_then(|(runtime, mut app)| { - #[cfg(all(feature = "sources-internal_metrics", feature = "sinks-console"))] - { - let config = internal_config(); - runtime.block_on(app.config.add_internal_config(config))?; - } - app.start(runtime.handle()).map(|app| (runtime, app)) - }) + Self::prepare() + .and_then(|(runtime, app)| app.start(runtime.handle()).map(|app| (runtime, app))) } pub fn prepare() -> Result<(Runtime, Self), ExitCode> {