From 15b8686f2d890197d7f8897211fff55e241fdc63 Mon Sep 17 00:00:00 2001 From: andrussal Date: Fri, 17 Jul 2026 07:16:38 +0200 Subject: [PATCH] feat(tf): unify cluster provisioning --- testing-framework/app/src/context.rs | 56 ++- testing-framework/app/src/deployment.rs | 5 +- testing-framework/app/src/extension.rs | 95 +++- testing-framework/app/src/local.rs | 242 +---------- .../core/src/scenario/control.rs | 4 + testing-framework/core/src/scenario/mod.rs | 5 + .../core/src/scenario/provisioning.rs | 398 +++++++++++++++++ .../deployers/local/src/cluster.rs | 261 +++++++++++ .../local/src/deployer/orchestrator.rs | 344 +++------------ testing-framework/deployers/local/src/lib.rs | 6 + .../deployers/local/src/manual/mod.rs | 100 ++--- .../deployers/local/src/provisioner.rs | 408 ++++++++++++++++++ 12 files changed, 1324 insertions(+), 600 deletions(-) create mode 100644 testing-framework/core/src/scenario/provisioning.rs create mode 100644 testing-framework/deployers/local/src/cluster.rs create mode 100644 testing-framework/deployers/local/src/provisioner.rs diff --git a/testing-framework/app/src/context.rs b/testing-framework/app/src/context.rs index 81b7393..dc9ad43 100644 --- a/testing-framework/app/src/context.rs +++ b/testing-framework/app/src/context.rs @@ -1,7 +1,8 @@ use testing_framework_core::scenario::{ - Application, DynError, NodeClients, internal::CleanupGuard, + Application, ClusterControlRequest, ClusterHandle, ClusterProvisioner, ClusterRequest, + DynError, NodeClients, internal::CleanupGuard, }; -use testing_framework_runner_local::LocalDeployerEnv; +use testing_framework_runner_local::{LocalClusterProvisioner, LocalDeployerEnv}; use crate::{ AppDeployError, AppDeployment, AppHandle, HandleRegistry, LocalAppCluster, @@ -14,22 +15,39 @@ use crate::{ /// outer scenario deployment and node clients while collecting application /// handles for later use by workloads. Exposed handles are dropped in reverse /// exposure order when the scenario runtime is released. -pub struct DeployContext { +pub struct DeployContext { deployment: E::Deployment, node_clients: NodeClients, + provisioner: P, handles: HandleRegistry, cleanup: AppCleanupStack, } -impl DeployContext +impl DeployContext where E: Application, { /// Creates a context for an outer scenario deployment and its node clients. pub fn new(deployment: E::Deployment, node_clients: NodeClients) -> Self { + Self::new_with_provisioner(deployment, node_clients, LocalClusterProvisioner) + } +} + +impl DeployContext +where + E: Application, + P: Send + Sync + 'static, +{ + /// Creates a context backed by an explicit cluster provisioner. + pub fn new_with_provisioner( + deployment: E::Deployment, + node_clients: NodeClients, + provisioner: P, + ) -> Self { Self { deployment, node_clients, + provisioner, handles: HandleRegistry::new(), cleanup: AppCleanupStack::default(), } @@ -41,7 +59,7 @@ where /// higher-level application handle. pub async fn deploy(&mut self, app: A) -> Result where - A: AppDeployment, + A: AppDeployment, { app.deploy(self).await } @@ -53,7 +71,7 @@ where /// scenario cleanup. pub async fn deploy_and_expose(&mut self, app: A) -> Result where - A: AppDeployment, + A: AppDeployment, { let handle = self.deploy(app).await?; @@ -132,6 +150,26 @@ where &self.handles } + /// Provisions a cluster through the active local backend. + pub async fn deploy_cluster( + &mut self, + request: ClusterRequest, + ) -> Result, DynError> + where + App: Application, + P: ClusterProvisioner, + { + let mut unit = self + .provisioner + .provision_cluster(request.with_control(ClusterControlRequest::Full)) + .await?; + let handle = unit.handle(); + if let Some(cleanup) = unit.take_cleanup() { + self.register_cleanup(cleanup); + } + Ok(handle) + } + /// Starts every node of an additional uniform local cluster. /// /// The cluster is registered with scenario cleanup before its handle is @@ -142,10 +180,10 @@ where ) -> Result, DynError> where App: LocalDeployerEnv, + P: ClusterProvisioner, { - let cluster = LocalAppCluster::start(deployment).await?; - self.register_cleanup(cluster.cleanup_guard()); - Ok(cluster) + self.deploy_cluster(ClusterRequest::managed(deployment)) + .await } /// Returns the outer scenario deployment descriptor. diff --git a/testing-framework/app/src/deployment.rs b/testing-framework/app/src/deployment.rs index c350097..f38e26d 100644 --- a/testing-framework/app/src/deployment.rs +++ b/testing-framework/app/src/deployment.rs @@ -1,5 +1,6 @@ use async_trait::async_trait; use testing_framework_core::scenario::{Application, DynError}; +use testing_framework_runner_local::LocalClusterProvisioner; use crate::DeployContext; @@ -14,7 +15,7 @@ impl AppHandle for T where T: Clone + Send + Sync + 'static {} /// Deploys one reusable application preset and returns its typed handle. #[async_trait] -pub trait AppDeployment: Send + 'static +pub trait AppDeployment: Send + 'static where E: Application, { @@ -26,5 +27,5 @@ where /// Child deployments can be composed with [`DeployContext::deploy`] or /// [`DeployContext::deploy_and_expose`]. Any handle exposed through the /// context remains available through the scenario runtime. - async fn deploy(self, ctx: &mut DeployContext) -> Result; + async fn deploy(self, ctx: &mut DeployContext) -> Result; } diff --git a/testing-framework/app/src/extension.rs b/testing-framework/app/src/extension.rs index eac0d43..7638fa6 100644 --- a/testing-framework/app/src/extension.rs +++ b/testing-framework/app/src/extension.rs @@ -3,6 +3,7 @@ use testing_framework_core::scenario::{ Application, CoreBuilderExt, DynError, NodeClients, PreparedRuntimeExtension, RunContext, RuntimeExtensionFactory, internal::CoreBuilderAccess, }; +use testing_framework_runner_local::LocalClusterProvisioner; use crate::{AppDeployment, AppHandle, AppRuntime, DeployContext}; @@ -10,29 +11,45 @@ use crate::{AppDeployment, AppHandle, AppRuntime, DeployContext}; /// /// A factory is normally installed through [`AppScenarioBuilderExt::with_app`] /// rather than constructed directly. -pub struct AppDeploymentFactory { +pub struct AppDeploymentFactory { app: A, + provisioner: P, } -impl AppDeploymentFactory { +impl AppDeploymentFactory { /// Creates a runtime extension factory for `app`. pub const fn new(app: A) -> Self { - Self { app } + Self { + app, + provisioner: LocalClusterProvisioner, + } + } +} + +impl AppDeploymentFactory { + /// Creates a runtime extension factory using an explicit provisioner. + pub const fn with_provisioner(app: A, provisioner: P) -> Self { + Self { app, provisioner } } } #[async_trait] -impl RuntimeExtensionFactory for AppDeploymentFactory +impl RuntimeExtensionFactory for AppDeploymentFactory where E: Application, - A: AppDeployment + Clone + Sync, + A: AppDeployment + Clone + Sync, + P: Clone + Send + Sync + 'static, { async fn prepare( &self, deployment: &E::Deployment, node_clients: NodeClients, ) -> Result { - let mut ctx = DeployContext::new(deployment.clone(), node_clients); + let mut ctx = DeployContext::new_with_provisioner( + deployment.clone(), + node_clients, + self.provisioner.clone(), + ); let handle = ctx.deploy(self.app.clone()).await?; @@ -63,6 +80,20 @@ pub trait AppScenarioBuilderExt: CoreBuilderAccess + Sized { { self.with_runtime_extension_factory(Box::new(AppDeploymentFactory::new(app))) } + + /// Registers an application deployment using an explicit backend + /// provisioner. + #[must_use] + fn with_app_using(self, app: A, provisioner: P) -> Self + where + A: AppDeployment + Clone + Sync, + P: Clone + Send + Sync + 'static, + { + self.with_runtime_extension_factory(Box::new(AppDeploymentFactory::with_provisioner( + app, + provisioner, + ))) + } } impl AppScenarioBuilderExt for T where T: CoreBuilderAccess {} @@ -136,7 +167,10 @@ mod tests { use async_trait::async_trait; use testing_framework_core::{ - scenario::{Application, DynError, NodeClients, RuntimeExtensionFactory}, + scenario::{ + Application, ClusterControlProfile, ClusterProvisioner, ClusterRequest, ClusterSource, + ClusterUnit, DynError, NodeClients, RuntimeExtensionFactory, + }, topology::DeploymentDescriptor, }; @@ -166,6 +200,43 @@ mod tests { dropped: Arc, } + #[derive(Clone)] + struct TestProvisioner; + + #[async_trait] + impl ClusterProvisioner for TestProvisioner { + async fn provision_cluster( + &self, + request: ClusterRequest, + ) -> Result, DynError> { + let deployment = match request.source() { + ClusterSource::Managed { deployment, .. } => Some(deployment.clone()), + ClusterSource::Attached { .. } | ClusterSource::External { .. } => None, + }; + Ok(ClusterUnit::new( + deployment, + NodeClients::default(), + ClusterControlProfile::FrameworkManaged, + )) + } + } + + #[derive(Clone)] + struct ProvisionedApp; + + #[async_trait] + impl AppDeployment for ProvisionedApp { + type Handle = testing_framework_core::scenario::ClusterHandle; + + async fn deploy( + self, + ctx: &mut DeployContext, + ) -> Result { + ctx.deploy_cluster(ClusterRequest::managed(TestDeployment)) + .await + } + } + #[derive(Clone)] struct OwnedHandle { _resource: Arc, @@ -226,6 +297,16 @@ mod tests { } } + #[tokio::test] + async fn factory_uses_explicit_backend_provisioner() { + let factory = AppDeploymentFactory::with_provisioner(ProvisionedApp, TestProvisioner); + + factory + .prepare(test_deployment(), NodeClients::default()) + .await + .expect("explicit provisioner should prepare the app"); + } + fn test_deployment() -> &'static TestDeployment { static DEPLOYMENT: TestDeployment = TestDeployment; &DEPLOYMENT diff --git a/testing-framework/app/src/local.rs b/testing-framework/app/src/local.rs index d3c11c9..a400d79 100644 --- a/testing-framework/app/src/local.rs +++ b/testing-framework/app/src/local.rs @@ -1,238 +1,8 @@ -use std::sync::{ - Arc, - atomic::{AtomicBool, Ordering}, -}; +use testing_framework_core::scenario::ClusterHandle; -use testing_framework_core::{ - scenario::{DynError, NodeClients, StartNodeOptions, StartedNode, internal::CleanupGuard}, - topology::DeploymentDescriptor, -}; -use testing_framework_runner_local::{LocalDeployerEnv, ManualCluster, ProcessDeployer}; - -/// Control handle for an additional uniform cluster of local processes. +/// Typed handle for a local cluster provisioned as part of an application +/// stack. /// -/// Unlike the outer scenario cluster, this cluster is deployed as one child of -/// a heterogeneous application stack. The scenario owns its lifetime and stops -/// every node during cleanup even if a handle clone remains elsewhere. -pub struct LocalAppCluster -where - E: LocalDeployerEnv, -{ - deployment: E::Deployment, - owner: Arc>, -} - -struct LocalClusterOwner -where - E: LocalDeployerEnv, -{ - cluster: ManualCluster, - closed: AtomicBool, -} - -impl Drop for LocalClusterOwner -where - E: LocalDeployerEnv, -{ - fn drop(&mut self) { - self.close(); - } -} - -impl LocalClusterOwner -where - E: LocalDeployerEnv, -{ - fn close(&self) { - if !self.closed.swap(true, Ordering::AcqRel) { - self.cluster.stop_all(); - } - } - - fn ensure_open(&self) -> Result<(), DynError> { - if self.closed.load(Ordering::Acquire) { - Err("local app cluster is no longer owned by an active run".into()) - } else { - Ok(()) - } - } -} - -struct LocalClusterCleanup -where - E: LocalDeployerEnv, -{ - owner: Arc>, -} - -impl CleanupGuard for LocalClusterCleanup -where - E: LocalDeployerEnv, -{ - fn cleanup(self: Box) { - self.owner.close(); - } -} - -impl Clone for LocalAppCluster -where - E: LocalDeployerEnv, -{ - fn clone(&self) -> Self { - Self { - deployment: self.deployment.clone(), - owner: Arc::clone(&self.owner), - } - } -} - -impl LocalAppCluster -where - E: LocalDeployerEnv, -{ - /// Starts every node described by `deployment` and waits for readiness. - pub(crate) async fn start(deployment: E::Deployment) -> Result { - let cluster = - ProcessDeployer::::default().manual_cluster_from_descriptors(deployment.clone()); - let owner = Arc::new(LocalClusterOwner { - cluster, - closed: AtomicBool::new(false), - }); - - start_all_nodes(&owner.cluster, deployment.node_count()).await?; - owner.cluster.wait_network_ready().await?; - - Ok(Self { deployment, owner }) - } - - /// Returns this cluster's deployment descriptor. - #[must_use] - pub fn deployment(&self) -> &E::Deployment { - &self.deployment - } - - /// Returns the number of nodes described by the deployment. - #[must_use] - pub fn node_count(&self) -> usize { - self.deployment.node_count() - } - - /// Returns a shared snapshot-capable collection of node clients. - #[must_use] - pub fn node_clients(&self) -> NodeClients { - self.owner.cluster.node_clients() - } - - /// Returns the client for `name`, if that node has started. - #[must_use] - pub fn node_client(&self, name: &str) -> Option { - self.owner.cluster.node_client(name) - } - - /// Returns the operating-system process ID for `name`, if it is running. - #[must_use] - pub fn node_pid(&self, name: &str) -> Option { - self.owner.cluster.node_pid(name) - } - - /// Returns a snapshot of all currently available node clients. - #[must_use] - pub fn clients(&self) -> Vec { - self.node_clients().snapshot() - } - - /// Returns the first available node client. - #[must_use] - pub fn first_client(&self) -> Option { - self.node_clients() - .with_clients(|clients| clients.first().cloned()) - } - - /// Starts the named node with default start options. - pub async fn start_node(&self, name: &str) -> Result, DynError> { - self.owner.ensure_open()?; - Ok(self.owner.cluster.start_node(name).await?) - } - - /// Starts the named node with explicit start options. - pub async fn start_node_with( - &self, - name: &str, - options: StartNodeOptions, - ) -> Result, DynError> { - self.owner.ensure_open()?; - Ok(self.owner.cluster.start_node_with(name, options).await?) - } - - /// Stops the named node. - pub async fn stop_node(&self, name: &str) -> Result<(), DynError> { - self.owner.ensure_open()?; - Ok(self.owner.cluster.stop_node(name).await?) - } - - /// Restarts the named node with its existing configuration. - pub async fn restart_node(&self, name: &str) -> Result<(), DynError> { - self.owner.ensure_open()?; - Ok(self.owner.cluster.restart_node(name).await?) - } - - /// Restarts the named node with explicit start options. - pub async fn restart_node_with( - &self, - name: &str, - options: StartNodeOptions, - ) -> Result<(), DynError> { - self.owner.ensure_open()?; - Ok(self.owner.cluster.restart_node_with(name, options).await?) - } - - /// Waits until the cluster-level network readiness condition succeeds. - pub async fn wait_network_ready(&self) -> Result<(), DynError> { - self.owner.ensure_open()?; - Ok(self.owner.cluster.wait_network_ready().await?) - } - - /// Waits until the named node reports ready. - pub async fn wait_node_ready(&self, name: &str) -> Result<(), DynError> { - self.owner.ensure_open()?; - Ok(self.owner.cluster.wait_node_ready(name).await?) - } - - /// Stops every node while keeping the cluster available for a later start. - pub fn stop_all(&self) -> Result<(), DynError> { - self.owner.ensure_open()?; - self.owner.cluster.stop_all(); - Ok(()) - } - - /// Starts every node with its original deployment and waits for readiness. - pub async fn start_all(&self) -> Result<(), DynError> { - self.owner.ensure_open()?; - start_all_nodes(&self.owner.cluster, self.node_count()).await?; - self.owner.cluster.wait_network_ready().await?; - Ok(()) - } - - /// Stops and starts every node, then waits for network readiness. - pub async fn restart_all(&self) -> Result<(), DynError> { - self.stop_all()?; - self.start_all().await - } - - pub(crate) fn cleanup_guard(&self) -> Box { - Box::new(LocalClusterCleanup { - owner: Arc::clone(&self.owner), - }) - } -} - -async fn start_all_nodes(cluster: &ManualCluster, node_count: usize) -> Result<(), DynError> -where - E: LocalDeployerEnv, -{ - for index in 0..node_count { - cluster.start_node(&format!("node-{index}")).await?; - } - - Ok(()) -} +/// This is the same cluster runtime used by local scenarios and manual tests; +/// the app layer only gives it a role-specific name. +pub type LocalAppCluster = ClusterHandle; diff --git a/testing-framework/core/src/scenario/control.rs b/testing-framework/core/src/scenario/control.rs index 951543b..4e3a4dc 100644 --- a/testing-framework/core/src/scenario/control.rs +++ b/testing-framework/core/src/scenario/control.rs @@ -33,6 +33,10 @@ pub trait NodeControlHandle: Send + Sync { Err("stop_node not supported by this deployer".into()) } + async fn wait_node_ready(&self, _name: &str) -> Result<(), DynError> { + Err("wait_node_ready not supported by this deployer".into()) + } + fn node_client(&self, _name: &str) -> Option { None } diff --git a/testing-framework/core/src/scenario/mod.rs b/testing-framework/core/src/scenario/mod.rs index 5190dcc..b17ee8f 100644 --- a/testing-framework/core/src/scenario/mod.rs +++ b/testing-framework/core/src/scenario/mod.rs @@ -14,6 +14,7 @@ mod deployment_policy; mod expectation; pub mod internal; mod observability; +mod provisioning; mod runtime; mod snapshot; mod sources; @@ -36,6 +37,10 @@ pub use definition::{Scenario, ScenarioBuildError, ScenarioBuilder}; pub use deployment_policy::{CleanupPolicy, DeploymentPolicy, RetryPolicy}; pub use expectation::Expectation; pub use observability::{ObservabilityCapabilityProvider, ObservabilityInputs}; +pub use provisioning::{ + ClusterControlRequest, ClusterHandle, ClusterProvisioner, ClusterRequest, ClusterSource, + ClusterStartMode, ClusterUnit, ProvisionedCluster, +}; pub use runtime::{ Deployer, HttpReadinessRequirement, NodeClients, PreparedRuntimeExtension, ReadinessError, RunContext, RunHandle, RunMetrics, Runner, RuntimeExtensionFactory, RuntimeExtensions, diff --git a/testing-framework/core/src/scenario/provisioning.rs b/testing-framework/core/src/scenario/provisioning.rs new file mode 100644 index 0000000..cd1e9a2 --- /dev/null +++ b/testing-framework/core/src/scenario/provisioning.rs @@ -0,0 +1,398 @@ +use std::sync::Arc; + +use async_trait::async_trait; + +use super::{ + Application, ClusterControlProfile, ClusterWaitHandle, DeploymentPolicy, DynError, + ExistingCluster, ExternalNodeSource, NodeClients, NodeControlHandle, StartNodeOptions, + StartedNode, internal::CleanupGuard, +}; +use crate::topology::DeploymentDescriptor; + +/// Source used to provision or connect to one cluster unit. +pub enum ClusterSource { + Managed { + deployment: E::Deployment, + external: Vec, + }, + Attached { + cluster: ExistingCluster, + external: Vec, + }, + External { + nodes: Vec, + }, +} + +impl Clone for ClusterSource { + fn clone(&self) -> Self { + match self { + Self::Managed { + deployment, + external, + } => Self::Managed { + deployment: deployment.clone(), + external: external.clone(), + }, + Self::Attached { cluster, external } => Self::Attached { + cluster: cluster.clone(), + external: external.clone(), + }, + Self::External { nodes } => Self::External { + nodes: nodes.clone(), + }, + } + } +} + +/// Determines whether managed nodes start during provisioning or on demand. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ClusterStartMode { + Eager, + OnDemand, +} + +/// Runtime control requested by the provisioning consumer. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ClusterControlRequest { + None, + Full, +} + +/// Backend-independent request for one cluster unit. +pub struct ClusterRequest { + source: ClusterSource, + policy: DeploymentPolicy, + start_mode: ClusterStartMode, + control: ClusterControlRequest, +} + +impl Clone for ClusterRequest { + fn clone(&self) -> Self { + Self { + source: self.source.clone(), + policy: self.policy, + start_mode: self.start_mode, + control: self.control, + } + } +} + +impl ClusterRequest { + #[must_use] + pub fn managed(deployment: E::Deployment) -> Self { + Self { + source: ClusterSource::Managed { + deployment, + external: Vec::new(), + }, + policy: DeploymentPolicy::default(), + start_mode: ClusterStartMode::Eager, + control: ClusterControlRequest::None, + } + } + + #[must_use] + pub fn attached(cluster: ExistingCluster) -> Self { + Self { + source: ClusterSource::Attached { + cluster, + external: Vec::new(), + }, + policy: DeploymentPolicy::default(), + start_mode: ClusterStartMode::Eager, + control: ClusterControlRequest::None, + } + } + + #[must_use] + pub fn external(nodes: Vec) -> Self { + Self { + source: ClusterSource::External { nodes }, + policy: DeploymentPolicy::default(), + start_mode: ClusterStartMode::Eager, + control: ClusterControlRequest::None, + } + } + + #[must_use] + pub fn with_external_nodes(mut self, nodes: Vec) -> Self { + match &mut self.source { + ClusterSource::Managed { external, .. } | ClusterSource::Attached { external, .. } => { + external.extend(nodes) + } + ClusterSource::External { nodes: external } => external.extend(nodes), + } + self + } + + #[must_use] + pub const fn with_policy(mut self, policy: DeploymentPolicy) -> Self { + self.policy = policy; + self + } + + #[must_use] + pub const fn with_start_mode(mut self, start_mode: ClusterStartMode) -> Self { + self.start_mode = start_mode; + self + } + + #[must_use] + pub const fn with_control(mut self, control: ClusterControlRequest) -> Self { + self.control = control; + self + } + + #[must_use] + pub const fn source(&self) -> &ClusterSource { + &self.source + } + + #[must_use] + pub const fn policy(&self) -> DeploymentPolicy { + self.policy + } + + #[must_use] + pub const fn start_mode(&self) -> ClusterStartMode { + self.start_mode + } + + #[must_use] + pub const fn control(&self) -> ClusterControlRequest { + self.control + } +} + +/// Runtime surfaces and lifetime returned for one provisioned cluster. +pub struct ClusterUnit { + deployment: Option, + node_clients: NodeClients, + control_profile: ClusterControlProfile, + node_control: Option>>, + cluster_wait: Option>>, + cleanup: Option>, +} + +/// Backend-independent access to one managed, attached, or external cluster. +pub struct ClusterHandle { + deployment: Option, + node_clients: NodeClients, + control_profile: ClusterControlProfile, + node_control: Option>>, + cluster_wait: Option>>, +} + +impl Clone for ClusterHandle { + fn clone(&self) -> Self { + Self { + deployment: self.deployment.clone(), + node_clients: self.node_clients.clone(), + control_profile: self.control_profile, + node_control: self.node_control.clone(), + cluster_wait: self.cluster_wait.clone(), + } + } +} + +impl ClusterHandle { + #[must_use] + pub fn deployment(&self) -> Option<&E::Deployment> { + self.deployment.as_ref() + } + + #[must_use] + pub fn node_count(&self) -> usize { + self.deployment + .as_ref() + .map_or(0, DeploymentDescriptor::node_count) + } + + #[must_use] + pub fn node_clients(&self) -> NodeClients { + self.node_clients.clone() + } + + #[must_use] + pub fn clients(&self) -> Vec { + self.node_clients.snapshot() + } + + #[must_use] + pub fn first_client(&self) -> Option { + self.node_clients + .with_clients(|clients| clients.first().cloned()) + } + + #[must_use] + pub fn node_client(&self, name: &str) -> Option { + self.node_control.as_ref()?.node_client(name) + } + + #[must_use] + pub fn node_pid(&self, name: &str) -> Option { + self.node_control.as_ref()?.node_pid(name) + } + + #[must_use] + pub const fn control_profile(&self) -> ClusterControlProfile { + self.control_profile + } + + pub async fn start_node(&self, name: &str) -> Result, DynError> { + self.require_control()?.start_node(name).await + } + + pub async fn start_node_with( + &self, + name: &str, + options: StartNodeOptions, + ) -> Result, DynError> { + self.require_control()?.start_node_with(name, options).await + } + + pub async fn stop_node(&self, name: &str) -> Result<(), DynError> { + self.require_control()?.stop_node(name).await + } + + pub async fn restart_node(&self, name: &str) -> Result<(), DynError> { + self.require_control()?.restart_node(name).await + } + + pub async fn restart_node_with( + &self, + name: &str, + options: StartNodeOptions, + ) -> Result<(), DynError> { + self.require_control()? + .restart_node_with(name, options) + .await + } + + pub async fn wait_network_ready(&self) -> Result<(), DynError> { + self.cluster_wait + .as_ref() + .ok_or_else(|| -> DynError { "cluster readiness is not available".into() })? + .wait_network_ready() + .await + } + + pub async fn wait_node_ready(&self, name: &str) -> Result<(), DynError> { + self.require_control()?.wait_node_ready(name).await + } + + fn require_control(&self) -> Result<&Arc>, DynError> { + self.node_control + .as_ref() + .ok_or_else(|| "cluster node control is not available".into()) + } +} + +/// Concrete backend handle paired with common runtime surfaces and ownership. +pub struct ProvisionedCluster { + handle: Option, + unit: ClusterUnit, +} + +impl ProvisionedCluster { + #[must_use] + pub const fn new(handle: Option, unit: ClusterUnit) -> Self { + Self { handle, unit } + } + + #[must_use] + pub fn handle(&self) -> Option<&H> { + self.handle.as_ref() + } + + #[must_use] + pub fn into_parts(self) -> (Option, ClusterUnit) { + (self.handle, self.unit) + } +} + +/// Backend operation used by app deployment contexts to provision a cluster. +#[async_trait] +pub trait ClusterProvisioner: Clone + Send + Sync + 'static { + async fn provision_cluster( + &self, + request: ClusterRequest, + ) -> Result, DynError>; +} + +impl ClusterUnit { + #[must_use] + pub fn new( + deployment: Option, + node_clients: NodeClients, + control_profile: ClusterControlProfile, + ) -> Self { + Self { + deployment, + node_clients, + control_profile, + node_control: None, + cluster_wait: None, + cleanup: None, + } + } + + #[must_use] + pub fn with_node_control(mut self, node_control: Arc>) -> Self { + self.node_control = Some(node_control); + self + } + + #[must_use] + pub fn with_cluster_wait(mut self, cluster_wait: Arc>) -> Self { + self.cluster_wait = Some(cluster_wait); + self + } + + #[must_use] + pub fn with_cleanup(mut self, cleanup: Box) -> Self { + self.cleanup = Some(cleanup); + self + } + + #[must_use] + pub fn deployment(&self) -> Option<&E::Deployment> { + self.deployment.as_ref() + } + + #[must_use] + pub const fn node_clients(&self) -> &NodeClients { + &self.node_clients + } + + #[must_use] + pub const fn control_profile(&self) -> ClusterControlProfile { + self.control_profile + } + + #[must_use] + pub fn node_control(&self) -> Option>> { + self.node_control.clone() + } + + #[must_use] + pub fn cluster_wait(&self) -> Option>> { + self.cluster_wait.clone() + } + + pub fn take_cleanup(&mut self) -> Option> { + self.cleanup.take() + } + + #[must_use] + pub fn handle(&self) -> ClusterHandle { + ClusterHandle { + deployment: self.deployment.clone(), + node_clients: self.node_clients.clone(), + control_profile: self.control_profile, + node_control: self.node_control.clone(), + cluster_wait: self.cluster_wait.clone(), + } + } +} diff --git a/testing-framework/deployers/local/src/cluster.rs b/testing-framework/deployers/local/src/cluster.rs new file mode 100644 index 0000000..adaa031 --- /dev/null +++ b/testing-framework/deployers/local/src/cluster.rs @@ -0,0 +1,261 @@ +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, +}; + +use async_trait::async_trait; +use testing_framework_core::scenario::{ + ClusterWaitHandle, DynError, ExternalNodeSource, NodeClients, NodeControlHandle, + ReadinessError, StartNodeOptions, StartedNode, internal::CleanupGuard, +}; + +use crate::{ + LocalDeployerEnv, NodeManager, NodeManagerSeed, env::Node, external::build_external_client, +}; + +/// Shared local cluster runtime used by scenarios, app deployments, and manual +/// tests. +pub struct LocalCluster { + deployment: E::Deployment, + owner: Arc>, +} + +struct LocalClusterOwner { + nodes: NodeManager, + closed: AtomicBool, +} + +impl LocalClusterOwner { + fn close(&self) { + if !self.closed.swap(true, Ordering::AcqRel) { + self.nodes.stop_all(); + } + } + + fn ensure_open(&self) -> Result<(), DynError> { + if self.closed.load(Ordering::Acquire) { + Err("local cluster is no longer owned by an active run".into()) + } else { + Ok(()) + } + } +} + +impl Drop for LocalClusterOwner { + fn drop(&mut self) { + self.close(); + } +} + +struct LocalClusterCleanup { + owner: Arc>, +} + +impl CleanupGuard for LocalClusterCleanup { + fn cleanup(self: Box) { + self.owner.close(); + } +} + +impl Clone for LocalCluster { + fn clone(&self) -> Self { + Self { + deployment: self.deployment.clone(), + owner: Arc::clone(&self.owner), + } + } +} + +impl LocalCluster { + pub(crate) fn empty(deployment: E::Deployment, keep_tempdir: bool) -> Self { + let nodes = NodeManager::new_with_seed( + deployment.clone(), + NodeClients::default(), + keep_tempdir, + NodeManagerSeed::default(), + ); + Self { + deployment, + owner: Arc::new(LocalClusterOwner { + nodes, + closed: AtomicBool::new(false), + }), + } + } + + pub(crate) fn initialize_with_nodes(&self, nodes: Vec>) { + self.owner.nodes.initialize_with_nodes(nodes); + } + + pub(crate) fn add_external_sources( + &self, + sources: impl IntoIterator, + ) -> Result<(), DynError> { + for source in sources { + let client = E::external_node_client(&source) + .or_else(|_| build_external_client::(&source))?; + self.owner.nodes.node_clients().add_node(client); + } + Ok(()) + } + + #[must_use] + pub fn deployment(&self) -> &E::Deployment { + &self.deployment + } + + #[must_use] + pub fn node_count(&self) -> usize { + testing_framework_core::topology::DeploymentDescriptor::node_count(&self.deployment) + } + + #[must_use] + pub fn node_clients(&self) -> NodeClients { + self.owner.nodes.node_clients() + } + + #[must_use] + pub fn node_client(&self, name: &str) -> Option { + self.owner.nodes.node_client(name) + } + + #[must_use] + pub fn node_pid(&self, name: &str) -> Option { + self.owner.nodes.node_pid(name) + } + + #[must_use] + pub fn clients(&self) -> Vec { + self.node_clients().snapshot() + } + + #[must_use] + pub fn first_client(&self) -> Option { + self.node_clients() + .with_clients(|clients| clients.first().cloned()) + } + + pub async fn start_node(&self, name: &str) -> Result, DynError> { + self.start_node_with(name, StartNodeOptions::default()) + .await + } + + pub async fn start_node_with( + &self, + name: &str, + options: StartNodeOptions, + ) -> Result, DynError> { + self.owner.ensure_open()?; + Ok(self.owner.nodes.start_node_with(name, options).await?) + } + + pub async fn stop_node(&self, name: &str) -> Result<(), DynError> { + self.owner.ensure_open()?; + Ok(self.owner.nodes.stop_node(name).await?) + } + + pub async fn restart_node(&self, name: &str) -> Result<(), DynError> { + self.owner.ensure_open()?; + Ok(self.owner.nodes.restart_node(name).await?) + } + + pub async fn restart_node_with( + &self, + name: &str, + options: StartNodeOptions, + ) -> Result<(), DynError> { + self.owner.ensure_open()?; + Ok(self.owner.nodes.restart_node_with(name, options).await?) + } + + pub async fn wait_network_ready(&self) -> Result<(), DynError> { + self.owner.ensure_open()?; + Ok(self.wait_network_ready_typed().await?) + } + + pub(crate) async fn wait_network_ready_typed(&self) -> Result<(), ReadinessError> { + self.owner.nodes.wait_network_ready().await + } + + pub async fn wait_node_ready(&self, name: &str) -> Result<(), DynError> { + self.owner.ensure_open()?; + Ok(self.owner.nodes.wait_node_ready(name).await?) + } + + pub fn stop_all(&self) -> Result<(), DynError> { + self.owner.ensure_open()?; + self.owner.nodes.stop_all(); + Ok(()) + } + + pub async fn start_all(&self) -> Result<(), DynError> { + self.owner.ensure_open()?; + for index in 0..self.node_count() { + self.start_node(&format!("node-{index}")).await?; + } + self.wait_network_ready().await + } + + pub async fn restart_all(&self) -> Result<(), DynError> { + self.stop_all()?; + self.start_all().await + } + + #[must_use] + #[doc(hidden)] + pub fn cleanup_guard(&self) -> Box { + Box::new(LocalClusterCleanup { + owner: Arc::clone(&self.owner), + }) + } +} + +#[async_trait] +impl NodeControlHandle for LocalCluster { + async fn restart_node(&self, name: &str) -> Result<(), DynError> { + self.restart_node(name).await + } + + async fn restart_node_with( + &self, + name: &str, + options: StartNodeOptions, + ) -> Result<(), DynError> { + self.restart_node_with(name, options).await + } + + async fn start_node(&self, name: &str) -> Result, DynError> { + self.start_node(name).await + } + + async fn start_node_with( + &self, + name: &str, + options: StartNodeOptions, + ) -> Result, DynError> { + self.start_node_with(name, options).await + } + + async fn stop_node(&self, name: &str) -> Result<(), DynError> { + self.stop_node(name).await + } + + async fn wait_node_ready(&self, name: &str) -> Result<(), DynError> { + self.wait_node_ready(name).await + } + + fn node_client(&self, name: &str) -> Option { + self.node_client(name) + } + + fn node_pid(&self, name: &str) -> Option { + self.node_pid(name) + } +} + +#[async_trait] +impl ClusterWaitHandle for LocalCluster { + async fn wait_network_ready(&self) -> Result<(), DynError> { + self.wait_network_ready().await + } +} diff --git a/testing-framework/deployers/local/src/deployer/orchestrator.rs b/testing-framework/deployers/local/src/deployer/orchestrator.rs index 47884c5..fe01336 100644 --- a/testing-framework/deployers/local/src/deployer/orchestrator.rs +++ b/testing-framework/deployers/local/src/deployer/orchestrator.rs @@ -1,59 +1,24 @@ -use std::{ - marker::PhantomData, - sync::{ - Arc, - atomic::{AtomicUsize, Ordering}, - }, - time::Duration, -}; +use std::{marker::PhantomData, sync::Arc, time::Duration}; use async_trait::async_trait; use testing_framework_core::{ scenario::{ - Application, ClusterControlProfile, ClusterMode, Deployer, DeploymentPolicy, DynError, - HttpReadinessRequirement, Metrics, NodeClients, NodeControlCapability, NodeControlHandle, - RetryPolicy, Runner, RuntimeExtensions, Scenario, ScenarioError, - internal::{ - CleanupGuard, RuntimeAssembly, SourceOrchestrationPlan, build_source_orchestration_plan, - }, + Application, ClusterControlProfile, ClusterControlRequest, ClusterMode, ClusterRequest, + ClusterWaitHandle, Deployer, DeploymentPolicy, DynError, Metrics, NodeClients, + NodeControlCapability, NodeControlHandle, Runner, RuntimeExtensions, Scenario, + ScenarioError, + internal::{CleanupGuard, RuntimeAssembly}, }, topology::DeploymentDescriptor, }; use thiserror::Error; -use tokio_retry::{ - RetryIf, - strategy::{ExponentialBackoff, jitter}, -}; -use tracing::{debug, info, warn}; +use tracing::info; use crate::{ - env::{LocalDeployerEnv, Node, wait_local_readiness}, - external::build_external_client, - keep_tempdir_from_env, + LocalClusterProvisioner, LocalClusterProvisionerError, env::LocalDeployerEnv, manual::ManualCluster, - node_control::{NodeManager, NodeManagerSeed}, }; -const READINESS_ATTEMPTS: usize = 3; -const READINESS_BACKOFF_BASE_MS: u64 = 250; -const READINESS_BACKOFF_MAX_SECS: u64 = 2; - -struct LocalProcessGuard { - nodes: Vec>, -} - -impl LocalProcessGuard { - fn new(nodes: Vec>) -> Self { - Self { nodes } - } -} - -impl CleanupGuard for LocalProcessGuard { - fn cleanup(self: Box) { - // Nodes own local processes; dropping them stops the processes. - drop(self.nodes); - } -} /// Spawns nodes as local processes. #[derive(Clone)] pub struct ProcessDeployer { @@ -100,37 +65,6 @@ pub enum ProcessDeployerError { }, } -#[derive(Debug, Error)] -enum RetryAttemptError { - #[error("failed to spawn local topology: {source}")] - Spawn { - #[source] - source: DynError, - }, - #[error("readiness probe failed: {source}")] - Readiness { - #[source] - source: DynError, - }, -} - -impl From for ProcessDeployerError { - fn from(value: RetryAttemptError) -> Self { - match value { - RetryAttemptError::Spawn { source } => Self::Spawn { source }, - RetryAttemptError::Readiness { source } => Self::ReadinessFailed { source }, - } - } -} - -#[derive(Clone, Copy)] -struct RetryExecutionConfig { - max_attempts: usize, - keep_tempdir: bool, - readiness_enabled: bool, - readiness_requirement: HttpReadinessRequirement, -} - impl From for ProcessDeployerError { fn from(value: ScenarioError) -> Self { match value { @@ -142,6 +76,19 @@ impl From for ProcessDeployerError { } } +impl From for ProcessDeployerError { + fn from(value: LocalClusterProvisionerError) -> Self { + match value { + LocalClusterProvisionerError::Spawn { source } => Self::Spawn { source }, + LocalClusterProvisionerError::Readiness { source } => Self::ReadinessFailed { source }, + LocalClusterProvisionerError::Source { source } => Self::SourceOrchestration { source }, + LocalClusterProvisionerError::AttachedUnsupported => Self::SourceOrchestration { + source: LocalClusterProvisionerError::AttachedUnsupported.into(), + }, + } + } +} + #[async_trait] impl Deployer for ProcessDeployer { type Error = ProcessDeployerError; @@ -189,24 +136,20 @@ impl ProcessDeployer { ) -> Result, ProcessDeployerError> { validate_supported_cluster_mode(scenario)?; - // Source planning is currently resolved here before node spawn/runtime setup. - let source_plan = build_source_orchestration_plan(scenario).map_err(|source| { - ProcessDeployerError::SourceOrchestration { - source: source.into(), - } - })?; - log_local_deploy_start( scenario.deployment().node_count(), scenario.deployment_policy(), false, ); - let nodes = Self::spawn_nodes_for_scenario(scenario, self.membership_check).await?; - let node_clients = NodeClients::::new(nodes.iter().map(|node| node.client()).collect()); - - let node_clients = merge_source_clients_for_local::(&source_plan, node_clients) - .map_err(|source| ProcessDeployerError::SourceOrchestration { source })?; + let provisioned = LocalClusterProvisioner + .provision( + cluster_request(scenario, ClusterControlRequest::None), + self.membership_check, + ) + .await?; + let (_cluster, mut unit) = provisioned.into_parts(); + let node_clients = unit.node_clients().clone(); let (runtime_extensions, runtime_cleanup) = scenario .prepare_runtime_extensions(node_clients.clone()) @@ -218,16 +161,14 @@ impl ProcessDeployer { node_clients, scenario.duration(), scenario.expectation_cooldown(), - scenario.cluster_control_profile(), + unit.control_profile(), runtime_extensions, runtime_cleanup, - None, + unit.node_control(), + unit.cluster_wait(), ) .await?; - - let cleanup_guard: Box = Box::new(LocalProcessGuard::::new(nodes)); - - Ok(runtime.assembly.build_runner(Some(cleanup_guard))) + Ok(runtime.assembly.build_runner(unit.take_cleanup())) } async fn deploy_with_node_control( @@ -236,24 +177,20 @@ impl ProcessDeployer { ) -> Result, ProcessDeployerError> { validate_supported_cluster_mode(scenario)?; - // Source planning is currently resolved here before node spawn/runtime setup. - let source_plan = build_source_orchestration_plan(scenario).map_err(|source| { - ProcessDeployerError::SourceOrchestration { - source: source.into(), - } - })?; - log_local_deploy_start( scenario.deployment().node_count(), scenario.deployment_policy(), true, ); - let nodes = Self::spawn_nodes_for_scenario(scenario, self.membership_check).await?; - let node_control = self.node_control_from(scenario, nodes); - let node_clients = - merge_source_clients_for_local::(&source_plan, node_control.node_clients()) - .map_err(|source| ProcessDeployerError::SourceOrchestration { source })?; + let provisioned = LocalClusterProvisioner + .provision( + cluster_request(scenario, ClusterControlRequest::Full), + self.membership_check, + ) + .await?; + let (_cluster, mut unit) = provisioned.into_parts(); + let node_clients = unit.node_clients().clone(); let (runtime_extensions, runtime_cleanup) = scenario .prepare_runtime_extensions(node_clients.clone()) .await @@ -263,67 +200,14 @@ impl ProcessDeployer { node_clients, scenario.duration(), scenario.expectation_cooldown(), - scenario.cluster_control_profile(), + unit.control_profile(), runtime_extensions, runtime_cleanup, - Some(node_control), + unit.node_control(), + unit.cluster_wait(), ) .await?; - - Ok(runtime.assembly.build_runner(None)) - } - - fn node_control_from( - &self, - scenario: &Scenario, - nodes: Vec>, - ) -> Arc> { - let node_control = Arc::new(NodeManager::new_with_seed( - scenario.deployment().clone(), - NodeClients::default(), - keep_tempdir(scenario.deployment_policy()), - NodeManagerSeed::default(), - )); - node_control.initialize_with_nodes(nodes); - node_control - } - - async fn spawn_nodes_for_scenario( - scenario: &Scenario, - membership_check: bool, - ) -> Result>, ProcessDeployerError> { - info!( - nodes = scenario.deployment().node_count(), - "spawning local nodes" - ); - Self::spawn_with_readiness_retry( - scenario.deployment(), - membership_check, - scenario.deployment_policy(), - ) - .await - } - - async fn spawn_with_readiness_retry( - descriptors: &E::Deployment, - membership_check: bool, - deployment_policy: DeploymentPolicy, - ) -> Result>, ProcessDeployerError> { - let (retry_policy, execution) = - build_retry_execution_config(deployment_policy, membership_check); - let attempts = Arc::new(AtomicUsize::new(0)); - let strategy = retry_backoff_strategy(retry_policy, execution.max_attempts); - let operation = { - let attempts = Arc::clone(&attempts); - move || { - let attempts = Arc::clone(&attempts); - async move { run_retry_attempt::(descriptors, execution, attempts).await } - } - }; - let should_retry = retry_decision(Arc::clone(&attempts), execution.max_attempts); - - let nodes = RetryIf::spawn(strategy, operation, should_retry).await?; - Ok(nodes) + Ok(runtime.assembly.build_runner(unit.take_cleanup())) } } @@ -343,31 +227,26 @@ fn ensure_local_cluster_mode(mode: ClusterMode) -> Result<(), ProcessDeployerErr Ok(()) } -fn merge_source_clients_for_local( - source_plan: &SourceOrchestrationPlan, - node_clients: NodeClients, -) -> Result, DynError> { - for source in source_plan.external_sources() { - let client = - E::external_node_client(source).or_else(|_| build_external_client::(source))?; - node_clients.add_node(client); - } - Ok(node_clients) -} - -fn build_retry_execution_config( - deployment_policy: DeploymentPolicy, - membership_check: bool, -) -> (RetryPolicy, RetryExecutionConfig) { - let retry_policy = retry_policy_from(deployment_policy); - let execution = RetryExecutionConfig { - max_attempts: retry_policy.max_attempts.max(1), - keep_tempdir: keep_tempdir(deployment_policy), - readiness_enabled: deployment_policy.readiness_enabled && membership_check, - readiness_requirement: deployment_policy.readiness_requirement, +fn cluster_request( + scenario: &Scenario, + control: ClusterControlRequest, +) -> ClusterRequest { + let request = match scenario.cluster_mode() { + ClusterMode::Managed => ClusterRequest::managed(scenario.deployment().clone()) + .with_external_nodes(scenario.external_nodes().to_vec()), + ClusterMode::ExistingCluster => ClusterRequest::attached( + scenario + .existing_cluster() + .expect("existing-cluster mode must contain an attached source") + .clone(), + ) + .with_external_nodes(scenario.external_nodes().to_vec()), + ClusterMode::ExternalOnly => ClusterRequest::external(scenario.external_nodes().to_vec()), }; - (retry_policy, execution) + request + .with_policy(scenario.deployment_policy()) + .with_control(control) } #[cfg(test)] @@ -393,87 +272,6 @@ mod tests { } } -async fn run_retry_attempt( - descriptors: &E::Deployment, - execution: RetryExecutionConfig, - attempts: Arc, -) -> Result>, RetryAttemptError> { - let attempt = attempts.fetch_add(1, Ordering::Relaxed) + 1; - let nodes = spawn_nodes_for_attempt::(descriptors, execution.keep_tempdir).await?; - run_readiness_for_attempt::(attempt, nodes, execution).await -} - -fn retry_policy_from(deployment_policy: DeploymentPolicy) -> RetryPolicy { - deployment_policy - .retry_policy - .unwrap_or_else(default_local_retry_policy) -} - -fn retry_backoff_strategy( - retry_policy: RetryPolicy, - max_attempts: usize, -) -> impl Iterator { - ExponentialBackoff::from_millis(retry_policy.base_delay.as_millis() as u64) - .max_delay(retry_policy.max_delay) - .map(jitter) - .take(max_attempts.saturating_sub(1)) -} - -async fn spawn_nodes_for_attempt( - descriptors: &E::Deployment, - keep_tempdir: bool, -) -> Result>, RetryAttemptError> { - NodeManager::::spawn_initial_nodes(descriptors, keep_tempdir) - .await - .map_err(|source| RetryAttemptError::Spawn { - source: source.into(), - }) -} - -async fn run_readiness_for_attempt( - attempt: usize, - nodes: Vec>, - execution: RetryExecutionConfig, -) -> Result>, RetryAttemptError> { - if !execution.readiness_enabled { - info!("skipping local readiness checks"); - return Ok(nodes); - } - - match wait_local_readiness::(&nodes, execution.readiness_requirement).await { - Ok(()) => { - info!(attempt, "local nodes are ready"); - Ok(nodes) - } - Err(source) => { - let error: DynError = source.into(); - debug!(attempt, error = ?error, "local readiness failed"); - drop(nodes); - Err(RetryAttemptError::Readiness { source: error }) - } - } -} - -fn retry_decision( - attempts: Arc, - max_attempts: usize, -) -> impl FnMut(&RetryAttemptError) -> bool { - move |error: &RetryAttemptError| { - let attempt = attempts.load(Ordering::Relaxed); - if attempt < max_attempts { - warn!( - attempt, - max_attempts, - error = %error, - "local spawn/readiness failed; retrying with backoff" - ); - true - } else { - false - } - } -} - impl Default for ProcessDeployer { fn default() -> Self { Self { @@ -483,18 +281,6 @@ impl Default for ProcessDeployer { } } -const fn default_local_retry_policy() -> RetryPolicy { - RetryPolicy::new( - READINESS_ATTEMPTS, - Duration::from_millis(READINESS_BACKOFF_BASE_MS), - Duration::from_secs(READINESS_BACKOFF_MAX_SECS), - ) -} - -fn keep_tempdir(policy: DeploymentPolicy) -> bool { - policy.cleanup_policy.preserve_artifacts || keep_tempdir_from_env() -} - fn log_local_deploy_start(node_count: usize, policy: DeploymentPolicy, has_node_control: bool) { info!( nodes = node_count, @@ -518,6 +304,7 @@ async fn run_context_for( runtime_extensions: RuntimeExtensions, runtime_cleanup: Option>, node_control: Option>>, + cluster_wait: Option>>, ) -> Result, ProcessDeployerError> { if node_clients.is_empty() && runtime_extensions.is_empty() { return Err(ProcessDeployerError::RuntimePreflight); @@ -536,6 +323,9 @@ async fn run_context_for( if let Some(node_control) = node_control { assembly = assembly.with_node_control(node_control); } + if let Some(cluster_wait) = cluster_wait { + assembly = assembly.with_cluster_wait(cluster_wait); + } Ok(RuntimeContext { assembly }) } diff --git a/testing-framework/deployers/local/src/lib.rs b/testing-framework/deployers/local/src/lib.rs index 72692b8..86adeea 100644 --- a/testing-framework/deployers/local/src/lib.rs +++ b/testing-framework/deployers/local/src/lib.rs @@ -1,10 +1,12 @@ pub mod binary; +mod cluster; mod deployer; pub mod env; mod external; mod manual; mod node_control; pub mod process; +mod provisioner; pub use binary::{ BinaryProvider, BinaryProviderError, BinaryProviderRef, BuildBinaryProvider, BuildCommand, @@ -12,6 +14,7 @@ pub use binary::{ DownloadProcessorFn, DownloadUrl, EnvBinaryProvider, FallbackBinaryProvider, PathBinaryProvider, }; +pub use cluster::LocalCluster; pub use deployer::{ProcessDeployer, ProcessDeployerError}; pub use env::{ BuiltNodeConfig, LocalBinaryApp, LocalBuildContext, LocalConfigArgMode, LocalDeployerEnv, @@ -27,6 +30,9 @@ pub use process::{ LaunchEnvVar, LaunchFile, LaunchSpec, NodeEndpointPort, NodeEndpoints, ProcessNode, ProcessSpawnError, allocate_available_port, }; +pub use provisioner::{ + LocalClusterProvisioner, LocalClusterProvisionerError, ProvisionedLocalCluster, +}; const KEEP_LOGS_ENV: &str = "TF_KEEP_LOGS"; diff --git a/testing-framework/deployers/local/src/manual/mod.rs b/testing-framework/deployers/local/src/manual/mod.rs index ffe0be2..6214044 100644 --- a/testing-framework/deployers/local/src/manual/mod.rs +++ b/testing-framework/deployers/local/src/manual/mod.rs @@ -7,51 +7,39 @@ use testing_framework_core::{ }; use thiserror::Error; -use crate::{ - env::LocalDeployerEnv, - external::build_external_client, - keep_tempdir_from_env, - node_control::{NodeManager, NodeManagerError, NodeManagerSeed}, -}; +use crate::{LocalCluster, LocalClusterProvisioner, LocalDeployerEnv}; #[derive(Debug, Error)] pub enum ManualClusterError { #[error(transparent)] - Dynamic(#[from] NodeManagerError), + Dynamic(#[from] DynError), } -/// Imperative, in-process cluster that can start nodes on demand. +/// Imperative view over the same local cluster runtime used by scenarios and +/// apps. pub struct ManualCluster { - nodes: NodeManager, + cluster: LocalCluster, } impl ManualCluster { - pub fn from_topology(descriptors: E::Deployment) -> Self { - let nodes = NodeManager::new_with_seed( - descriptors, - testing_framework_core::scenario::NodeClients::default(), - keep_tempdir_from_env(), - NodeManagerSeed::default(), - ); - - Self { nodes } + pub fn from_topology(deployment: E::Deployment) -> Self { + Self { + cluster: LocalClusterProvisioner.manual_cluster(deployment), + } } #[must_use] pub fn node_client(&self, name: &str) -> Option { - self.nodes.node_client(name) + self.cluster.node_client(name) } #[must_use] pub fn node_pid(&self, name: &str) -> Option { - self.nodes.node_pid(name) + self.cluster.node_pid(name) } pub async fn start_node(&self, name: &str) -> Result, ManualClusterError> { - Ok(self - .nodes - .start_node_with(name, StartNodeOptions::::default()) - .await?) + Ok(self.cluster.start_node(name).await?) } pub async fn start_node_with( @@ -59,15 +47,15 @@ impl ManualCluster { name: &str, options: StartNodeOptions, ) -> Result, ManualClusterError> { - Ok(self.nodes.start_node_with(name, options).await?) + Ok(self.cluster.start_node_with(name, options).await?) } pub fn stop_all(&self) { - self.nodes.stop_all(); + let _ = self.cluster.stop_all(); } pub async fn restart_node(&self, name: &str) -> Result<(), ManualClusterError> { - Ok(self.nodes.restart_node(name).await?) + Ok(self.cluster.restart_node(name).await?) } pub async fn restart_node_with( @@ -75,62 +63,45 @@ impl ManualCluster { name: &str, options: StartNodeOptions, ) -> Result<(), ManualClusterError> { - Ok(self.nodes.restart_node_with(name, options).await?) + Ok(self.cluster.restart_node_with(name, options).await?) } pub async fn stop_node(&self, name: &str) -> Result<(), ManualClusterError> { - Ok(self.nodes.stop_node(name).await?) + Ok(self.cluster.stop_node(name).await?) } pub async fn wait_network_ready(&self) -> Result<(), ReadinessError> { - self.nodes.wait_network_ready().await + self.cluster.wait_network_ready_typed().await } pub async fn wait_node_ready(&self, name: &str) -> Result<(), ManualClusterError> { - self.nodes.wait_node_ready(name).await?; - Ok(()) + Ok(self.cluster.wait_node_ready(name).await?) } #[must_use] pub fn node_clients(&self) -> NodeClients { - self.nodes.node_clients() + self.cluster.node_clients() } pub fn add_external_sources( &self, - external_sources: impl IntoIterator, + sources: impl IntoIterator, ) -> Result<(), DynError> { - let node_clients = self.nodes.node_clients(); - for source in external_sources { - let client = E::external_node_client(&source) - .or_else(|_| build_external_client::(&source))?; - node_clients.add_node(client); - } - - Ok(()) + self.cluster.add_external_sources(sources) } pub fn add_external_clients(&self, clients: impl IntoIterator) { - let node_clients = self.nodes.node_clients(); + let node_clients = self.cluster.node_clients(); for client in clients { node_clients.add_node(client); } } } -impl Drop for ManualCluster { - fn drop(&mut self) { - self.stop_all(); - } -} - #[async_trait::async_trait] impl NodeControlHandle for ManualCluster { async fn restart_node(&self, name: &str) -> Result<(), DynError> { - self.nodes - .restart_node(name) - .await - .map_err(|err| err.into()) + self.cluster.restart_node(name).await } async fn restart_node_with( @@ -138,21 +109,15 @@ impl NodeControlHandle for ManualCluster { name: &str, options: StartNodeOptions, ) -> Result<(), DynError> { - self.nodes - .restart_node_with(name, options) - .await - .map_err(|err| err.into()) + self.cluster.restart_node_with(name, options).await } async fn stop_node(&self, name: &str) -> Result<(), DynError> { - self.nodes.stop_node(name).await.map_err(|err| err.into()) + self.cluster.stop_node(name).await } async fn start_node(&self, name: &str) -> Result, DynError> { - self.nodes - .start_node_with(name, StartNodeOptions::::default()) - .await - .map_err(|err| err.into()) + self.cluster.start_node(name).await } async fn start_node_with( @@ -160,25 +125,22 @@ impl NodeControlHandle for ManualCluster { name: &str, options: StartNodeOptions, ) -> Result, DynError> { - self.nodes - .start_node_with(name, options) - .await - .map_err(|err| err.into()) + self.cluster.start_node_with(name, options).await } fn node_client(&self, name: &str) -> Option { - self.node_client(name) + self.cluster.node_client(name) } fn node_pid(&self, name: &str) -> Option { - self.node_pid(name) + self.cluster.node_pid(name) } } #[async_trait::async_trait] impl ClusterWaitHandle for ManualCluster { async fn wait_network_ready(&self) -> Result<(), DynError> { - self.wait_network_ready().await.map_err(|err| err.into()) + self.cluster.wait_network_ready().await } } diff --git a/testing-framework/deployers/local/src/provisioner.rs b/testing-framework/deployers/local/src/provisioner.rs new file mode 100644 index 0000000..bc5f402 --- /dev/null +++ b/testing-framework/deployers/local/src/provisioner.rs @@ -0,0 +1,408 @@ +use std::{ + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::Duration, +}; + +use testing_framework_core::scenario::{ + ClusterControlProfile, ClusterControlRequest, ClusterProvisioner, ClusterRequest, + ClusterSource, ClusterStartMode, ClusterUnit, DeploymentPolicy, DynError, + HttpReadinessRequirement, NodeClients, ProvisionedCluster, RetryPolicy, +}; +use tokio_retry::{ + RetryIf, + strategy::{ExponentialBackoff, jitter}, +}; +use tracing::{debug, info, warn}; + +use crate::{ + LocalCluster, LocalDeployerEnv, NodeManager, + env::{Node, wait_local_readiness}, + external::build_external_client, + keep_tempdir_from_env, +}; + +const READINESS_ATTEMPTS: usize = 3; +const READINESS_BACKOFF_BASE_MS: u64 = 250; +const READINESS_BACKOFF_MAX_SECS: u64 = 2; + +#[derive(Clone, Copy)] +struct RetryExecutionConfig { + max_attempts: usize, + keep_tempdir: bool, + readiness_enabled: bool, + readiness_requirement: HttpReadinessRequirement, +} + +#[derive(Debug, thiserror::Error)] +enum RetryAttemptError { + #[error("failed to spawn local topology: {source}")] + Spawn { + #[source] + source: DynError, + }, + #[error("readiness probe failed: {source}")] + Readiness { + #[source] + source: DynError, + }, +} + +#[derive(Debug, thiserror::Error)] +pub enum LocalClusterProvisionerError { + #[error("failed to spawn local topology: {source}")] + Spawn { + #[source] + source: DynError, + }, + #[error("readiness probe failed: {source}")] + Readiness { + #[source] + source: DynError, + }, + #[error("source orchestration failed: {source}")] + Source { + #[source] + source: DynError, + }, + #[error("local cluster provisioner does not support attached clusters")] + AttachedUnsupported, +} + +impl From for LocalClusterProvisionerError { + fn from(value: RetryAttemptError) -> Self { + match value { + RetryAttemptError::Spawn { source } => Self::Spawn { source }, + RetryAttemptError::Readiness { source } => Self::Readiness { source }, + } + } +} + +/// Local implementation of the shared cluster provisioning operation. +#[derive(Clone, Copy, Debug, Default)] +pub struct LocalClusterProvisioner; + +/// Concrete local handle plus the backend-independent runtime surfaces. +pub type ProvisionedLocalCluster = ProvisionedCluster>; + +impl LocalClusterProvisioner { + #[must_use] + pub fn manual_cluster( + &self, + deployment: E::Deployment, + ) -> LocalCluster { + LocalCluster::empty(deployment, keep_tempdir_from_env()) + } + + pub async fn provision( + &self, + request: ClusterRequest, + membership_check: bool, + ) -> Result, LocalClusterProvisionerError> { + match request.source().clone() { + ClusterSource::Managed { + deployment, + external, + } => { + let keep_tempdir = + request.policy().cleanup_policy.preserve_artifacts || keep_tempdir_from_env(); + let cluster = LocalCluster::empty(deployment.clone(), keep_tempdir); + + if request.start_mode() == ClusterStartMode::Eager { + let nodes = spawn_with_readiness_retry::( + &deployment, + membership_check, + request.policy(), + ) + .await?; + cluster.initialize_with_nodes(nodes); + } + + cluster + .add_external_sources(external) + .map_err(|source| LocalClusterProvisionerError::Source { source })?; + + let profile = if request.start_mode() == ClusterStartMode::OnDemand { + ClusterControlProfile::ManualControlled + } else { + ClusterControlProfile::FrameworkManaged + }; + let mut unit = ClusterUnit::new(Some(deployment), cluster.node_clients(), profile) + .with_cluster_wait(Arc::new(cluster.clone())) + .with_cleanup(cluster.cleanup_guard()); + + if request.control() == ClusterControlRequest::Full { + unit = unit.with_node_control(Arc::new(cluster.clone())); + } + + Ok(ProvisionedCluster::new(Some(cluster), unit)) + } + ClusterSource::External { nodes } => { + let clients = NodeClients::default(); + for source in nodes { + let client = E::external_node_client(&source) + .or_else(|_| build_external_client::(&source)) + .map_err(|source| LocalClusterProvisionerError::Source { source })?; + clients.add_node(client); + } + Ok(ProvisionedCluster::new( + None, + ClusterUnit::new(None, clients, ClusterControlProfile::ExternalUncontrolled), + )) + } + ClusterSource::Attached { .. } => { + Err(LocalClusterProvisionerError::AttachedUnsupported) + } + } + } +} + +#[async_trait::async_trait] +impl ClusterProvisioner for LocalClusterProvisioner { + async fn provision_cluster( + &self, + request: ClusterRequest, + ) -> Result, DynError> { + let (_, unit) = self + .provision(request, true) + .await + .map_err(DynError::from)? + .into_parts(); + Ok(unit) + } +} + +async fn spawn_with_readiness_retry( + deployment: &E::Deployment, + membership_check: bool, + policy: DeploymentPolicy, +) -> Result>, LocalClusterProvisionerError> { + let (retry_policy, execution) = build_retry_execution_config(policy, membership_check); + let attempts = Arc::new(AtomicUsize::new(0)); + let strategy = ExponentialBackoff::from_millis(retry_policy.base_delay.as_millis() as u64) + .max_delay(retry_policy.max_delay) + .map(jitter) + .take(execution.max_attempts.saturating_sub(1)); + let operation = { + let attempts = Arc::clone(&attempts); + move || { + let attempts = Arc::clone(&attempts); + async move { run_retry_attempt::(deployment, execution, attempts).await } + } + }; + let should_retry = { + let attempts = Arc::clone(&attempts); + move |error: &RetryAttemptError| { + let attempt = attempts.load(Ordering::Relaxed); + if attempt < execution.max_attempts { + warn!( + attempt, + max_attempts = execution.max_attempts, + error = %error, + "local spawn/readiness failed; retrying with backoff" + ); + true + } else { + false + } + } + }; + + RetryIf::spawn(strategy, operation, should_retry) + .await + .map_err(Into::into) +} + +fn build_retry_execution_config( + policy: DeploymentPolicy, + membership_check: bool, +) -> (RetryPolicy, RetryExecutionConfig) { + let retry_policy = policy.retry_policy.unwrap_or_else(|| { + RetryPolicy::new( + READINESS_ATTEMPTS, + Duration::from_millis(READINESS_BACKOFF_BASE_MS), + Duration::from_secs(READINESS_BACKOFF_MAX_SECS), + ) + }); + let execution = RetryExecutionConfig { + max_attempts: retry_policy.max_attempts.max(1), + keep_tempdir: policy.cleanup_policy.preserve_artifacts || keep_tempdir_from_env(), + readiness_enabled: policy.readiness_enabled && membership_check, + readiness_requirement: policy.readiness_requirement, + }; + (retry_policy, execution) +} + +async fn run_retry_attempt( + deployment: &E::Deployment, + execution: RetryExecutionConfig, + attempts: Arc, +) -> Result>, RetryAttemptError> { + let attempt = attempts.fetch_add(1, Ordering::Relaxed) + 1; + let nodes = NodeManager::::spawn_initial_nodes(deployment, execution.keep_tempdir) + .await + .map_err(|source| RetryAttemptError::Spawn { + source: source.into(), + })?; + + if !execution.readiness_enabled { + info!("skipping local readiness checks"); + return Ok(nodes); + } + + match wait_local_readiness::(&nodes, execution.readiness_requirement).await { + Ok(()) => { + info!(attempt, "local nodes are ready"); + Ok(nodes) + } + Err(source) => { + let source: DynError = source.into(); + debug!(attempt, error = ?source, "local readiness failed"); + drop(nodes); + Err(RetryAttemptError::Readiness { source }) + } + } +} + +#[cfg(test)] +mod tests { + use std::{collections::HashMap, path::Path}; + + use testing_framework_core::{ + scenario::{ + Application, ClusterControlProfile, ClusterControlRequest, ClusterRequest, + ClusterStartMode, DynError, ExistingCluster, ExternalNodeSource, StartNodeOptions, + }, + topology::DeploymentDescriptor, + }; + + use super::LocalClusterProvisioner; + use crate::{ + BuiltNodeConfig, LaunchSpec, LocalDeployerEnv, NodeConfigEntry, NodeEndpoints, + ProcessSpawnError, + }; + + #[derive(Clone)] + struct EmptyDeployment; + + impl DeploymentDescriptor for EmptyDeployment { + fn node_count(&self) -> usize { + 0 + } + } + + #[derive(Clone)] + struct EmptyConfig; + + struct TestEnv; + + #[async_trait::async_trait] + impl Application for TestEnv { + type Deployment = EmptyDeployment; + type NodeClient = String; + type NodeConfig = EmptyConfig; + + fn external_node_client(source: &ExternalNodeSource) -> Result { + Ok(source.endpoint().to_owned()) + } + } + + #[async_trait::async_trait] + impl LocalDeployerEnv for TestEnv { + fn build_node_config( + _topology: &Self::Deployment, + _index: usize, + _peer_ports_by_name: &HashMap, + _options: &StartNodeOptions, + _peer_ports: &[u16], + ) -> Result, DynError> { + unreachable!("empty deployment never builds node configs") + } + + fn build_initial_node_configs( + _topology: &Self::Deployment, + ) -> Result>, ProcessSpawnError> { + Ok(Vec::new()) + } + + fn build_launch_spec( + _config: &EmptyConfig, + _dir: &Path, + _label: &str, + ) -> Result { + unreachable!("empty deployment never launches nodes") + } + + fn node_endpoints(_config: &EmptyConfig) -> Result { + unreachable!("empty deployment has no endpoints") + } + + fn node_client(_endpoints: &NodeEndpoints) -> Result { + unreachable!("empty deployment has no clients") + } + } + + #[tokio::test] + async fn on_demand_managed_unit_exposes_shared_control_and_cleanup() { + let request = ClusterRequest::managed(EmptyDeployment) + .with_start_mode(ClusterStartMode::OnDemand) + .with_control(ClusterControlRequest::Full); + let provisioned = LocalClusterProvisioner + .provision::(request, true) + .await + .expect("manual unit should provision"); + let (cluster, mut unit) = provisioned.into_parts(); + let cluster = cluster.expect("managed unit should expose its concrete local handle"); + + assert_eq!( + unit.control_profile(), + ClusterControlProfile::ManualControlled + ); + assert!(unit.node_control().is_some()); + assert!(unit.cluster_wait().is_some()); + + unit.take_cleanup() + .expect("managed unit should own cleanup") + .cleanup(); + assert!(cluster.stop_all().is_err(), "cleanup must lock out clones"); + } + + #[tokio::test] + async fn external_unit_is_borrowed_and_uncontrolled() { + let source = ExternalNodeSource::new("external-0".into(), "http://node-0".into()); + let provisioned = LocalClusterProvisioner + .provision::(ClusterRequest::external(vec![source]), true) + .await + .expect("external unit should resolve"); + let (cluster, mut unit) = provisioned.into_parts(); + + assert!(cluster.is_none()); + assert_eq!(unit.node_clients().snapshot(), vec!["http://node-0"]); + assert_eq!( + unit.control_profile(), + ClusterControlProfile::ExternalUncontrolled + ); + assert!(unit.node_control().is_none()); + assert!(unit.cluster_wait().is_none()); + assert!(unit.take_cleanup().is_none()); + } + + #[tokio::test] + async fn attached_unit_reports_backend_support_gap() { + let request = ClusterRequest::::attached(ExistingCluster::for_compose_project( + "existing".into(), + )); + let error = LocalClusterProvisioner + .provision(request, true) + .await + .err() + .expect("local attached provisioning should be rejected"); + + assert_eq!( + error.to_string(), + "local cluster provisioner does not support attached clusters" + ); + } +}