logos-blockchain/logos-blockchain-testing
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-blockchain-testing.git synced 2026-07-18 13:39:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(tf): unify cluster provisioning

Browse Source
This commit is contained in:
andrussal 2026-07-17 07:16:38 +02:00
parent 4826328562
commit 15b8686f2d
12 changed files with 1324 additions and 600 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
testing-framework/app/src/context.rs
View File

@ -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<E: Application> {
pub struct DeployContext<E: Application, P = LocalClusterProvisioner> {
deployment: E::Deployment,
node_clients: NodeClients<E>,
provisioner: P,
handles: HandleRegistry,
cleanup: AppCleanupStack,
}
impl<E> DeployContext<E>
impl<E> DeployContext<E, LocalClusterProvisioner>
where
E: Application,
{
/// Creates a context for an outer scenario deployment and its node clients.
pub fn new(deployment: E::Deployment, node_clients: NodeClients<E>) -> Self {
Self::new_with_provisioner(deployment, node_clients, LocalClusterProvisioner)
}
}
impl<E, P> DeployContext<E, P>
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<E>,
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<A>(&mut self, app: A) -> Result<A::Handle, DynError>
where
A: AppDeployment<E>,
A: AppDeployment<E, P>,
{
app.deploy(self).await
}
@ -53,7 +71,7 @@ where
/// scenario cleanup.
pub async fn deploy_and_expose<A>(&mut self, app: A) -> Result<A::Handle, DynError>
where
A: AppDeployment<E>,
A: AppDeployment<E, P>,
{
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<App>(
&mut self,
request: ClusterRequest<App>,
) -> Result<ClusterHandle<App>, DynError>
where
App: Application,
P: ClusterProvisioner<App>,
{
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<LocalAppCluster<App>, DynError>
where
App: LocalDeployerEnv,
P: ClusterProvisioner<App>,
{
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.

5
testing-framework/app/src/deployment.rs
View File

@ -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<T> AppHandle for T where T: Clone + Send + Sync + 'static {}
/// Deploys one reusable application preset and returns its typed handle.
#[async_trait]
pub trait AppDeployment<E>: Send + 'static
pub trait AppDeployment<E, P = LocalClusterProvisioner>: 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<E>) -> Result<Self::Handle, DynError>;
async fn deploy(self, ctx: &mut DeployContext<E, P>) -> Result<Self::Handle, DynError>;
}

95
testing-framework/app/src/extension.rs
View File

@ -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<A> {
pub struct AppDeploymentFactory<A, P = LocalClusterProvisioner> {
app: A,
provisioner: P,
}
impl<A> AppDeploymentFactory<A> {
impl<A> AppDeploymentFactory<A, LocalClusterProvisioner> {
/// Creates a runtime extension factory for `app`.
pub const fn new(app: A) -> Self {
Self { app }
Self {
app,
provisioner: LocalClusterProvisioner,
}
}
}
impl<A, P> AppDeploymentFactory<A, P> {
/// 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<E, A> RuntimeExtensionFactory<E> for AppDeploymentFactory<A>
impl<E, A, P> RuntimeExtensionFactory<E> for AppDeploymentFactory<A, P>
where
E: Application,
A: AppDeployment<E> + Clone + Sync,
A: AppDeployment<E, P> + Clone + Sync,
P: Clone + Send + Sync + 'static,
{
async fn prepare(
&self,
deployment: &E::Deployment,
node_clients: NodeClients<E>,
) -> Result<PreparedRuntimeExtension, DynError> {
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<A, P>(self, app: A, provisioner: P) -> Self
where
A: AppDeployment<Self::Env, P> + Clone + Sync,
P: Clone + Send + Sync + 'static,
{
self.with_runtime_extension_factory(Box::new(AppDeploymentFactory::with_provisioner(
app,
provisioner,
)))
}
}
impl<T> 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<AtomicBool>,
}
#[derive(Clone)]
struct TestProvisioner;
#[async_trait]
impl ClusterProvisioner<TestEnv> for TestProvisioner {
async fn provision_cluster(
&self,
request: ClusterRequest<TestEnv>,
) -> Result<ClusterUnit<TestEnv>, 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<TestEnv, TestProvisioner> for ProvisionedApp {
type Handle = testing_framework_core::scenario::ClusterHandle<TestEnv>;
async fn deploy(
self,
ctx: &mut DeployContext<TestEnv, TestProvisioner>,
) -> Result<Self::Handle, DynError> {
ctx.deploy_cluster(ClusterRequest::managed(TestDeployment))
.await
}
}
#[derive(Clone)]
struct OwnedHandle {
_resource: Arc<DropProbe>,
@ -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

242
testing-framework/app/src/local.rs
View File

@ -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<E>
where
E: LocalDeployerEnv,
{
deployment: E::Deployment,
owner: Arc<LocalClusterOwner<E>>,
}
struct LocalClusterOwner<E>
where
E: LocalDeployerEnv,
{
cluster: ManualCluster<E>,
closed: AtomicBool,
}
impl<E> Drop for LocalClusterOwner<E>
where
E: LocalDeployerEnv,
{
fn drop(&mut self) {
self.close();
}
}
impl<E> LocalClusterOwner<E>
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<E>
where
E: LocalDeployerEnv,
{
owner: Arc<LocalClusterOwner<E>>,
}
impl<E> CleanupGuard for LocalClusterCleanup<E>
where
E: LocalDeployerEnv,
{
fn cleanup(self: Box<Self>) {
self.owner.close();
}
}
impl<E> Clone for LocalAppCluster<E>
where
E: LocalDeployerEnv,
{
fn clone(&self) -> Self {
Self {
deployment: self.deployment.clone(),
owner: Arc::clone(&self.owner),
}
}
}
impl<E> LocalAppCluster<E>
where
E: LocalDeployerEnv,
{
/// Starts every node described by `deployment` and waits for readiness.
pub(crate) async fn start(deployment: E::Deployment) -> Result<Self, DynError> {
let cluster =
ProcessDeployer::<E>::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<E> {
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<E::NodeClient> {
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<u32> {
self.owner.cluster.node_pid(name)
}
/// Returns a snapshot of all currently available node clients.
#[must_use]
pub fn clients(&self) -> Vec<E::NodeClient> {
self.node_clients().snapshot()
}
/// Returns the first available node client.
#[must_use]
pub fn first_client(&self) -> Option<E::NodeClient> {
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<StartedNode<E>, 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<E>,
) -> Result<StartedNode<E>, 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<E>,
) -> 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<dyn CleanupGuard> {
Box::new(LocalClusterCleanup {
owner: Arc::clone(&self.owner),
})
}
}
async fn start_all_nodes<E>(cluster: &ManualCluster<E>, 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<E> = ClusterHandle<E>;

4
testing-framework/core/src/scenario/control.rs
View File

@ -33,6 +33,10 @@ pub trait NodeControlHandle<E: Application>: 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<E::NodeClient> {
None
}

5
testing-framework/core/src/scenario/mod.rs
View File

@ -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,

398
testing-framework/core/src/scenario/provisioning.rs Normal file
View File

@ -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<E: Application> {
Managed {
deployment: E::Deployment,
external: Vec<ExternalNodeSource>,
},
Attached {
cluster: ExistingCluster,
external: Vec<ExternalNodeSource>,
},
External {
nodes: Vec<ExternalNodeSource>,
},
}
impl<E: Application> Clone for ClusterSource<E> {
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<E: Application> {
source: ClusterSource<E>,
policy: DeploymentPolicy,
start_mode: ClusterStartMode,
control: ClusterControlRequest,
}
impl<E: Application> Clone for ClusterRequest<E> {
fn clone(&self) -> Self {
Self {
source: self.source.clone(),
policy: self.policy,
start_mode: self.start_mode,
control: self.control,
}
}
}
impl<E: Application> ClusterRequest<E> {
#[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<ExternalNodeSource>) -> 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<ExternalNodeSource>) -> 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<E> {
&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<E: Application> {
deployment: Option<E::Deployment>,
node_clients: NodeClients<E>,
control_profile: ClusterControlProfile,
node_control: Option<Arc<dyn NodeControlHandle<E>>>,
cluster_wait: Option<Arc<dyn ClusterWaitHandle<E>>>,
cleanup: Option<Box<dyn CleanupGuard>>,
}
/// Backend-independent access to one managed, attached, or external cluster.
pub struct ClusterHandle<E: Application> {
deployment: Option<E::Deployment>,
node_clients: NodeClients<E>,
control_profile: ClusterControlProfile,
node_control: Option<Arc<dyn NodeControlHandle<E>>>,
cluster_wait: Option<Arc<dyn ClusterWaitHandle<E>>>,
}
impl<E: Application> Clone for ClusterHandle<E> {
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<E: Application> ClusterHandle<E> {
#[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<E> {
self.node_clients.clone()
}
#[must_use]
pub fn clients(&self) -> Vec<E::NodeClient> {
self.node_clients.snapshot()
}
#[must_use]
pub fn first_client(&self) -> Option<E::NodeClient> {
self.node_clients
.with_clients(|clients| clients.first().cloned())
}
#[must_use]
pub fn node_client(&self, name: &str) -> Option<E::NodeClient> {
self.node_control.as_ref()?.node_client(name)
}
#[must_use]
pub fn node_pid(&self, name: &str) -> Option<u32> {
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<StartedNode<E>, DynError> {
self.require_control()?.start_node(name).await
}
pub async fn start_node_with(
&self,
name: &str,
options: StartNodeOptions<E>,
) -> Result<StartedNode<E>, 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<E>,
) -> 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<dyn NodeControlHandle<E>>, 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<E: Application, H> {
handle: Option<H>,
unit: ClusterUnit<E>,
}
impl<E: Application, H> ProvisionedCluster<E, H> {
#[must_use]
pub const fn new(handle: Option<H>, unit: ClusterUnit<E>) -> Self {
Self { handle, unit }
}
#[must_use]
pub fn handle(&self) -> Option<&H> {
self.handle.as_ref()
}
#[must_use]
pub fn into_parts(self) -> (Option<H>, ClusterUnit<E>) {
(self.handle, self.unit)
}
}
/// Backend operation used by app deployment contexts to provision a cluster.
#[async_trait]
pub trait ClusterProvisioner<E: Application>: Clone + Send + Sync + 'static {
async fn provision_cluster(
&self,
request: ClusterRequest<E>,
) -> Result<ClusterUnit<E>, DynError>;
}
impl<E: Application> ClusterUnit<E> {
#[must_use]
pub fn new(
deployment: Option<E::Deployment>,
node_clients: NodeClients<E>,
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<dyn NodeControlHandle<E>>) -> Self {
self.node_control = Some(node_control);
self
}
#[must_use]
pub fn with_cluster_wait(mut self, cluster_wait: Arc<dyn ClusterWaitHandle<E>>) -> Self {
self.cluster_wait = Some(cluster_wait);
self
}
#[must_use]
pub fn with_cleanup(mut self, cleanup: Box<dyn CleanupGuard>) -> 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<E> {
&self.node_clients
}
#[must_use]
pub const fn control_profile(&self) -> ClusterControlProfile {
self.control_profile
}
#[must_use]
pub fn node_control(&self) -> Option<Arc<dyn NodeControlHandle<E>>> {
self.node_control.clone()
}
#[must_use]
pub fn cluster_wait(&self) -> Option<Arc<dyn ClusterWaitHandle<E>>> {
self.cluster_wait.clone()
}
pub fn take_cleanup(&mut self) -> Option<Box<dyn CleanupGuard>> {
self.cleanup.take()
}
#[must_use]
pub fn handle(&self) -> ClusterHandle<E> {
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(),
}
}
}

261
testing-framework/deployers/local/src/cluster.rs Normal file
View File

@ -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<E: LocalDeployerEnv> {
deployment: E::Deployment,
owner: Arc<LocalClusterOwner<E>>,
}
struct LocalClusterOwner<E: LocalDeployerEnv> {
nodes: NodeManager<E>,
closed: AtomicBool,
}
impl<E: LocalDeployerEnv> LocalClusterOwner<E> {
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<E: LocalDeployerEnv> Drop for LocalClusterOwner<E> {
fn drop(&mut self) {
self.close();
}
}
struct LocalClusterCleanup<E: LocalDeployerEnv> {
owner: Arc<LocalClusterOwner<E>>,
}
impl<E: LocalDeployerEnv> CleanupGuard for LocalClusterCleanup<E> {
fn cleanup(self: Box<Self>) {
self.owner.close();
}
}
impl<E: LocalDeployerEnv> Clone for LocalCluster<E> {
fn clone(&self) -> Self {
Self {
deployment: self.deployment.clone(),
owner: Arc::clone(&self.owner),
}
}
}
impl<E: LocalDeployerEnv> LocalCluster<E> {
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<Node<E>>) {
self.owner.nodes.initialize_with_nodes(nodes);
}
pub(crate) fn add_external_sources(
&self,
sources: impl IntoIterator<Item = ExternalNodeSource>,
) -> Result<(), DynError> {
for source in sources {
let client = E::external_node_client(&source)
.or_else(|_| build_external_client::<E>(&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<E> {
self.owner.nodes.node_clients()
}
#[must_use]
pub fn node_client(&self, name: &str) -> Option<E::NodeClient> {
self.owner.nodes.node_client(name)
}
#[must_use]
pub fn node_pid(&self, name: &str) -> Option<u32> {
self.owner.nodes.node_pid(name)
}
#[must_use]
pub fn clients(&self) -> Vec<E::NodeClient> {
self.node_clients().snapshot()
}
#[must_use]
pub fn first_client(&self) -> Option<E::NodeClient> {
self.node_clients()
.with_clients(|clients| clients.first().cloned())
}
pub async fn start_node(&self, name: &str) -> Result<StartedNode<E>, DynError> {
self.start_node_with(name, StartNodeOptions::default())
.await
}
pub async fn start_node_with(
&self,
name: &str,
options: StartNodeOptions<E>,
) -> Result<StartedNode<E>, 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<E>,
) -> 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<dyn CleanupGuard> {
Box::new(LocalClusterCleanup {
owner: Arc::clone(&self.owner),
})
}
}
#[async_trait]
impl<E: LocalDeployerEnv> NodeControlHandle<E> for LocalCluster<E> {
async fn restart_node(&self, name: &str) -> Result<(), DynError> {
self.restart_node(name).await
}
async fn restart_node_with(
&self,
name: &str,
options: StartNodeOptions<E>,
) -> Result<(), DynError> {
self.restart_node_with(name, options).await
}
async fn start_node(&self, name: &str) -> Result<StartedNode<E>, DynError> {
self.start_node(name).await
}
async fn start_node_with(
&self,
name: &str,
options: StartNodeOptions<E>,
) -> Result<StartedNode<E>, 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<E::NodeClient> {
self.node_client(name)
}
fn node_pid(&self, name: &str) -> Option<u32> {
self.node_pid(name)
}
}
#[async_trait]
impl<E: LocalDeployerEnv> ClusterWaitHandle<E> for LocalCluster<E> {
async fn wait_network_ready(&self) -> Result<(), DynError> {
self.wait_network_ready().await
}
}

344
testing-framework/deployers/local/src/deployer/orchestrator.rs
View File

@ -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<E: LocalDeployerEnv> {
nodes: Vec<Node<E>>,
}
impl<E: LocalDeployerEnv> LocalProcessGuard<E> {
fn new(nodes: Vec<Node<E>>) -> Self {
Self { nodes }
}
}
impl<E: LocalDeployerEnv> CleanupGuard for LocalProcessGuard<E> {
fn cleanup(self: Box<Self>) {
// Nodes own local processes; dropping them stops the processes.
drop(self.nodes);
}
}
/// Spawns nodes as local processes.
#[derive(Clone)]
pub struct ProcessDeployer<E: LocalDeployerEnv> {
@ -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<RetryAttemptError> 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<ScenarioError> for ProcessDeployerError {
fn from(value: ScenarioError) -> Self {
match value {
@ -142,6 +76,19 @@ impl From<ScenarioError> for ProcessDeployerError {
}
}
impl From<LocalClusterProvisionerError> 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<E: LocalDeployerEnv> Deployer<E, ()> for ProcessDeployer<E> {
type Error = ProcessDeployerError;
@ -189,24 +136,20 @@ impl<E: LocalDeployerEnv> ProcessDeployer<E> {
) -> Result<Runner<E>, 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::<E>::new(nodes.iter().map(|node| node.client()).collect());
let node_clients = merge_source_clients_for_local::<E>(&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<E: LocalDeployerEnv> ProcessDeployer<E> {
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<dyn CleanupGuard> = Box::new(LocalProcessGuard::<E>::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<E: LocalDeployerEnv> ProcessDeployer<E> {
) -> Result<Runner<E>, 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::<E>(&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<E: LocalDeployerEnv> ProcessDeployer<E> {
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<E, NodeControlCapability>,
nodes: Vec<Node<E>>,
) -> Arc<NodeManager<E>> {
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<Caps>(
scenario: &Scenario<E, Caps>,
membership_check: bool,
) -> Result<Vec<Node<E>>, 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<Vec<Node<E>>, 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::<E>(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<E: LocalDeployerEnv>(
source_plan: &SourceOrchestrationPlan,
node_clients: NodeClients<E>,
) -> Result<NodeClients<E>, DynError> {
for source in source_plan.external_sources() {
let client =
E::external_node_client(source).or_else(|_| build_external_client::<E>(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<E: Application, Caps>(
scenario: &Scenario<E, Caps>,
control: ClusterControlRequest,
) -> ClusterRequest<E> {
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<E: LocalDeployerEnv>(
descriptors: &E::Deployment,
execution: RetryExecutionConfig,
attempts: Arc<AtomicUsize>,
) -> Result<Vec<Node<E>>, RetryAttemptError> {
let attempt = attempts.fetch_add(1, Ordering::Relaxed) + 1;
let nodes = spawn_nodes_for_attempt::<E>(descriptors, execution.keep_tempdir).await?;
run_readiness_for_attempt::<E>(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<Item = Duration> {
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<E: LocalDeployerEnv>(
descriptors: &E::Deployment,
keep_tempdir: bool,
) -> Result<Vec<Node<E>>, RetryAttemptError> {
NodeManager::<E>::spawn_initial_nodes(descriptors, keep_tempdir)
.await
.map_err(|source| RetryAttemptError::Spawn {
source: source.into(),
})
}
async fn run_readiness_for_attempt<E: LocalDeployerEnv>(
attempt: usize,
nodes: Vec<Node<E>>,
execution: RetryExecutionConfig,
) -> Result<Vec<Node<E>>, RetryAttemptError> {
if !execution.readiness_enabled {
info!("skipping local readiness checks");
return Ok(nodes);
}
match wait_local_readiness::<E>(&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<AtomicUsize>,
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<E: LocalDeployerEnv> Default for ProcessDeployer<E> {
fn default() -> Self {
Self {
@ -483,18 +281,6 @@ impl<E: LocalDeployerEnv> Default for ProcessDeployer<E> {
}
}
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<E: Application>(
runtime_extensions: RuntimeExtensions,
runtime_cleanup: Option<Box<dyn CleanupGuard>>,
node_control: Option<Arc<dyn NodeControlHandle<E>>>,
cluster_wait: Option<Arc<dyn ClusterWaitHandle<E>>>,
) -> Result<RuntimeContext<E>, ProcessDeployerError> {
if node_clients.is_empty() && runtime_extensions.is_empty() {
return Err(ProcessDeployerError::RuntimePreflight);
@ -536,6 +323,9 @@ async fn run_context_for<E: Application>(
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 })
}

6
testing-framework/deployers/local/src/lib.rs
View File

@ -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";

100
testing-framework/deployers/local/src/manual/mod.rs
View File

@ -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<E: LocalDeployerEnv> {
nodes: NodeManager<E>,
cluster: LocalCluster<E>,
}
impl<E: LocalDeployerEnv> ManualCluster<E> {
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<E::NodeClient> {
self.nodes.node_client(name)
self.cluster.node_client(name)
}
#[must_use]
pub fn node_pid(&self, name: &str) -> Option<u32> {
self.nodes.node_pid(name)
self.cluster.node_pid(name)
}
pub async fn start_node(&self, name: &str) -> Result<StartedNode<E>, ManualClusterError> {
Ok(self
.nodes
.start_node_with(name, StartNodeOptions::<E>::default())
.await?)
Ok(self.cluster.start_node(name).await?)
}
pub async fn start_node_with(
@ -59,15 +47,15 @@ impl<E: LocalDeployerEnv> ManualCluster<E> {
name: &str,
options: StartNodeOptions<E>,
) -> Result<StartedNode<E>, 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<E: LocalDeployerEnv> ManualCluster<E> {
name: &str,
options: StartNodeOptions<E>,
) -> 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<E> {
self.nodes.node_clients()
self.cluster.node_clients()
}
pub fn add_external_sources(
&self,
external_sources: impl IntoIterator<Item = ExternalNodeSource>,
sources: impl IntoIterator<Item = ExternalNodeSource>,
) -> 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::<E>(&source))?;
node_clients.add_node(client);
}
Ok(())
self.cluster.add_external_sources(sources)
}
pub fn add_external_clients(&self, clients: impl IntoIterator<Item = E::NodeClient>) {
let node_clients = self.nodes.node_clients();
let node_clients = self.cluster.node_clients();
for client in clients {
node_clients.add_node(client);
}
}
}
impl<E: LocalDeployerEnv> Drop for ManualCluster<E> {
fn drop(&mut self) {
self.stop_all();
}
}
#[async_trait::async_trait]
impl<E: LocalDeployerEnv> NodeControlHandle<E> for ManualCluster<E> {
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<E: LocalDeployerEnv> NodeControlHandle<E> for ManualCluster<E> {
name: &str,
options: StartNodeOptions<E>,
) -> 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<StartedNode<E>, DynError> {
self.nodes
.start_node_with(name, StartNodeOptions::<E>::default())
.await
.map_err(|err| err.into())
self.cluster.start_node(name).await
}
async fn start_node_with(
@ -160,25 +125,22 @@ impl<E: LocalDeployerEnv> NodeControlHandle<E> for ManualCluster<E> {
name: &str,
options: StartNodeOptions<E>,
) -> Result<StartedNode<E>, 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<E::NodeClient> {
self.node_client(name)
self.cluster.node_client(name)
}
fn node_pid(&self, name: &str) -> Option<u32> {
self.node_pid(name)
self.cluster.node_pid(name)
}
}
#[async_trait::async_trait]
impl<E: LocalDeployerEnv> ClusterWaitHandle<E> for ManualCluster<E> {
async fn wait_network_ready(&self) -> Result<(), DynError> {
self.wait_network_ready().await.map_err(|err| err.into())
self.cluster.wait_network_ready().await
}
}

408
testing-framework/deployers/local/src/provisioner.rs Normal file
View File

@ -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<RetryAttemptError> 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<E> = ProvisionedCluster<E, LocalCluster<E>>;
impl LocalClusterProvisioner {
#[must_use]
pub fn manual_cluster<E: LocalDeployerEnv>(
&self,
deployment: E::Deployment,
) -> LocalCluster<E> {
LocalCluster::empty(deployment, keep_tempdir_from_env())
}
pub async fn provision<E: LocalDeployerEnv>(
&self,
request: ClusterRequest<E>,
membership_check: bool,
) -> Result<ProvisionedLocalCluster<E>, 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::<E>(
&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::<E>(&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<E: LocalDeployerEnv> ClusterProvisioner<E> for LocalClusterProvisioner {
async fn provision_cluster(
&self,
request: ClusterRequest<E>,
) -> Result<ClusterUnit<E>, DynError> {
let (_, unit) = self
.provision(request, true)
.await
.map_err(DynError::from)?
.into_parts();
Ok(unit)
}
}
async fn spawn_with_readiness_retry<E: LocalDeployerEnv>(
deployment: &E::Deployment,
membership_check: bool,
policy: DeploymentPolicy,
) -> Result<Vec<Node<E>>, 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::<E>(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<E: LocalDeployerEnv>(
deployment: &E::Deployment,
execution: RetryExecutionConfig,
attempts: Arc<AtomicUsize>,
) -> Result<Vec<Node<E>>, RetryAttemptError> {
let attempt = attempts.fetch_add(1, Ordering::Relaxed) + 1;
let nodes = NodeManager::<E>::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::<E>(&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<Self::NodeClient, DynError> {
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<String, u16>,
_options: &StartNodeOptions<Self>,
_peer_ports: &[u16],
) -> Result<BuiltNodeConfig<EmptyConfig>, DynError> {
unreachable!("empty deployment never builds node configs")
}
fn build_initial_node_configs(
_topology: &Self::Deployment,
) -> Result<Vec<NodeConfigEntry<EmptyConfig>>, ProcessSpawnError> {
Ok(Vec::new())
}
fn build_launch_spec(
_config: &EmptyConfig,
_dir: &Path,
_label: &str,
) -> Result<LaunchSpec, DynError> {
unreachable!("empty deployment never launches nodes")
}
fn node_endpoints(_config: &EmptyConfig) -> Result<NodeEndpoints, DynError> {
unreachable!("empty deployment has no endpoints")
}
fn node_client(_endpoints: &NodeEndpoints) -> Result<Self::NodeClient, DynError> {
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::<TestEnv>(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::<TestEnv>(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::<TestEnv>::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"
);
}
}