From 85b290422389be6d82ed4b13d2238fa113f49b3b Mon Sep 17 00:00:00 2001 From: andrussal Date: Fri, 17 Jul 2026 06:03:20 +0200 Subject: [PATCH] feat(tf): integrate heterogeneous local app stacks --- Cargo.lock | 25 ++ Cargo.toml | 1 + examples/README.md | 25 ++ examples/kvstore/examples/Cargo.toml | 6 + .../examples/src/bin/app_host_convergence.rs | 117 ++++++++ .../examples/src/bin/basic_convergence.rs | 6 +- .../kvstore/testing/integration/Cargo.toml | 1 + .../kvstore/testing/integration/src/app.rs | 106 ++++++- .../testing/integration/src/local_env.rs | 37 ++- .../testing/integration/src/scenario.rs | 9 +- examples/kvstore/testing/workloads/Cargo.toml | 1 + examples/kvstore/testing/workloads/src/lib.rs | 6 +- .../kvstore/testing/workloads/src/write.rs | 51 +++- examples/multi_app/README.md | 56 ++++ examples/multi_app/examples/Cargo.toml | 22 ++ .../examples/src/bin/typed_stack_smoke.rs | 269 ++++++++++++++++++ examples/openraft_kv/examples/Cargo.toml | 6 + .../examples/src/bin/app_host_smoke.rs | 118 ++++++++ examples/openraft_kv/examples/src/lib.rs | 44 +-- .../testing/integration/Cargo.toml | 1 + .../testing/integration/src/app.rs | 124 +++++++- .../testing/integration/src/local_env.rs | 43 ++- .../testing/integration/src/scenario.rs | 14 +- .../openraft_kv/testing/workloads/Cargo.toml | 1 + .../testing/workloads/src/handle_access.rs | 63 ++++ .../openraft_kv/testing/workloads/src/lib.rs | 3 + 26 files changed, 1116 insertions(+), 39 deletions(-) create mode 100644 examples/README.md create mode 100644 examples/kvstore/examples/src/bin/app_host_convergence.rs create mode 100644 examples/multi_app/README.md create mode 100644 examples/multi_app/examples/Cargo.toml create mode 100644 examples/multi_app/examples/src/bin/typed_stack_smoke.rs create mode 100644 examples/openraft_kv/examples/src/bin/app_host_smoke.rs create mode 100644 examples/openraft_kv/testing/workloads/src/handle_access.rs diff --git a/Cargo.lock b/Cargo.lock index 2f8c5eb..9dbccbd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1700,10 +1700,12 @@ name = "kvstore-examples" version = "0.1.0" dependencies = [ "anyhow", + "async-trait", "kvstore-node", "kvstore-runtime-ext", "kvstore-runtime-workloads", "serde", + "testing-framework-app", "testing-framework-core", "testing-framework-runner-compose", "testing-framework-runner-k8s", @@ -1735,6 +1737,7 @@ dependencies = [ "async-trait", "kvstore-node", "serde", + "testing-framework-app", "testing-framework-core", "testing-framework-runner-compose", "testing-framework-runner-k8s", @@ -1749,6 +1752,7 @@ dependencies = [ "kvstore-node", "kvstore-runtime-ext", "serde", + "testing-framework-app", "testing-framework-core", "tokio", "tracing", @@ -1920,6 +1924,23 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "multi-app-examples" +version = "0.1.0" +dependencies = [ + "anyhow", + "async-trait", + "kvstore-runtime-ext", + "openraft-kv-runtime-ext", + "openraft-kv-runtime-workloads", + "serde", + "testing-framework-app", + "testing-framework-core", + "tokio", + "tracing", + "tracing-subscriber", +] + [[package]] name = "native-tls" version = "0.2.14" @@ -2085,9 +2106,11 @@ name = "openraft-kv-examples" version = "0.1.0" dependencies = [ "anyhow", + "async-trait", "openraft-kv-node", "openraft-kv-runtime-ext", "openraft-kv-runtime-workloads", + "testing-framework-app", "testing-framework-core", "testing-framework-runner-k8s", "tokio", @@ -2120,6 +2143,7 @@ dependencies = [ "async-trait", "openraft-kv-node", "reqwest", + "testing-framework-app", "testing-framework-core", "testing-framework-runner-compose", "testing-framework-runner-k8s", @@ -2134,6 +2158,7 @@ dependencies = [ "async-trait", "openraft-kv-node", "openraft-kv-runtime-ext", + "testing-framework-app", "testing-framework-core", "thiserror 2.0.18", "tokio", diff --git a/Cargo.toml b/Cargo.toml index 6f56b42..05d8355 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,7 @@ members = [ "examples/metrics_counter/metrics-counter-node", "examples/metrics_counter/testing/integration", "examples/metrics_counter/testing/workloads", + "examples/multi_app/examples", "examples/nats/examples", "examples/nats/testing/integration", "examples/nats/testing/workloads", diff --git a/examples/README.md b/examples/README.md new file mode 100644 index 0000000..c751212 --- /dev/null +++ b/examples/README.md @@ -0,0 +1,25 @@ +# Examples + +Use the app layer first when the test is about an application or composed +system interface. + +Canonical app-layer examples: + +- `kvstore_app_host_convergence`: one local app cluster exposed through + `AppHost` +- `openraft_kv_app_host_smoke`: one richer local app cluster exposed through + `AppHost` +- `multi_app_typed_stack_smoke`: composed app stack exposing typed child and + stack handles + +The older direct `ScenarioBuilder` examples are still useful for +backend-specific coverage: + +- local uniform cluster behavior +- compose runner behavior +- k8s runner behavior +- manual cluster control + +Do not use those older examples as the pattern for new composed systems. New +multi-app tests should define an `AppDeployment`, expose typed handles, and run +through `AppHost::scenario().with_app(...)`. diff --git a/examples/kvstore/examples/Cargo.toml b/examples/kvstore/examples/Cargo.toml index 79bc82b..3580d0f 100644 --- a/examples/kvstore/examples/Cargo.toml +++ b/examples/kvstore/examples/Cargo.toml @@ -4,6 +4,10 @@ license.workspace = true name = "kvstore-examples" version.workspace = true +[[bin]] +name = "kvstore_app_host_convergence" +path = "src/bin/app_host_convergence.rs" + [[bin]] name = "kvstore_basic_convergence" path = "src/bin/basic_convergence.rs" @@ -21,9 +25,11 @@ name = "kvstore_k8s_manual_convergence" path = "src/bin/k8s_manual_convergence.rs" [dependencies] +async-trait = { workspace = true } kvstore-node = { path = "../kvstore-node" } kvstore-runtime-ext = { path = "../testing/integration" } kvstore-runtime-workloads = { path = "../testing/workloads" } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } testing-framework-runner-compose = { workspace = true } testing-framework-runner-k8s = { workspace = true } diff --git a/examples/kvstore/examples/src/bin/app_host_convergence.rs b/examples/kvstore/examples/src/bin/app_host_convergence.rs new file mode 100644 index 0000000..c454956 --- /dev/null +++ b/examples/kvstore/examples/src/bin/app_host_convergence.rs @@ -0,0 +1,117 @@ +use std::time::Duration; + +use async_trait::async_trait; +use kvstore_runtime_ext::{KvEnv, KvLocalApp}; +use serde::{Deserialize, Serialize}; +use testing_framework_app::{ + AppHost, AppHostEnv, AppHostLocalDeployer, AppRunContextExt, AppScenarioBuilderExt, + LocalAppCluster, +}; +use testing_framework_core::scenario::{Deployer, DynError, RunContext, Workload}; +use tracing::info; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + let mut scenario = AppHost::scenario() + .with_app(KvLocalApp::nodes(3)) + .with_run_duration(Duration::from_secs(5)) + .with_workload(KvAppHostConvergence::new(3)) + .build()?; + + let deployer = AppHostLocalDeployer::default(); + let runner = deployer.deploy(&scenario).await?; + runner.run(&mut scenario).await?; + + Ok(()) +} + +#[derive(Clone)] +struct KvAppHostConvergence { + expected_nodes: usize, +} + +impl KvAppHostConvergence { + const fn new(expected_nodes: usize) -> Self { + Self { expected_nodes } + } +} + +#[async_trait] +impl Workload for KvAppHostConvergence { + fn name(&self) -> &str { + "kv_app_host_convergence" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let cluster = ctx.require_app::>()?; + + ensure_cluster_shape(&cluster, self.expected_nodes)?; + put_value(&cluster, "before-restart").await?; + cluster.restart_node("node-0").await?; + cluster.wait_node_ready("node-0").await?; + put_value(&cluster, "after-restart").await?; + + info!( + nodes = self.expected_nodes, + "kv app host cluster exposes app-local node control and clients" + ); + + Ok(()) + } +} + +fn ensure_cluster_shape( + cluster: &LocalAppCluster, + expected_nodes: usize, +) -> Result<(), DynError> { + if cluster.node_count() != expected_nodes || cluster.clients().len() != expected_nodes { + return Err(format!("kv app host expected {expected_nodes} nodes").into()); + } + + if cluster.node_client("node-0").is_none() { + return Err("kv app host cannot access node-0 client".into()); + } + + if cluster.node_pid("node-0").is_none() { + return Err("kv app host cannot access node-0 process id".into()); + } + + Ok(()) +} + +async fn put_value(cluster: &LocalAppCluster, value: &str) -> Result<(), DynError> { + let Some(client) = cluster.first_client() else { + return Err("kv app host has no clients".into()); + }; + + let response: KvPutResponse = client + .put( + "/kv/app-host-smoke", + &KvPutRequest { + value: value.to_owned(), + expected_version: None, + }, + ) + .await?; + + if !response.applied { + return Err(format!("kv app host write was rejected for value {value}").into()); + } + + Ok(()) +} + +#[derive(Serialize)] +struct KvPutRequest { + value: String, + expected_version: Option, +} + +#[derive(Deserialize)] +struct KvPutResponse { + applied: bool, +} diff --git a/examples/kvstore/examples/src/bin/basic_convergence.rs b/examples/kvstore/examples/src/bin/basic_convergence.rs index 6ebc130..e9f8041 100644 --- a/examples/kvstore/examples/src/bin/basic_convergence.rs +++ b/examples/kvstore/examples/src/bin/basic_convergence.rs @@ -2,7 +2,8 @@ use std::time::Duration; use kvstore_runtime_ext::KvLocalDeployer; use kvstore_runtime_workloads::{ - KvBuilderExt, KvConverges, KvScenarioBuilder, KvTopology, KvWriteWorkload, + KvBuilderExt, KvClusterAccessible, KvConverges, KvExistingClusterApp, KvScenarioBuilder, + KvWriteWorkload, }; use testing_framework_core::scenario::Deployer; @@ -12,8 +13,9 @@ async fn main() -> anyhow::Result<()> { .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .init(); - let mut scenario = KvScenarioBuilder::deployment_with(|_| KvTopology::new(3)) + let mut scenario = KvScenarioBuilder::with_existing_kvstore_app(KvExistingClusterApp::nodes(3)) .with_run_duration(Duration::from_secs(30)) + .with_workload(KvClusterAccessible::new(3)) .with_workload( KvWriteWorkload::new() .operations(300) diff --git a/examples/kvstore/testing/integration/Cargo.toml b/examples/kvstore/testing/integration/Cargo.toml index 9773cac..d555f8b 100644 --- a/examples/kvstore/testing/integration/Cargo.toml +++ b/examples/kvstore/testing/integration/Cargo.toml @@ -5,6 +5,7 @@ name = "kvstore-runtime-ext" version.workspace = true [dependencies] +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } testing-framework-runner-compose = { workspace = true } testing-framework-runner-k8s = { workspace = true } diff --git a/examples/kvstore/testing/integration/src/app.rs b/examples/kvstore/testing/integration/src/app.rs index bd0dffe..ebbd04e 100644 --- a/examples/kvstore/testing/integration/src/app.rs +++ b/examples/kvstore/testing/integration/src/app.rs @@ -3,9 +3,13 @@ use std::io::Error; use async_trait::async_trait; use kvstore_node::KvHttpClient; use serde::{Deserialize, Serialize}; -use testing_framework_core::scenario::{ - Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError, - NodeAccess, serialize_cluster_yaml_config, +use testing_framework_app::{AppDeployment, AppHostEnv, DeployContext, LocalAppCluster}; +use testing_framework_core::{ + scenario::{ + Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError, + NodeAccess, NodeClients, serialize_cluster_yaml_config, + }, + topology::DeploymentDescriptor, }; pub type KvTopology = testing_framework_core::topology::ClusterTopology; @@ -40,6 +44,102 @@ impl Application for KvEnv { } } +#[derive(Clone)] +pub struct KvStoreCluster { + deployment: KvTopology, + node_clients: NodeClients, +} + +impl KvStoreCluster { + #[must_use] + pub const fn new(deployment: KvTopology, node_clients: NodeClients) -> Self { + Self { + deployment, + node_clients, + } + } + + #[must_use] + pub fn topology(&self) -> &KvTopology { + &self.deployment + } + + #[must_use] + pub fn node_count(&self) -> usize { + self.deployment.node_count() + } + + #[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()) + } +} + +#[derive(Clone)] +pub struct KvExistingClusterApp { + topology: KvTopology, +} + +impl KvExistingClusterApp { + #[must_use] + pub fn nodes(nodes: usize) -> Self { + Self { + topology: KvTopology::new(nodes), + } + } + + #[must_use] + pub fn topology(&self) -> KvTopology { + self.topology.clone() + } +} + +#[async_trait] +impl AppDeployment for KvExistingClusterApp { + type Handle = KvStoreCluster; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + Ok(KvStoreCluster::new( + ctx.deployment().clone(), + ctx.node_clients().clone(), + )) + } +} + +#[derive(Clone)] +pub struct KvLocalApp { + deployment: KvTopology, +} + +impl KvLocalApp { + #[must_use] + pub fn nodes(nodes: usize) -> Self { + Self { + deployment: KvTopology::new(nodes), + } + } + + #[must_use] + pub fn deployment(&self) -> KvTopology { + self.deployment.clone() + } +} + +#[async_trait] +impl AppDeployment for KvLocalApp { + type Handle = LocalAppCluster; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + ctx.deploy_local_cluster::(self.deployment).await + } +} + impl ClusterNodeConfigApplication for KvEnv { type ConfigError = Error; diff --git a/examples/kvstore/testing/integration/src/local_env.rs b/examples/kvstore/testing/integration/src/local_env.rs index 82e8778..e643d3f 100644 --- a/examples/kvstore/testing/integration/src/local_env.rs +++ b/examples/kvstore/testing/integration/src/local_env.rs @@ -1,8 +1,9 @@ -use std::collections::HashMap; +use std::{collections::HashMap, path::PathBuf, sync::Arc}; use testing_framework_core::scenario::{DynError, StartNodeOptions}; use testing_framework_runner_local::{ - LocalBinaryApp, LocalNodePorts, LocalPeerNode, LocalProcessSpec, + BinaryProviderRef, BuildBinaryProvider, BuildCommand, EnvBinaryProvider, + FallbackBinaryProvider, LocalBinaryApp, LocalNodePorts, LocalPeerNode, LocalProcessSpec, build_local_cluster_node_config, yaml_node_config, }; @@ -28,7 +29,9 @@ impl LocalBinaryApp for KvEnv { } fn local_process_spec() -> LocalProcessSpec { - LocalProcessSpec::new("KVSTORE_NODE_BIN").with_rust_log("kvstore_node=info") + LocalProcessSpec::new("KVSTORE_NODE_BIN") + .with_binary_provider(kvstore_binary_provider()) + .with_rust_log("kvstore_node=info") } fn render_local_config(config: &KvNodeConfig) -> Result, DynError> { @@ -39,3 +42,31 @@ impl LocalBinaryApp for KvEnv { config.http_port } } + +fn kvstore_binary_provider() -> FallbackBinaryProvider { + let workspace = workspace_root(); + let providers: [BinaryProviderRef; 2] = [ + Arc::new(EnvBinaryProvider::new("KVSTORE_NODE_BIN")), + Arc::new(BuildBinaryProvider { + command: BuildCommand::new("cargo").with_args([ + "build", + "-p", + "kvstore-node", + "--bin", + "kvstore-node", + ]), + output_path: PathBuf::from(format!( + "target/debug/kvstore-node{}", + std::env::consts::EXE_SUFFIX + )), + working_dir: Some(workspace), + lock_dir: None, + }), + ]; + + FallbackBinaryProvider::new(providers) +} + +fn workspace_root() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../..") +} diff --git a/examples/kvstore/testing/integration/src/scenario.rs b/examples/kvstore/testing/integration/src/scenario.rs index 5a0e770..39a7e1b 100644 --- a/examples/kvstore/testing/integration/src/scenario.rs +++ b/examples/kvstore/testing/integration/src/scenario.rs @@ -1,15 +1,22 @@ +use testing_framework_app::AppScenarioBuilderExt; use testing_framework_core::scenario::ScenarioBuilder; -use crate::{KvEnv, KvTopology}; +use crate::{KvEnv, KvExistingClusterApp, KvTopology}; pub type KvScenarioBuilder = ScenarioBuilder; pub trait KvBuilderExt: Sized { fn deployment_with(f: impl FnOnce(KvTopology) -> KvTopology) -> Self; + + fn with_existing_kvstore_app(app: KvExistingClusterApp) -> Self; } impl KvBuilderExt for KvScenarioBuilder { fn deployment_with(f: impl FnOnce(KvTopology) -> KvTopology) -> Self { KvScenarioBuilder::with_deployment(f(KvTopology::new(3))) } + + fn with_existing_kvstore_app(app: KvExistingClusterApp) -> Self { + KvScenarioBuilder::with_deployment(app.topology()).with_app(app) + } } diff --git a/examples/kvstore/testing/workloads/Cargo.toml b/examples/kvstore/testing/workloads/Cargo.toml index e400c2a..14960d6 100644 --- a/examples/kvstore/testing/workloads/Cargo.toml +++ b/examples/kvstore/testing/workloads/Cargo.toml @@ -7,6 +7,7 @@ version.workspace = true [dependencies] kvstore-node = { path = "../../kvstore-node" } kvstore-runtime-ext = { path = "../integration" } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } async-trait = { workspace = true } diff --git a/examples/kvstore/testing/workloads/src/lib.rs b/examples/kvstore/testing/workloads/src/lib.rs index 23ac375..618afff 100644 --- a/examples/kvstore/testing/workloads/src/lib.rs +++ b/examples/kvstore/testing/workloads/src/lib.rs @@ -2,5 +2,7 @@ mod expectations; mod write; pub use expectations::KvConverges; -pub use kvstore_runtime_ext::{KvBuilderExt, KvEnv, KvScenarioBuilder, KvTopology}; -pub use write::KvWriteWorkload; +pub use kvstore_runtime_ext::{ + KvBuilderExt, KvEnv, KvExistingClusterApp, KvScenarioBuilder, KvStoreCluster, KvTopology, +}; +pub use write::{KvClusterAccessible, KvWriteWorkload}; diff --git a/examples/kvstore/testing/workloads/src/write.rs b/examples/kvstore/testing/workloads/src/write.rs index ec65a05..073a21b 100644 --- a/examples/kvstore/testing/workloads/src/write.rs +++ b/examples/kvstore/testing/workloads/src/write.rs @@ -1,8 +1,9 @@ use std::time::Duration; use async_trait::async_trait; -use kvstore_runtime_ext::KvEnv; +use kvstore_runtime_ext::{KvEnv, KvStoreCluster}; use serde::{Deserialize, Serialize}; +use testing_framework_app::AppRunContextExt; use testing_framework_core::scenario::{DynError, RunContext, Workload}; use tracing::info; @@ -68,6 +69,18 @@ impl Default for KvWriteWorkload { } } +#[derive(Clone)] +pub struct KvClusterAccessible { + expected_nodes: usize, +} + +impl KvClusterAccessible { + #[must_use] + pub const fn new(expected_nodes: usize) -> Self { + Self { expected_nodes } + } +} + #[async_trait] impl Workload for KvWriteWorkload { fn name(&self) -> &str { @@ -126,6 +139,42 @@ impl Workload for KvWriteWorkload { } } +#[async_trait] +impl Workload for KvClusterAccessible { + fn name(&self) -> &str { + "kv_cluster_accessible" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let cluster = ctx.require_app::()?; + let client_count = cluster.clients().len(); + + if cluster.node_count() != self.expected_nodes { + return Err(format!( + "kv app topology has {} nodes, expected {}", + cluster.node_count(), + self.expected_nodes + ) + .into()); + } + + if client_count != self.expected_nodes { + return Err(format!( + "kv app handle has {client_count} clients, expected {}", + self.expected_nodes + ) + .into()); + } + + info!( + nodes = self.expected_nodes, + "kv app handle is accessible from workload" + ); + + Ok(()) + } +} + fn compute_interval(rate_per_sec: usize) -> Option { if rate_per_sec == 0 { return None; diff --git a/examples/multi_app/README.md b/examples/multi_app/README.md new file mode 100644 index 0000000..abec9bb --- /dev/null +++ b/examples/multi_app/README.md @@ -0,0 +1,56 @@ +# Multi-App Examples + +This directory shows the canonical app-layer pattern for composed systems. + +Use this shape when a scenario needs several apps or resources that should feel +like one system to the workload: + +```rust +let mut scenario = AppHost::scenario() + .with_app(ExampleStackApp::new()) + .with_workload(ExampleStackWorkload::new()) + .build()?; +``` + +The stack app deploys child apps, exposes their typed handles, and returns a +composed stack handle: + +```rust +let store = StoreHandle::new(ctx.deploy_and_expose(self.store).await?); +let consensus = ConsensusHandle::new(ctx.deploy_and_expose(self.consensus).await?); +let wallet = WalletHandle::new(store.clone(), consensus.clone()); +let stack = ExampleStackHandle::new(store.clone(), consensus.clone(), wallet.clone()); + +ctx.expose(store)?; +ctx.expose(consensus)?; +ctx.expose(wallet)?; +ctx.expose(stack.clone())?; +``` + +Workloads then request concrete handles: + +```rust +let stack = ctx.require_app::()?; +let wallet = ctx.require_app::()?; +``` + +Resource lifecycle comes from the TF adapter used by a child deployment: + +- `LocalProcessApp` manages one local binary process. +- `LocalAppCluster` manages a uniform local cluster. + +Attached and external sources are not registered again at the app layer. The +outer scenario resolves them through its existing source providers, and the app +deployment receives the resulting node clients and control profile through +`DeployContext`. When the active deployer provides node control or cluster +readiness, the same handles are available through `DeployContext::node_control` +and `DeployContext::cluster_wait`; the app layer does not create replacements. + +Managed adapters register cleanup during deployment. Cleanup runs in reverse +acquisition order and does not depend on the last handle clone being dropped. +An app-specific deployment only describes composition; it does not implement a +second lifecycle interface. + +For a single uniform cluster, the core `ScenarioBuilder` flow remains +valid. For composed systems, prefer this app-layer shape instead of building a +fake outer cluster or adding app-specific code to TF. diff --git a/examples/multi_app/examples/Cargo.toml b/examples/multi_app/examples/Cargo.toml new file mode 100644 index 0000000..bd40515 --- /dev/null +++ b/examples/multi_app/examples/Cargo.toml @@ -0,0 +1,22 @@ +[package] +edition.workspace = true +license.workspace = true +name = "multi-app-examples" +version.workspace = true + +[[bin]] +name = "multi_app_typed_stack_smoke" +path = "src/bin/typed_stack_smoke.rs" + +[dependencies] +anyhow = "1.0" +async-trait = { workspace = true } +kvstore-runtime-ext = { path = "../../kvstore/testing/integration" } +openraft-kv-runtime-ext = { path = "../../openraft_kv/testing/integration" } +openraft-kv-runtime-workloads = { path = "../../openraft_kv/testing/workloads" } +serde = { workspace = true } +testing-framework-app = { workspace = true } +testing-framework-core = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tracing = { workspace = true } +tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/examples/multi_app/examples/src/bin/typed_stack_smoke.rs b/examples/multi_app/examples/src/bin/typed_stack_smoke.rs new file mode 100644 index 0000000..a857b0b --- /dev/null +++ b/examples/multi_app/examples/src/bin/typed_stack_smoke.rs @@ -0,0 +1,269 @@ +use std::time::Duration; + +use async_trait::async_trait; +use kvstore_runtime_ext::{KvEnv, KvLocalApp}; +use openraft_kv_runtime_ext::{OpenRaftKvEnv, OpenRaftKvLocalApp}; +use openraft_kv_runtime_workloads::{ + OpenRaftMembership, ensure_cluster_size, expected_kv, resolve_client_for_node, wait_for_leader, + wait_for_membership, wait_for_replication, write_batch, +}; +use serde::{Deserialize, Serialize}; +use testing_framework_app::{ + AppDeployment, AppHost, AppHostEnv, AppHostLocalDeployer, AppRunContextExt, + AppScenarioBuilderExt, DeployContext, LocalAppCluster, +}; +use testing_framework_core::scenario::{Deployer, DynError, RunContext, Workload}; +use tracing::info; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + let mut scenario = AppHost::scenario() + .with_app(ExampleStackApp::new()) + .with_run_duration(Duration::from_secs(5)) + .with_workload(ExampleStackWorkload::new()) + .build()?; + + let deployer = AppHostLocalDeployer::default(); + let runner = deployer.deploy(&scenario).await?; + runner.run(&mut scenario).await?; + + Ok(()) +} + +#[derive(Clone)] +struct ExampleStackApp { + store: KvLocalApp, + consensus: OpenRaftKvLocalApp, +} + +impl ExampleStackApp { + fn new() -> Self { + Self { + store: KvLocalApp::nodes(2), + consensus: OpenRaftKvLocalApp::nodes(3), + } + } +} + +#[async_trait] +impl AppDeployment for ExampleStackApp { + type Handle = ExampleStackHandle; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + let store = StoreHandle::new(ctx.deploy_and_expose(self.store).await?); + let consensus = ConsensusHandle::new(ctx.deploy_and_expose(self.consensus).await?); + let wallet = WalletHandle::new(store.clone(), consensus.clone()); + let stack = ExampleStackHandle::new(store.clone(), consensus.clone(), wallet.clone()); + + ctx.expose(store)?; + ctx.expose(consensus)?; + ctx.expose(wallet)?; + ctx.expose(stack.clone())?; + + Ok(stack) + } +} + +#[derive(Clone)] +struct ExampleStackHandle { + store: StoreHandle, + consensus: ConsensusHandle, + wallet: WalletHandle, +} + +impl ExampleStackHandle { + const fn new(store: StoreHandle, consensus: ConsensusHandle, wallet: WalletHandle) -> Self { + Self { + store, + consensus, + wallet, + } + } + + const fn store(&self) -> &StoreHandle { + &self.store + } + + const fn consensus(&self) -> &ConsensusHandle { + &self.consensus + } + + const fn wallet(&self) -> &WalletHandle { + &self.wallet + } +} + +#[derive(Clone)] +struct StoreHandle { + cluster: LocalAppCluster, +} + +impl StoreHandle { + const fn new(cluster: LocalAppCluster) -> Self { + Self { cluster } + } + + fn node_count(&self) -> usize { + self.cluster.node_count() + } + + async fn put(&self, key: &str, value: &str) -> Result<(), DynError> { + let Some(client) = self.cluster.first_client() else { + return Err("store handle has no kv clients".into()); + }; + + let response: KvPutResponse = client + .put( + key, + &KvPutRequest { + value: value.to_owned(), + expected_version: None, + }, + ) + .await?; + + if !response.applied { + return Err(format!("store write for {key} was rejected").into()); + } + + Ok(()) + } +} + +#[derive(Clone)] +struct ConsensusHandle { + cluster: LocalAppCluster, +} + +impl ConsensusHandle { + const fn new(cluster: LocalAppCluster) -> Self { + Self { cluster } + } + + fn node_count(&self) -> usize { + self.cluster.node_count() + } + + async fn bootstrap_and_write(&self, prefix: &str, writes: usize) -> Result<(), DynError> { + let clients = self.cluster.clients(); + + ensure_cluster_size(&clients, self.node_count())?; + clients[0].init_self().await?; + + let leader_id = wait_for_leader(&clients, Duration::from_secs(30), None).await?; + let membership = OpenRaftMembership::discover(&clients).await?; + let leader = resolve_client_for_node(&clients, leader_id, Duration::from_secs(30)).await?; + + for learner in membership.learner_targets(leader_id) { + leader + .add_learner(learner.node_id, &learner.public_addr) + .await?; + } + + let voter_ids = membership.voter_ids(); + leader.change_membership(voter_ids.iter().copied()).await?; + wait_for_membership(&clients, &voter_ids, Duration::from_secs(30)).await?; + + write_batch(&leader, prefix, 0, writes).await?; + wait_for_replication( + &clients, + &expected_kv(prefix, writes), + Duration::from_secs(30), + ) + .await?; + + Ok(()) + } +} + +#[derive(Clone)] +struct WalletHandle { + store: StoreHandle, + consensus: ConsensusHandle, +} + +impl WalletHandle { + const fn new(store: StoreHandle, consensus: ConsensusHandle) -> Self { + Self { store, consensus } + } + + async fn submit_transfer(&self, id: &str) -> Result<(), DynError> { + self.store.put("/kv/wallet-transfer", id).await?; + self.consensus + .bootstrap_and_write("wallet-transfer", 3) + .await?; + + Ok(()) + } +} + +#[derive(Clone)] +struct ExampleStackWorkload; + +impl ExampleStackWorkload { + const fn new() -> Self { + Self + } +} + +#[async_trait] +impl Workload for ExampleStackWorkload { + fn name(&self) -> &str { + "multi_app_typed_stack_smoke" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let stack = ctx.require_app::()?; + let store = ctx.require_app::()?; + let consensus = ctx.require_app::()?; + let wallet = ctx.require_app::()?; + + ensure_stack_handles_match(&stack, &store, &consensus, &wallet)?; + store.put("/kv/typed-stack-smoke", "store-ready").await?; + wallet.submit_transfer("transfer-1").await?; + + info!( + store_nodes = store.node_count(), + consensus_nodes = consensus.node_count(), + "typed multi-app stack handles are available to workloads" + ); + + Ok(()) + } +} + +fn ensure_stack_handles_match( + stack: &ExampleStackHandle, + store: &StoreHandle, + consensus: &ConsensusHandle, + wallet: &WalletHandle, +) -> Result<(), DynError> { + if stack.store().node_count() != store.node_count() { + return Err("stack store handle does not match exposed store handle".into()); + } + + if stack.consensus().node_count() != consensus.node_count() { + return Err("stack consensus handle does not match exposed consensus handle".into()); + } + + if stack.wallet().store.node_count() != wallet.store.node_count() { + return Err("stack wallet handle does not match exposed wallet handle".into()); + } + + Ok(()) +} + +#[derive(Serialize)] +struct KvPutRequest { + value: String, + expected_version: Option, +} + +#[derive(Deserialize)] +struct KvPutResponse { + applied: bool, +} diff --git a/examples/openraft_kv/examples/Cargo.toml b/examples/openraft_kv/examples/Cargo.toml index b514dc6..5b9e46b 100644 --- a/examples/openraft_kv/examples/Cargo.toml +++ b/examples/openraft_kv/examples/Cargo.toml @@ -4,6 +4,10 @@ license.workspace = true name = "openraft-kv-examples" version.workspace = true +[[bin]] +name = "openraft_kv_app_host_smoke" +path = "src/bin/app_host_smoke.rs" + [[bin]] name = "openraft_kv_basic_failover" path = "src/bin/basic_failover.rs" @@ -18,9 +22,11 @@ path = "src/bin/k8s_failover.rs" [dependencies] anyhow = "1.0" +async-trait = { workspace = true } openraft-kv-node = { path = "../openraft-kv-node" } openraft-kv-runtime-ext = { path = "../testing/integration" } openraft-kv-runtime-workloads = { path = "../testing/workloads" } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } testing-framework-runner-k8s = { workspace = true } tokio = { workspace = true, features = ["full"] } diff --git a/examples/openraft_kv/examples/src/bin/app_host_smoke.rs b/examples/openraft_kv/examples/src/bin/app_host_smoke.rs new file mode 100644 index 0000000..94d97e6 --- /dev/null +++ b/examples/openraft_kv/examples/src/bin/app_host_smoke.rs @@ -0,0 +1,118 @@ +use std::time::Duration; + +use async_trait::async_trait; +use openraft_kv_runtime_ext::{OpenRaftKvEnv, OpenRaftKvLocalApp}; +use openraft_kv_runtime_workloads::{ + OpenRaftMembership, ensure_cluster_size, expected_kv, resolve_client_for_node, wait_for_leader, + wait_for_membership, wait_for_replication, write_batch, +}; +use testing_framework_app::{ + AppHost, AppHostEnv, AppHostLocalDeployer, AppRunContextExt, AppScenarioBuilderExt, + LocalAppCluster, +}; +use testing_framework_core::scenario::{Deployer, DynError, RunContext, Workload}; +use tracing::info; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + + let mut scenario = AppHost::scenario() + .with_app(OpenRaftKvLocalApp::nodes(3)) + .with_run_duration(Duration::from_secs(5)) + .with_workload(OpenRaftKvAppHostSmoke::new(3)) + .build()?; + + let deployer = AppHostLocalDeployer::default(); + let runner = deployer.deploy(&scenario).await?; + runner.run(&mut scenario).await?; + + Ok(()) +} + +#[derive(Clone)] +struct OpenRaftKvAppHostSmoke { + expected_nodes: usize, + timeout: Duration, +} + +impl OpenRaftKvAppHostSmoke { + const fn new(expected_nodes: usize) -> Self { + Self { + expected_nodes, + timeout: Duration::from_secs(30), + } + } +} + +#[async_trait] +impl Workload for OpenRaftKvAppHostSmoke { + fn name(&self) -> &str { + "openraft_kv_app_host_smoke" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let cluster = ctx.require_app::>()?; + + ensure_cluster_shape(&cluster, self.expected_nodes)?; + bootstrap_voter_cluster(&cluster, self.timeout).await?; + + info!( + nodes = self.expected_nodes, + "openraft app host cluster exposes app-local clients and process access" + ); + + Ok(()) + } +} + +fn ensure_cluster_shape( + cluster: &LocalAppCluster, + expected_nodes: usize, +) -> Result<(), DynError> { + if cluster.node_count() != expected_nodes { + return Err(format!("openraft app host expected {expected_nodes} nodes").into()); + } + + if cluster.node_client("node-0").is_none() { + return Err("openraft app host cannot access node-0 client".into()); + } + + if cluster.node_pid("node-0").is_none() { + return Err("openraft app host cannot access node-0 process id".into()); + } + + ensure_cluster_size(&cluster.clients(), expected_nodes)?; + + Ok(()) +} + +async fn bootstrap_voter_cluster( + cluster: &LocalAppCluster, + timeout: Duration, +) -> Result<(), DynError> { + let clients = cluster.clients(); + + clients[0].init_self().await?; + + let initial_leader = wait_for_leader(&clients, timeout, None).await?; + let membership = OpenRaftMembership::discover(&clients).await?; + let leader = resolve_client_for_node(&clients, initial_leader, timeout).await?; + + for learner in membership.learner_targets(initial_leader) { + leader + .add_learner(learner.node_id, &learner.public_addr) + .await?; + } + + let voter_ids = membership.voter_ids(); + leader.change_membership(voter_ids.iter().copied()).await?; + wait_for_membership(&clients, &voter_ids, timeout).await?; + + write_batch(&leader, "app-host-raft-key", 0, 3).await?; + wait_for_replication(&clients, &expected_kv("app-host-raft-key", 3), timeout).await?; + + Ok(()) +} diff --git a/examples/openraft_kv/examples/src/lib.rs b/examples/openraft_kv/examples/src/lib.rs index 8e19a67..1101c38 100644 --- a/examples/openraft_kv/examples/src/lib.rs +++ b/examples/openraft_kv/examples/src/lib.rs @@ -1,7 +1,11 @@ use std::time::Duration; -use openraft_kv_runtime_ext::{OpenRaftKvBuilderExt, OpenRaftKvEnv, OpenRaftKvScenarioBuilder}; -use openraft_kv_runtime_workloads::{OpenRaftKvConverges, OpenRaftKvFailoverWorkload}; +use openraft_kv_runtime_ext::{ + OpenRaftKvBuilderExt, OpenRaftKvEnv, OpenRaftKvExistingClusterApp, OpenRaftKvScenarioBuilder, +}; +use openraft_kv_runtime_workloads::{ + OpenRaftKvClusterAccessible, OpenRaftKvConverges, OpenRaftKvFailoverWorkload, +}; use testing_framework_core::scenario::{NodeControlCapability, Scenario}; /// Number of writes issued before the leader restart. @@ -19,23 +23,23 @@ pub fn build_failover_scenario( run_duration: Duration, workload_timeout: Duration, ) -> anyhow::Result> { - Ok( - OpenRaftKvScenarioBuilder::deployment_with(|deployment| deployment) - .with_cluster_observer() - .enable_node_control() - .with_run_duration(run_duration) - .with_workload( - OpenRaftKvFailoverWorkload::new() - .first_batch(INITIAL_WRITE_BATCH) - .second_batch(SECOND_WRITE_BATCH) - .timeout(workload_timeout) - .key_prefix(RAFT_KEY_PREFIX), - ) - .with_expectation( - OpenRaftKvConverges::new(TOTAL_WRITES) - .timeout(run_duration) - .key_prefix(RAFT_KEY_PREFIX), - ) - .build()?, + Ok(OpenRaftKvScenarioBuilder::with_existing_openraft_kv_app( + OpenRaftKvExistingClusterApp::nodes(3), ) + .enable_node_control() + .with_run_duration(run_duration) + .with_workload(OpenRaftKvClusterAccessible::new(3)) + .with_workload( + OpenRaftKvFailoverWorkload::new() + .first_batch(INITIAL_WRITE_BATCH) + .second_batch(SECOND_WRITE_BATCH) + .timeout(workload_timeout) + .key_prefix(RAFT_KEY_PREFIX), + ) + .with_expectation( + OpenRaftKvConverges::new(TOTAL_WRITES) + .timeout(run_duration) + .key_prefix(RAFT_KEY_PREFIX), + ) + .build()?) } diff --git a/examples/openraft_kv/testing/integration/Cargo.toml b/examples/openraft_kv/testing/integration/Cargo.toml index ef40e34..e058731 100644 --- a/examples/openraft_kv/testing/integration/Cargo.toml +++ b/examples/openraft_kv/testing/integration/Cargo.toml @@ -8,6 +8,7 @@ version.workspace = true async-trait = { workspace = true } openraft-kv-node = { path = "../../openraft-kv-node" } reqwest = { workspace = true } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } testing-framework-runner-compose = { workspace = true } testing-framework-runner-k8s = { workspace = true } diff --git a/examples/openraft_kv/testing/integration/src/app.rs b/examples/openraft_kv/testing/integration/src/app.rs index bc2c7f1..6b37059 100644 --- a/examples/openraft_kv/testing/integration/src/app.rs +++ b/examples/openraft_kv/testing/integration/src/app.rs @@ -1,9 +1,14 @@ use std::io::Error; +use async_trait::async_trait; use openraft_kv_node::{OpenRaftKvClient, OpenRaftKvNodeConfig}; -use testing_framework_core::scenario::{ - Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError, - NodeAccess, serialize_cluster_yaml_config, +use testing_framework_app::{AppDeployment, AppHostEnv, DeployContext, LocalAppCluster}; +use testing_framework_core::{ + scenario::{ + Application, ClusterNodeConfigApplication, ClusterNodeView, ClusterPeerView, DynError, + NodeAccess, NodeClients, serialize_cluster_yaml_config, + }, + topology::DeploymentDescriptor, }; /// Three-node topology used by the OpenRaft example scenarios. @@ -57,3 +62,116 @@ impl ClusterNodeConfigApplication for OpenRaftKvEnv { serialize_cluster_yaml_config(config).map_err(Error::other) } } + +/// Runtime handle exposed by the OpenRaft app deployment. +#[derive(Clone)] +pub struct OpenRaftKvCluster { + deployment: OpenRaftKvTopology, + node_clients: NodeClients, +} + +impl OpenRaftKvCluster { + #[must_use] + pub const fn new( + deployment: OpenRaftKvTopology, + node_clients: NodeClients, + ) -> Self { + Self { + deployment, + node_clients, + } + } + + #[must_use] + pub fn topology(&self) -> &OpenRaftKvTopology { + &self.deployment + } + + #[must_use] + pub fn node_count(&self) -> usize { + self.deployment.node_count() + } + + #[must_use] + pub fn clients(&self) -> Vec { + self.node_clients.snapshot() + } + + pub async fn states(&self) -> Result, DynError> { + let clients = self.clients(); + let mut states = Vec::with_capacity(clients.len()); + + for client in clients { + states.push(client.state().await?); + } + + states.sort_by_key(|state| state.node_id); + + Ok(states) + } +} + +/// App preset for the OpenRaft key-value example. +#[derive(Clone)] +pub struct OpenRaftKvExistingClusterApp { + topology: OpenRaftKvTopology, +} + +impl OpenRaftKvExistingClusterApp { + #[must_use] + pub fn nodes(nodes: usize) -> Self { + Self { + topology: OpenRaftKvTopology::new(nodes), + } + } + + #[must_use] + pub fn topology(&self) -> OpenRaftKvTopology { + self.topology.clone() + } +} + +#[async_trait] +impl AppDeployment for OpenRaftKvExistingClusterApp { + type Handle = OpenRaftKvCluster; + + async fn deploy( + self, + ctx: &mut DeployContext, + ) -> Result { + Ok(OpenRaftKvCluster::new( + ctx.deployment().clone(), + ctx.node_clients().clone(), + )) + } +} + +/// Local app preset that starts its own OpenRaft cluster. +#[derive(Clone)] +pub struct OpenRaftKvLocalApp { + deployment: OpenRaftKvTopology, +} + +impl OpenRaftKvLocalApp { + #[must_use] + pub fn nodes(nodes: usize) -> Self { + Self { + deployment: OpenRaftKvTopology::new(nodes), + } + } + + #[must_use] + pub fn deployment(&self) -> OpenRaftKvTopology { + self.deployment.clone() + } +} + +#[async_trait] +impl AppDeployment for OpenRaftKvLocalApp { + type Handle = LocalAppCluster; + + async fn deploy(self, ctx: &mut DeployContext) -> Result { + ctx.deploy_local_cluster::(self.deployment) + .await + } +} diff --git a/examples/openraft_kv/testing/integration/src/local_env.rs b/examples/openraft_kv/testing/integration/src/local_env.rs index c01183f..e65f440 100644 --- a/examples/openraft_kv/testing/integration/src/local_env.rs +++ b/examples/openraft_kv/testing/integration/src/local_env.rs @@ -1,4 +1,8 @@ -use std::collections::{BTreeMap, HashMap}; +use std::{ + collections::{BTreeMap, HashMap}, + path::PathBuf, + sync::Arc, +}; use openraft_kv_node::OpenRaftKvNodeConfig; use testing_framework_core::{ @@ -6,7 +10,8 @@ use testing_framework_core::{ topology::DeploymentDescriptor, }; use testing_framework_runner_local::{ - BuiltNodeConfig, LocalDeployerEnv, LocalNodePorts, LocalProcessSpec, NodeConfigEntry, + BinaryProviderRef, BuildBinaryProvider, BuildCommand, BuiltNodeConfig, EnvBinaryProvider, + FallbackBinaryProvider, LocalDeployerEnv, LocalNodePorts, LocalProcessSpec, NodeConfigEntry, reserve_local_node_ports, yaml_node_config, }; @@ -80,7 +85,11 @@ impl LocalDeployerEnv for OpenRaftKvEnv { } fn local_process_spec() -> Option { - Some(LocalProcessSpec::new("OPENRAFT_KV_NODE_BIN").with_rust_log("info")) + Some( + LocalProcessSpec::new("OPENRAFT_KV_NODE_BIN") + .with_binary_provider(openraft_binary_provider()) + .with_rust_log("info"), + ) } fn render_local_config(config: &OpenRaftKvNodeConfig) -> Result, DynError> { @@ -92,6 +101,34 @@ impl LocalDeployerEnv for OpenRaftKvEnv { } } +fn openraft_binary_provider() -> FallbackBinaryProvider { + let workspace = workspace_root(); + let providers: [BinaryProviderRef; 2] = [ + Arc::new(EnvBinaryProvider::new("OPENRAFT_KV_NODE_BIN")), + Arc::new(BuildBinaryProvider { + command: BuildCommand::new("cargo").with_args([ + "build", + "-p", + "openraft-kv-node", + "--bin", + "openraft-kv-node", + ]), + output_path: PathBuf::from(format!( + "target/debug/openraft-kv-node{}", + std::env::consts::EXE_SUFFIX + )), + working_dir: Some(workspace), + lock_dir: None, + }), + ]; + + FallbackBinaryProvider::new(providers) +} + +fn workspace_root() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../../..") +} + fn local_node_config( index: usize, network_port: u16, diff --git a/examples/openraft_kv/testing/integration/src/scenario.rs b/examples/openraft_kv/testing/integration/src/scenario.rs index d2de604..432d6e6 100644 --- a/examples/openraft_kv/testing/integration/src/scenario.rs +++ b/examples/openraft_kv/testing/integration/src/scenario.rs @@ -1,7 +1,9 @@ +use testing_framework_app::AppScenarioBuilderExt; use testing_framework_core::scenario::{CoreBuilderExt, ScenarioBuilder}; use crate::{ - OpenRaftClusterObserver, OpenRaftKvEnv, OpenRaftKvTopology, openraft_cluster_source_provider, + OpenRaftClusterObserver, OpenRaftKvEnv, OpenRaftKvExistingClusterApp, OpenRaftKvTopology, + openraft_cluster_source_provider, }; /// Scenario builder alias used by the OpenRaft example binaries. @@ -15,6 +17,10 @@ pub trait OpenRaftKvBuilderExt: Sized { /// Attaches the default OpenRaft cluster observer to the scenario. fn with_cluster_observer(self) -> Self; + + /// Starts from an OpenRaft app preset and exposes its cluster handle to + /// workloads. + fn with_existing_openraft_kv_app(app: OpenRaftKvExistingClusterApp) -> Self; } impl OpenRaftKvBuilderExt for OpenRaftKvScenarioBuilder { @@ -29,4 +35,10 @@ impl OpenRaftKvBuilderExt for OpenRaftKvScenarioBuilder { OpenRaftClusterObserver::config(), ) } + + fn with_existing_openraft_kv_app(app: OpenRaftKvExistingClusterApp) -> Self { + OpenRaftKvScenarioBuilder::with_deployment(app.topology()) + .with_app(app) + .with_cluster_observer() + } } diff --git a/examples/openraft_kv/testing/workloads/Cargo.toml b/examples/openraft_kv/testing/workloads/Cargo.toml index 0682b55..5e03aaf 100644 --- a/examples/openraft_kv/testing/workloads/Cargo.toml +++ b/examples/openraft_kv/testing/workloads/Cargo.toml @@ -9,6 +9,7 @@ anyhow = "1.0" async-trait = { workspace = true } openraft-kv-node = { path = "../../openraft-kv-node" } openraft-kv-runtime-ext = { path = "../integration" } +testing-framework-app = { workspace = true } testing-framework-core = { workspace = true } thiserror = "2.0" tokio = { workspace = true, features = ["full"] } diff --git a/examples/openraft_kv/testing/workloads/src/handle_access.rs b/examples/openraft_kv/testing/workloads/src/handle_access.rs new file mode 100644 index 0000000..758959c --- /dev/null +++ b/examples/openraft_kv/testing/workloads/src/handle_access.rs @@ -0,0 +1,63 @@ +use async_trait::async_trait; +use openraft_kv_runtime_ext::{OpenRaftKvCluster, OpenRaftKvEnv}; +use testing_framework_app::AppRunContextExt; +use testing_framework_core::scenario::{DynError, RunContext, Workload}; +use tracing::info; + +#[derive(Clone)] +pub struct OpenRaftKvClusterAccessible { + expected_nodes: usize, +} + +impl OpenRaftKvClusterAccessible { + #[must_use] + pub const fn new(expected_nodes: usize) -> Self { + Self { expected_nodes } + } +} + +#[async_trait] +impl Workload for OpenRaftKvClusterAccessible { + fn name(&self) -> &str { + "openraft_kv_cluster_accessible" + } + + async fn start(&self, ctx: &RunContext) -> Result<(), DynError> { + let cluster = ctx.require_app::()?; + let states = cluster.states().await?; + let client_count = cluster.clients().len(); + + if cluster.node_count() != self.expected_nodes { + return Err(format!( + "openraft app topology has {} nodes, expected {}", + cluster.node_count(), + self.expected_nodes + ) + .into()); + } + + if client_count != self.expected_nodes { + return Err(format!( + "openraft app handle has {client_count} clients, expected {}", + self.expected_nodes + ) + .into()); + } + + if states.len() != self.expected_nodes { + return Err(format!( + "openraft app handle read {} node states, expected {}", + states.len(), + self.expected_nodes + ) + .into()); + } + + info!( + nodes = self.expected_nodes, + "openraft app handle is accessible from workload" + ); + + Ok(()) + } +} diff --git a/examples/openraft_kv/testing/workloads/src/lib.rs b/examples/openraft_kv/testing/workloads/src/lib.rs index 02bd5a8..4629c5d 100644 --- a/examples/openraft_kv/testing/workloads/src/lib.rs +++ b/examples/openraft_kv/testing/workloads/src/lib.rs @@ -1,11 +1,14 @@ mod convergence; mod failover; +mod handle_access; mod support; /// Replication expectation used by the OpenRaft example binaries. pub use convergence::OpenRaftKvConverges; /// Failover workload used by the OpenRaft example binaries. pub use failover::OpenRaftKvFailoverWorkload; +/// Workload that proves app handles are available to OpenRaft workloads. +pub use handle_access::OpenRaftKvClusterAccessible; /// Shared cluster helpers used by the OpenRaft workload and manual k8s example. pub use support::{ FULL_VOTER_SET, OpenRaftClusterError, OpenRaftMembership, ensure_cluster_size, expected_kv,