zombienet_orchestrator/network/node/core.rs
1//! Provider-generic node behaviour, shared by every node type.
2//!
3//! [`NodeCore`] wraps the handle to the running process (the provider's
4//! [`DynNode`]) plus the small amount of runtime state zombienet keeps about
5//! it, and implements everything that doesn't depend on _what_ the node is
6//! running: logs, lifecycle (pause/resume/restart), scripts and db snapshots.
7//!
8//! Chain-flavoured nodes ([`NetworkNode`](super::NetworkNode) for substrate,
9//! [`JamNetworkNode`](super::jam::JamNetworkNode) for JAM) embed one of these
10//! and add their own protocol-specific surface on top.
11
12use std::{
13 path::{Path, PathBuf},
14 sync::{
15 atomic::{AtomicBool, AtomicU64, Ordering},
16 Arc,
17 },
18 time::{Duration, SystemTime, UNIX_EPOCH},
19};
20
21use anyhow::anyhow;
22use configuration::types::AssetLocation;
23use fancy_regex::Regex;
24use glob_match::glob_match;
25use provider::{
26 types::{ExecutionResult, InnerSnapshotDb, RunScriptOptions},
27 DynNode,
28};
29use serde::Serialize;
30use tracing::{debug, trace};
31
32use super::{
33 serialize_provider_node, spawned::NodeKind, BoxedClosure, LogLineCount, LogLineCountOptions,
34};
35use crate::shared::types::NodeSnapshot;
36
37/// The provider handle and runtime state every spawned node has.
38///
39/// Cloning a `NodeCore` shares the same underlying process handle _and_ the
40/// same running/last-start state, so all the clones of a node stay in sync.
41#[derive(Clone, Serialize)]
42pub struct NodeCore {
43 #[serde(serialize_with = "serialize_provider_node")]
44 pub(crate) inner: DynNode,
45 pub(crate) name: String,
46 /// What flavour of node this is. Stored (and serialized to `zombie.json`)
47 /// so a node can be identified without its concrete type at hand.
48 pub(crate) kind: NodeKind,
49 #[serde(skip)]
50 is_running: Arc<AtomicBool>,
51 // Store the last timestamp when we start the node
52 #[serde(skip)]
53 last_start_ts: Arc<AtomicU64>,
54}
55
56impl NodeCore {
57 pub(crate) fn new(name: impl Into<String>, inner: DynNode, kind: NodeKind) -> Self {
58 Self {
59 inner,
60 name: name.into(),
61 kind,
62 is_running: Arc::new(AtomicBool::new(false)),
63 last_start_ts: Arc::new(AtomicU64::new(0)),
64 }
65 }
66
67 /// The provider node (the actual running process/pod/container).
68 pub fn inner(&self) -> &DynNode {
69 &self.inner
70 }
71
72 pub fn name(&self) -> &str {
73 &self.name
74 }
75
76 /// What flavour of node this is.
77 pub fn kind(&self) -> NodeKind {
78 self.kind
79 }
80
81 /// Args used for bootstrap the node.
82 /// NOTE: this may not be in sync if you restart the node with new args.
83 pub fn args(&self) -> Vec<&str> {
84 self.inner.args()
85 }
86
87 /// Check if the node is currently running (not paused).
88 ///
89 /// This returns the internal running state.
90 pub fn is_running(&self) -> bool {
91 self.is_running.load(Ordering::Acquire)
92 }
93
94 /// Get the last timestamp when the node start.
95 pub fn last_start_ts(&self) -> u64 {
96 self.last_start_ts.load(Ordering::Acquire)
97 }
98
99 pub(crate) fn set_is_running(&self, is_running: bool) {
100 self.is_running.store(is_running, Ordering::Release);
101 }
102
103 /// Set the timestamp when the node was started
104 pub(crate) fn set_last_start_ts(&self, ts: u64) {
105 self.last_start_ts.store(ts, Ordering::Release);
106 }
107
108 /// On-disk base directory of the node — root of `data/`, `relay-data/`,
109 /// `cfg/`, etc.
110 /// This will be the _base directory_ of the inner (provider) node.
111 pub fn base_dir(&self) -> &PathBuf {
112 self.inner.base_dir()
113 }
114
115 /// Pause the node, this is implemented by pausing the
116 /// actual process (e.g polkadot) with sending `SIGSTOP` signal
117 ///
118 /// Note: If you're using this method with the native provider on the attached network, the live network has to be running
119 /// with global setting `teardown_on_failure` disabled.
120 pub async fn pause(&self) -> Result<(), anyhow::Error> {
121 // Cleared first so the watcher does not read the pause as a crash;
122 // put back if the provider refused (a process without the wrapper).
123 let was_running = self.is_running();
124 self.set_is_running(false);
125 self.inner.pause().await.inspect_err(|_| {
126 self.set_is_running(was_running);
127 })?;
128 Ok(())
129 }
130
131 /// Resume the node, this is implemented by resuming the
132 /// actual process (e.g polkadot) with sending `SIGCONT` signal
133 ///
134 /// Note: If you're using this method with the native provider on the attached network, the live network has to be running
135 /// with global setting `teardown_on_failure` disabled.
136 pub async fn resume(&self) -> Result<(), anyhow::Error> {
137 let was_running = self.is_running();
138 self.set_is_running(true);
139 self.inner.resume().await.inspect_err(|_| {
140 self.set_is_running(was_running);
141 })?;
142 Ok(())
143 }
144
145 /// Restart the node using the same `cmd`, `args` and `env` (and same isolated dir)
146 ///
147 /// Note: If you're using this method with the native provider on the attached network, the live network has to be running
148 /// with global setting `teardown_on_failure` disabled.
149 pub async fn restart(&self, after: Option<Duration>) -> Result<(), anyhow::Error> {
150 // Not put back on failure: the provider may have stopped the process
151 // before failing to start it again.
152 self.set_is_running(false);
153 self.inner.restart(after).await?;
154 self.set_is_running(true);
155 self.set_last_start_ts(SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs());
156 Ok(())
157 }
158
159 /// Get the logs of the node
160 /// TODO: do we need the `since` param, maybe we could be handy later for loop filtering
161 pub async fn logs(&self) -> Result<String, anyhow::Error> {
162 Ok(self.inner.logs().await?)
163 }
164
165 /// Wait until a the number of matching log lines is reach
166 pub async fn wait_log_line_count(
167 &self,
168 pattern: impl Into<String>,
169 is_glob: bool,
170 count: usize,
171 ) -> Result<(), anyhow::Error> {
172 let pattern = pattern.into();
173 let pattern_clone = pattern.clone();
174 debug!("waiting until we find pattern {pattern} {count} times");
175 let match_fn: BoxedClosure = if is_glob {
176 Box::new(move |line: &str| Ok(glob_match(&pattern, line)))
177 } else {
178 let re = Regex::new(&pattern)?;
179 Box::new(move |line: &str| re.is_match(line).map_err(|e| anyhow!(e.to_string())))
180 };
181
182 loop {
183 let mut q = 0_usize;
184 let logs = self.logs().await?;
185 for line in logs.lines() {
186 trace!("line is {line}");
187 if match_fn(line)? {
188 trace!("pattern {pattern_clone} match in line {line}");
189 q += 1;
190 if q >= count {
191 return Ok(());
192 }
193 }
194 }
195
196 tokio::time::sleep(Duration::from_secs(2)).await;
197 }
198 }
199
200 /// Waits until the number of matching log lines satisfies a custom condition,
201 /// optionally waiting for the entire duration of the timeout.
202 ///
203 /// This method searches log lines for a given substring or glob pattern,
204 /// and evaluates the number of matching lines using a user-provided predicate function.
205 /// Optionally, it can wait for the full timeout duration to ensure the condition
206 /// holds consistently (e.g., for verifying absence of logs).
207 ///
208 /// # Arguments
209 /// * `substring` - The substring or pattern to match within log lines.
210 /// * `is_glob` - Whether to treat `substring` as a glob pattern (`true`) or a regex (`false`).
211 /// * `options` - Configuration for timeout, match count predicate, and full-duration waiting.
212 ///
213 /// # Returns
214 /// * `Ok(LogLineCount::TargetReached(n))` if the predicate was satisfied within the timeout,
215 /// * `Ok(LogLineCount::TargetFails(n))` if the predicate was not satisfied in time,
216 /// * `Err(e)` if an error occurred during log retrieval or matching.
217 ///
218 /// # Example
219 /// ```rust
220 /// # use std::{sync::Arc, time::Duration};
221 /// # use provider::NativeProvider;
222 /// # use support::{fs::local::LocalFileSystem};
223 /// # use zombienet_orchestrator::{Orchestrator, network::node::{NetworkNode, LogLineCountOptions}};
224 /// # use configuration::NetworkConfig;
225 /// # async fn example() -> Result<(), anyhow::Error> {
226 /// # let provider = NativeProvider::new(LocalFileSystem {});
227 /// # let orchestrator = Orchestrator::new(LocalFileSystem {}, provider);
228 /// # let config = NetworkConfig::load_from_toml("config.toml")?;
229 /// # let network = orchestrator.spawn(config).await?;
230 /// let node = network.get_node("alice")?;
231 /// // Wait (up to 10 seconds) until pattern occurs once
232 /// let options = LogLineCountOptions {
233 /// predicate: Arc::new(|count| count == 1),
234 /// timeout: Duration::from_secs(10),
235 /// wait_until_timeout_elapses: false,
236 /// };
237 /// let result = node
238 /// .wait_log_line_count_with_timeout("error", false, options)
239 /// .await?;
240 /// # Ok(())
241 /// # }
242 /// ```
243 pub async fn wait_log_line_count_with_timeout(
244 &self,
245 substring: impl Into<String>,
246 is_glob: bool,
247 options: LogLineCountOptions,
248 ) -> Result<LogLineCount, anyhow::Error> {
249 let substring = substring.into();
250 debug!(
251 "waiting until match lines count within {} seconds",
252 options.timeout.as_secs_f64()
253 );
254
255 let start = tokio::time::Instant::now();
256
257 let match_fn: BoxedClosure = if is_glob {
258 Box::new(move |line: &str| Ok(glob_match(&substring, line)))
259 } else {
260 let re = Regex::new(&substring)?;
261 Box::new(move |line: &str| re.is_match(line).map_err(|e| anyhow!(e.to_string())))
262 };
263
264 if options.wait_until_timeout_elapses {
265 tokio::time::sleep(options.timeout).await;
266 }
267
268 let mut q;
269 loop {
270 q = 0_u32;
271 let logs = self.logs().await?;
272 for line in logs.lines() {
273 if match_fn(line)? {
274 q += 1;
275
276 // If `wait_until_timeout_elapses` is set then check the condition just once at the
277 // end after the whole log file is processed. This is to address the cases when the
278 // predicate becomes true and false again.
279 // eg. expected exactly 2 matching lines are expected but 3 are present
280 if !options.wait_until_timeout_elapses && (options.predicate)(q) {
281 return Ok(LogLineCount::TargetReached(q));
282 }
283 }
284 }
285
286 if start.elapsed() >= options.timeout {
287 break;
288 }
289
290 tokio::time::sleep(Duration::from_secs(2)).await;
291 }
292
293 if (options.predicate)(q) {
294 Ok(LogLineCount::TargetReached(q))
295 } else {
296 Ok(LogLineCount::TargetFailed(q))
297 }
298 }
299
300 /// Restart the node overriding the program and/or args, optionally
301 /// downloading `assets` (binaries) first.
302 ///
303 /// The node keeps the same `env` and isolated directory (e.g same database).
304 /// Callers are expected to have (re)generated `program`/`args` for their
305 /// own node flavour.
306 pub(crate) async fn restart_with(
307 &self,
308 assets: &[AssetLocation],
309 program: &str,
310 args: &[String],
311 after: Option<Duration>,
312 ) -> Result<(), anyhow::Error> {
313 self.set_is_running(false);
314 self.inner
315 .restart_with(assets, program, args, after)
316 .await?;
317 self.set_is_running(true);
318 self.set_last_start_ts(SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs());
319 Ok(())
320 }
321
322 /// Tar the node's database into `out_path` (gzipped).
323 ///
324 /// NOTE: Currently __only__ implemented in native provider. Also,
325 /// the caller is responsible for pausing the node first;
326 /// snapshotting a running node risks a torn RocksDB state.
327 pub(crate) async fn snapshot_db(
328 &self,
329 out_path: impl AsRef<Path>,
330 is_cumulus_based: bool,
331 ) -> Result<NodeSnapshot, anyhow::Error> {
332 let out_path = out_path.as_ref().to_path_buf();
333
334 let InnerSnapshotDb {
335 filename,
336 sha256,
337 size,
338 } = self.inner.snapshot_db(is_cumulus_based).await?;
339
340 // now we need to _move_ the inner file to the out_path
341 let remote_file_path = PathBuf::from(&filename);
342 self.inner
343 .receive_file(remote_file_path.as_ref(), out_path.as_ref())
344 .await?;
345
346 Ok(NodeSnapshot {
347 path: out_path,
348 sha256,
349 size,
350 node_name: self.name().into(),
351 })
352 }
353
354 /// Run a script inside the node's container/environment
355 ///
356 /// The script will be uploaded to the node, made executable, and executed with
357 /// the provided arguments and environment variables.
358 ///
359 /// Returns `Ok(stdout)` on success, or `Err((exit_status, stderr))` on failure.
360 pub async fn run_script(
361 &self,
362 options: RunScriptOptions,
363 ) -> Result<ExecutionResult, anyhow::Error> {
364 self.inner
365 .run_script(options)
366 .await
367 .map_err(|e| anyhow!("Failed to run script: {e}"))
368 }
369}
370
371impl std::fmt::Debug for NodeCore {
372 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
373 f.debug_struct("NodeCore")
374 .field("inner", &"inner_skipped")
375 .field("name", &self.name)
376 .field("kind", &self.kind)
377 .field("is_running", &self.is_running())
378 .finish()
379 }
380}