logos-blockchain/logos-blockchain-testing
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-blockchain-testing.git synced 2026-02-17 11:43:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use node names for restart/stop control

Browse Source
This commit is contained in:
andrussal 2026-01-29 13:27:42 +01:00
parent 26f312cf6b
commit bb31c03cab
9 changed files with 88 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
examples/doc-snippets/src/node_control_accessing_control.rs
View File

@ -11,8 +11,8 @@ impl Workload for RestartWorkload {
async fn start(&self, ctx: &RunContext) -> Result<(), DynError> {
if let Some(control) = ctx.node_control() {
// Restart the first node (index 0) if supported.
control.restart_node(0).await?;
// Restart the first node by name if supported.
control.restart_node("node-0").await?;
}
Ok(())
}

2
examples/doc-snippets/src/node_control_trait.rs
View File

@ -3,5 +3,5 @@ use testing_framework_core::scenario::DynError;
#[async_trait]
pub trait NodeControlHandle: Send + Sync {
async fn restart_node(&self, index: usize) -> Result<(), DynError>;
async fn restart_node(&self, name: &str) -> Result<(), DynError>;
}

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

@ -8,7 +8,7 @@ use crate::{
/// Deployer-agnostic control surface for runtime node operations.
#[async_trait]
pub trait NodeControlHandle: Send + Sync {
async fn restart_node(&self, _index: usize) -> Result<(), DynError> {
async fn restart_node(&self, _name: &str) -> Result<(), DynError> {
Err("restart_node not supported by this deployer".into())
}
@ -24,7 +24,7 @@ pub trait NodeControlHandle: Send + Sync {
Err("start_node_with not supported by this deployer".into())
}
async fn stop_node(&self, _index: usize) -> Result<(), DynError> {
async fn stop_node(&self, _name: &str) -> Result<(), DynError> {
Err("stop_node not supported by this deployer".into())
}
@ -32,7 +32,7 @@ pub trait NodeControlHandle: Send + Sync {
None
}
fn node_pid(&self, _index: usize) -> Option<u32> {
fn node_pid(&self, _name: &str) -> Option<u32> {
None
}
}

12
testing-framework/deployers/compose/src/docker/control.rs
View File

@ -45,13 +45,9 @@ pub struct ComposeNodeControl {
#[async_trait::async_trait]
impl NodeControlHandle for ComposeNodeControl {
async fn restart_node(&self, index: usize) -> Result<(), DynError> {
restart_compose_service(
&self.compose_file,
&self.project_name,
&format!("node-{index}"),
)
.await
.map_err(|err| format!("node restart failed: {err}").into())
async fn restart_node(&self, name: &str) -> Result<(), DynError> {
restart_compose_service(&self.compose_file, &self.project_name, name)
.await
.map_err(|err| format!("node restart failed: {err}").into())
}
}

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

@ -50,8 +50,8 @@ impl LocalManualCluster {
}
#[must_use]
pub fn node_pid(&self, index: usize) -> Option<u32> {
self.nodes.node_pid(index)
pub fn node_pid(&self, name: &str) -> Option<u32> {
self.nodes.node_pid(name)
}
pub async fn start_node(&self, name: &str) -> Result<StartedNode, ManualClusterError> {
@ -73,12 +73,12 @@ impl LocalManualCluster {
self.nodes.stop_all();
}
pub async fn restart_node(&self, index: usize) -> Result<(), ManualClusterError> {
Ok(self.nodes.restart_node(index).await?)
pub async fn restart_node(&self, name: &str) -> Result<(), ManualClusterError> {
Ok(self.nodes.restart_node(name).await?)
}
pub async fn stop_node(&self, index: usize) -> Result<(), ManualClusterError> {
Ok(self.nodes.stop_node(index).await?)
pub async fn stop_node(&self, name: &str) -> Result<(), ManualClusterError> {
Ok(self.nodes.stop_node(name).await?)
}
pub async fn wait_network_ready(&self) -> Result<(), ReadinessError> {
@ -107,15 +107,15 @@ impl Drop for LocalManualCluster {
#[async_trait::async_trait]
impl NodeControlHandle for LocalManualCluster {
async fn restart_node(&self, index: usize) -> Result<(), DynError> {
async fn restart_node(&self, name: &str) -> Result<(), DynError> {
self.nodes
.restart_node(index)
.restart_node(name)
.await
.map_err(|err| err.into())
}
async fn stop_node(&self, index: usize) -> Result<(), DynError> {
self.nodes.stop_node(index).await.map_err(|err| err.into())
async fn stop_node(&self, name: &str) -> Result<(), DynError> {
self.nodes.stop_node(name).await.map_err(|err| err.into())
}
async fn start_node(&self, name: &str) -> Result<StartedNode, DynError> {
@ -138,8 +138,8 @@ impl NodeControlHandle for LocalManualCluster {
self.node_client(name)
}
fn node_pid(&self, index: usize) -> Option<u32> {
self.node_pid(index)
fn node_pid(&self, name: &str) -> Option<u32> {
self.node_pid(name)
}
}

58
testing-framework/deployers/local/src/node_control/mod.rs
View File

@ -44,8 +44,8 @@ pub enum LocalNodeManagerError {
PortAllocation { message: String },
#[error("node config patch failed: {message}")]
ConfigPatch { message: String },
#[error("node index {index} is out of bounds")]
NodeIndex { index: usize },
#[error("node name '{name}' is unknown")]
NodeName { name: String },
#[error("failed to restart node: {source}")]
Restart {
#[source]
@ -145,6 +145,7 @@ impl LocalNodeManager {
peer_ports: seed.peer_ports.clone(),
peer_ports_by_name: seed.peer_ports_by_name.clone(),
clients_by_name: HashMap::new(),
indices_by_name: HashMap::new(),
nodes: Vec::new(),
};
@ -169,12 +170,13 @@ impl LocalNodeManager {
}
#[must_use]
pub fn node_pid(&self, index: usize) -> Option<u32> {
pub fn node_pid(&self, name: &str) -> Option<u32> {
let mut state = self
.state
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let index = *state.indices_by_name.get(name)?;
let node = state.nodes.get_mut(index)?;
if node.is_running() {
Some(node.pid())
@ -195,6 +197,7 @@ impl LocalNodeManager {
.peer_ports_by_name
.clone_from(&self.seed.peer_ports_by_name);
state.clients_by_name.clear();
state.indices_by_name.clear();
state.node_count = self.seed.node_count;
self.node_clients.clear();
}
@ -211,6 +214,7 @@ impl LocalNodeManager {
state.peer_ports.clear();
state.peer_ports_by_name.clear();
state.clients_by_name.clear();
state.indices_by_name.clear();
state.node_count = 0;
for (idx, node) in nodes.into_iter().enumerate() {
@ -290,6 +294,8 @@ impl LocalNodeManager {
let index = state.node_count;
let label = if name.trim().is_empty() {
Self::default_label(index)
} else if name.starts_with("node-") {
name.to_string()
} else {
format!("node-{name}")
};
@ -334,18 +340,27 @@ impl LocalNodeManager {
})
}
pub async fn restart_node(&self, index: usize) -> Result<(), LocalNodeManagerError> {
let mut node = {
pub async fn restart_node(&self, name: &str) -> Result<(), LocalNodeManagerError> {
let (index, mut node) = {
let mut state = self
.state
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let Some(index) = state.indices_by_name.get(name).copied() else {
return Err(LocalNodeManagerError::NodeName {
name: name.to_string(),
});
};
if index >= state.nodes.len() {
return Err(LocalNodeManagerError::NodeIndex { index });
return Err(LocalNodeManagerError::NodeName {
name: name.to_string(),
});
}
state.nodes.remove(index)
let node = state.nodes.remove(index);
(index, node)
};
node.restart()
@ -366,18 +381,27 @@ impl LocalNodeManager {
Ok(())
}
pub async fn stop_node(&self, index: usize) -> Result<(), LocalNodeManagerError> {
let mut node = {
pub async fn stop_node(&self, name: &str) -> Result<(), LocalNodeManagerError> {
let (index, mut node) = {
let mut state = self
.state
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
let Some(index) = state.indices_by_name.get(name).copied() else {
return Err(LocalNodeManagerError::NodeName {
name: name.to_string(),
});
};
if index >= state.nodes.len() {
return Err(LocalNodeManagerError::NodeIndex { index });
return Err(LocalNodeManagerError::NodeName {
name: name.to_string(),
});
}
state.nodes.remove(index)
let node = state.nodes.remove(index);
(index, node)
};
node.stop().await;
@ -446,12 +470,12 @@ fn apply_patch_if_needed(
#[async_trait::async_trait]
impl NodeControlHandle for LocalNodeManager {
async fn restart_node(&self, index: usize) -> Result<(), DynError> {
self.restart_node(index).await.map_err(|err| err.into())
async fn restart_node(&self, name: &str) -> Result<(), DynError> {
self.restart_node(name).await.map_err(|err| err.into())
}
async fn stop_node(&self, index: usize) -> Result<(), DynError> {
self.stop_node(index).await.map_err(|err| err.into())
async fn stop_node(&self, name: &str) -> Result<(), DynError> {
self.stop_node(name).await.map_err(|err| err.into())
}
async fn start_node(&self, name: &str) -> Result<StartedNode, DynError> {
@ -474,7 +498,7 @@ impl NodeControlHandle for LocalNodeManager {
self.node_client(name)
}
fn node_pid(&self, index: usize) -> Option<u32> {
self.node_pid(index)
fn node_pid(&self, name: &str) -> Option<u32> {
self.node_pid(name)
}
}

3
testing-framework/deployers/local/src/node_control/state.rs
View File

@ -7,6 +7,7 @@ pub(crate) struct LocalNodeManagerState {
pub(crate) peer_ports: Vec<u16>,
pub(crate) peer_ports_by_name: HashMap<String, u16>,
pub(crate) clients_by_name: HashMap<String, ApiClient>,
pub(crate) indices_by_name: HashMap<String, usize>,
pub(crate) nodes: Vec<Node>,
}
@ -26,6 +27,8 @@ impl LocalNodeManagerState {
node: Node,
) {
self.register_common(node_name, network_port, client);
let index = self.nodes.len();
self.indices_by_name.insert(node_name.to_string(), index);
self.node_count += 1;
self.nodes.push(node);
}

23
testing-framework/deployers/local/tests/restart.rs
View File

@ -22,16 +22,17 @@ async fn local_restart_node() -> Result<(), Box<dyn std::error::Error + Send + S
let control = context.node_control().ok_or("node control not available")?;
let old_pid = control.node_pid(0).ok_or("missing node pid")?;
let node_name = "node-0";
let old_pid = control.node_pid(node_name).ok_or("missing node pid")?;
control.restart_node(0).await?;
control.restart_node(node_name).await?;
let new_pid = control.node_pid(0).ok_or("missing node pid")?;
let new_pid = control.node_pid(node_name).ok_or("missing node pid")?;
assert_ne!(old_pid, new_pid, "expected a new process after restart");
control.stop_node(0).await?;
control.stop_node(node_name).await?;
assert!(
control.node_pid(0).is_none(),
control.node_pid(node_name).is_none(),
"expected node pid to be absent after stop"
);
@ -47,18 +48,18 @@ async fn manual_cluster_restart_node() -> Result<(), Box<dyn std::error::Error +
let deployer = LocalDeployer::default();
let cluster = deployer.manual_cluster(TopologyConfig::with_node_numbers(1))?;
cluster.start_node("a").await?;
let node_name = cluster.start_node("a").await?.name;
let old_pid = cluster.node_pid(0).ok_or("missing node pid")?;
let old_pid = cluster.node_pid(&node_name).ok_or("missing node pid")?;
cluster.restart_node(0).await?;
cluster.restart_node(&node_name).await?;
let new_pid = cluster.node_pid(0).ok_or("missing node pid")?;
let new_pid = cluster.node_pid(&node_name).ok_or("missing node pid")?;
assert_ne!(old_pid, new_pid, "expected a new process after restart");
cluster.stop_node(0).await?;
cluster.stop_node(&node_name).await?;
assert!(
cluster.node_pid(0).is_none(),
cluster.node_pid(&node_name).is_none(),
"expected node pid to be absent after stop"
);

20
testing-framework/workflows/src/workloads/chaos.rs
View File

@ -44,7 +44,7 @@ impl RandomRestartWorkload {
if self.include_nodes {
if node_count > 1 {
for index in 0..node_count {
targets.push(Target::Node(index));
targets.push(Target::Node(format!("node-{index}")));
}
} else if node_count == 1 {
info!("chaos restart skipping nodes: only one node configured");
@ -76,7 +76,7 @@ impl RandomRestartWorkload {
let ready = now.checked_sub(self.target_cooldown).unwrap_or(now);
targets
.iter()
.copied()
.cloned()
.map(|target| (target, ready))
.collect()
}
@ -111,16 +111,16 @@ impl RandomRestartWorkload {
let available: Vec<Target> = targets
.iter()
.copied()
.cloned()
.filter(|target| cooldowns.get(target).is_none_or(|ready| *ready <= now))
.collect();
if let Some(choice) = available.choose(&mut thread_rng()).copied() {
if let Some(choice) = available.choose(&mut thread_rng()).cloned() {
tracing::debug!(?choice, "chaos restart picked target");
return Ok(choice);
}
if let Some(choice) = targets.choose(&mut thread_rng()).copied() {
if let Some(choice) = targets.choose(&mut thread_rng()).cloned() {
return Ok(choice);
}
return Err("chaos restart workload has no eligible targets".into());
@ -158,10 +158,10 @@ impl Workload for RandomRestartWorkload {
let target = self.pick_target(&targets, &cooldowns).await?;
match target {
Target::Node(index) => {
tracing::info!(index, "chaos restarting node");
Target::Node(ref name) => {
tracing::info!(name, "chaos restarting node");
handle
.restart_node(index)
.restart_node(name)
.await
.map_err(|err| format!("node restart failed: {err}"))?
}
@ -172,7 +172,7 @@ impl Workload for RandomRestartWorkload {
}
}
#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
enum Target {
Node(usize),
Node(String),
}