diff --git a/src/app.rs b/src/app.rs index 18ac0c1798571..259d846ea94eb 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; @@ -9,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")] @@ -28,9 +23,10 @@ use crate::{ cli::{handle_config_errors, LogFormat, Opts, RootOpts}, config::{self, Config, ConfigPath}, heartbeat, - signal::{ShutdownError, SignalHandler, SignalPair, SignalRx, SignalTo}, + signal::{SignalHandler, SignalPair, SignalRx, SignalTo}, topology::{ - self, ReloadOutcome, RunningTopology, SharedTopologyController, TopologyController, + ReloadOutcome, RunningTopology, SharedTopologyController, ShutdownErrorReceiver, + TopologyController, }, trace, }; @@ -50,8 +46,8 @@ 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, + pub internal_topologies: Vec, #[cfg(feature = "api")] pub api: config::api::Options, #[cfg(feature = "enterprise")] @@ -99,23 +95,18 @@ impl ApplicationConfig { #[cfg(feature = "enterprise")] 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()) - .await - .ok_or(exitcode::CONFIG)?; - #[cfg(feature = "api")] let api = config.api; - let result = topology::start_validated(config, diff, pieces).await; - let (topology, (graceful_crash_sender, graceful_crash_receiver)) = - result.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, + internal_topologies: Vec::new(), #[cfg(feature = "api")] api, #[cfg(feature = "enterprise")] @@ -123,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 { @@ -145,8 +144,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 } } @@ -254,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, @@ -264,7 +265,8 @@ impl Application { pub struct StartedApplication { pub config_paths: Vec, - pub graceful_crash_receiver: mpsc::UnboundedReceiver, + pub internal_topologies: Vec, + pub graceful_crash_receiver: ShutdownErrorReceiver, pub signals: SignalPair, pub topology_controller: SharedTopologyController, pub openssl_providers: Option>, @@ -282,6 +284,7 @@ impl StartedApplication { signals, topology_controller, openssl_providers, + internal_topologies, } = self; let mut graceful_crash = UnboundedReceiverStream::new(graceful_crash_receiver); @@ -314,6 +317,7 @@ impl StartedApplication { signal_rx, topology_controller, openssl_providers, + internal_topologies, } } } @@ -369,6 +373,7 @@ pub struct FinishedApplication { pub signal_rx: SignalRx, pub topology_controller: SharedTopologyController, pub openssl_providers: Option>, + pub internal_topologies: Vec, } impl FinishedApplication { @@ -378,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 @@ -392,6 +398,11 @@ impl FinishedApplication { SignalTo::Quit => Self::quit(), _ => unreachable!(), }; + + for topology in internal_topologies { + topology.stop().await; + } + drop(openssl_providers); status } diff --git a/src/components/validation/runner/mod.rs b/src/components/validation/runner/mod.rs index f38bdc75a3f41..cce737e9358be 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, + 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 = topology::build_or_log_errors(&config, &config_diff, HashMap::new()) - .await - .unwrap(); - let (topology, (_, mut crash_rx)) = - topology::start_validated(config, config_diff, pieces) - .await - .unwrap(); + let (topology, mut crash_rx) = + RunningTopology::start_init_validated(config).await.unwrap(); debug!("Component topology built and spawned."); topology_started.mark_as_done(); 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 f61917d336835..4727ee6479615 100644 --- a/src/config/unit_test/mod.rs +++ b/src/config/unit_test/mod.rs @@ -29,16 +29,13 @@ use crate::{ }, event::{Event, LogEvent, Value}, signal, - topology::{ - self, - builder::{self, Pieces}, - }, + topology::{builder::TopologyPieces, RunningTopology}, }; pub struct UnitTest { pub name: String, config: Config, - pieces: Pieces, + pieces: TopologyPieces, test_result_rxs: Vec>, } @@ -49,7 +46,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; @@ -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/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 655e872f9fe52..b91c2ffb0dd97 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::{self, RunningTopology}, + config::{Config, GenerateConfig}, + topology::{RunningTopology, ShutdownErrorReceiver}, 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, ShutdownErrorReceiver) { config.healthchecks.set_require_healthy(require_healthy); - let diff = ConfigDiff::initial(&config); - let pieces = topology::build_or_log_errors(&config, &diff, HashMap::new()) - .await - .unwrap(); - topology::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/builder.rs b/src/topology/builder.rs index 9215848d6c224..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, @@ -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, @@ -116,7 +107,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 +118,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 +668,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, @@ -687,6 +678,33 @@ pub struct Pieces { pub(crate) detach_triggers: HashMap, } +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: &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 2b3991914cd8a..06adb8d63bd39 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::builder::TopologyPieces; +pub use self::controller::{ReloadOutcome, SharedTopologyController, TopologyController}; +pub use self::running::{RunningTopology, ShutdownErrorReceiver}; + +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 = ( @@ -75,77 +75,9 @@ pub struct TapResource { type WatchTx = watch::Sender; pub type WatchRx = watch::Receiver; -pub async fn start_validated( - config: Config, - diff: ConfigDiff, - mut pieces: Pieces, -) -> 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, - buffers: HashMap, -) -> Option { - match builder::build_pieces(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 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..c068dfe73664b 100644 --- a/src/topology/running.rs +++ b/src/topology/running.rs @@ -15,23 +15,24 @@ use tracing::Instrument; use vector_buffers::topology::channel::BufferSender; use vector_common::trigger::DisabledTrigger; -use super::{TapOutput, TapResource}; +use super::{ + 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::Pieces, - fanout::{ControlChannel, ControlMessage}, - handle_errors, retain, take_healthchecks, - task::TaskOutput, - BuiltBuffer, TaskHandle, WatchRx, WatchTx, - }, }; +pub type ShutdownErrorReceiver = mpsc::UnboundedReceiver; + #[allow(dead_code)] pub struct RunningTopology { inputs: HashMap>, @@ -43,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, @@ -248,7 +249,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 @@ -273,7 +275,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 @@ -295,7 +299,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 +531,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 +662,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 +684,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 +806,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 +838,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 +881,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 +924,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", @@ -970,6 +982,58 @@ impl RunningTopology { self.source_tasks .insert(key.clone(), spawn_named(source_task, task_name.as_ref())); } + + 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 + } + + pub async fn start_validated( + config: Config, + diff: ConfigDiff, + mut pieces: TopologyPieces, + ) -> Option<(Self, ShutdownErrorReceiver)> { + 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); + + 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_rx)) + } } fn get_changed_outputs(diff: &ConfigDiff, output_ids: Inputs) -> Vec { 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/end_to_end.rs b/src/topology/test/end_to_end.rs index 75580479ed58c..e5e46fc87ac89 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, TopologyPieces}; +use crate::{ + config::{self, ConfigDiff, Format}, + test_util, Error, +}; + type Lock = Arc>; pub async fn respond( @@ -101,10 +103,10 @@ 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, _) = 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..a17d42a5fce1a 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::{RunningTopology, TopologyPieces}, }; use futures::{future, stream, StreamExt}; use tokio::{ @@ -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 = topology::build_or_log_errors(&config, &diff, HashMap::new()) - .await - .unwrap(); - - assert!(topology::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 = topology::build_or_log_errors(&config, &diff, HashMap::new()) - .await - .unwrap(); - assert!(topology::start_validated(config, diff, pieces) + assert!(RunningTopology::start_init_validated(config) .await .is_some()); } @@ -915,7 +906,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/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; diff --git a/src/validate.rs b/src/validate.rs index bd35909e47822..7fe876c168857 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,8 +185,8 @@ async fn validate_components( config: &Config, diff: &ConfigDiff, fmt: &mut Formatter, -) -> Option { - match topology::builder::build_pieces(config, diff, HashMap::new()).await { +) -> Option { + match topology::TopologyPieces::build(config, diff, HashMap::new()).await { Ok(pieces) => { fmt.success("Component configuration"); Some(pieces) @@ -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 {