From e5c2280a1cec5c53e20fc877e32dca50fa56639e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C4=99drzej=20Stuczy=C5=84ski?= Date: Thu, 9 Nov 2023 11:27:37 +0000 Subject: [PATCH] main run loop --- tools/nymvisor/src/cli/init.rs | 4 +- tools/nymvisor/src/cli/mod.rs | 2 +- tools/nymvisor/src/cli/run.rs | 79 ++++++++----------- tools/nymvisor/src/config/mod.rs | 4 + tools/nymvisor/src/error.rs | 6 ++ tools/nymvisor/src/tasks/launcher/backup.rs | 18 ++--- tools/nymvisor/src/tasks/launcher/mod.rs | 62 ++++++++++++++- .../src/tasks/upgrade_plan_watcher.rs | 7 +- tools/nymvisor/src/tasks/upstream_poller.rs | 2 +- tools/nymvisor/src/upgrades/mod.rs | 8 ++ 10 files changed, 126 insertions(+), 66 deletions(-) diff --git a/tools/nymvisor/src/cli/init.rs b/tools/nymvisor/src/cli/init.rs index 03118a58f8..5ab40376db 100644 --- a/tools/nymvisor/src/cli/init.rs +++ b/tools/nymvisor/src/cli/init.rs @@ -7,13 +7,13 @@ use crate::env::Env; use crate::error::NymvisorError; use crate::upgrades::types::{UpgradeInfo, UpgradePlan}; use nym_bin_common::build_information::BinaryBuildInformationOwned; -use nym_bin_common::logging::{setup_logging, setup_tracing_logger}; +use nym_bin_common::logging::setup_tracing_logger; use nym_bin_common::output_format::OutputFormat; use std::fs; use std::path::{Path, PathBuf}; use std::time::Duration; use time::OffsetDateTime; -use tracing::{debug, error, info, trace, warn}; +use tracing::{debug, info, trace, warn}; use url::Url; #[derive(clap::Args, Debug)] diff --git a/tools/nymvisor/src/cli/mod.rs b/tools/nymvisor/src/cli/mod.rs index cd0588ed61..4e872ab36e 100644 --- a/tools/nymvisor/src/cli/mod.rs +++ b/tools/nymvisor/src/cli/mod.rs @@ -13,7 +13,7 @@ use crate::error::NymvisorError; use clap::{Parser, Subcommand}; use lazy_static::lazy_static; use nym_bin_common::bin_info; -use std::path::{Path, PathBuf}; +use std::path::Path; use tracing::error; lazy_static! { diff --git a/tools/nymvisor/src/cli/run.rs b/tools/nymvisor/src/cli/run.rs index fb9343b338..9673ee5ca2 100644 --- a/tools/nymvisor/src/cli/run.rs +++ b/tools/nymvisor/src/cli/run.rs @@ -2,20 +2,14 @@ // SPDX-License-Identifier: Apache-2.0 use crate::cli::try_load_current_config; -use crate::daemon::Daemon; use crate::env::Env; use crate::error::NymvisorError; -use crate::upgrades::upgrade_binary; -use async_file_watcher::AsyncFileWatcher; -use futures::channel::mpsc; -use futures::future::{AbortHandle, Abortable}; -use futures::StreamExt; +use crate::tasks::launcher::DaemonLauncher; +use crate::tasks::upgrade_plan_watcher::start_upgrade_plan_watcher; +use crate::tasks::upstream_poller::UpstreamPoller; use nym_bin_common::logging::setup_tracing_logger; -use std::sync::Arc; -use std::time::Duration; use tokio::runtime; -use tokio::sync::Notify; -use tracing::info; +use tracing::{error, info}; #[derive(clap::Args, Debug)] pub(crate) struct Args { @@ -33,8 +27,6 @@ pub(crate) fn execute(args: Args) -> Result<(), NymvisorError> { info!("starting nymvisor for {}", config.daemon.name); - // upgrade_binary(&config)?; - // TODO: experiment with the minimal runtime // look at futures::executor::LocalPool // well, if the creation of the runtime failed, there isn't much we could do @@ -50,40 +42,31 @@ pub(crate) fn execute(args: Args) -> Result<(), NymvisorError> { // - the last one for polling upstream source for upgrade info // so once the daemon has finished, for whatever reason, abort the file watcher and upstream poller to terminate the nymvisor - todo!() - // // spawn the root task - // rt.block_on(async { - // println!("run"); - // - // let daemon = Daemon::from_config(&config); - // let running = daemon.execute_async(args.daemon_args)?; - // - // let handle1 = tokio::spawn(async move { - // let res = running.await; - // println!("the process has finished! with {res:?}"); - // }); - // - // let (events_sender, mut events_receiver) = mpsc::unbounded(); - // let mut watcher = AsyncFileWatcher::new_file_changes_watcher( - // config.upgrade_plan_filepath(), - // events_sender, - // )?; - // - // let (abort_handle, abort_registration) = AbortHandle::new_pair(); - // - // let handle2 = - // tokio::spawn(async move { Abortable::new(watcher.watch(), abort_registration).await }); - // - // let event = events_receiver.next().await; - // println!("watcher event: {event:?}"); - // interrupt_notify.notify_one(); - // - // handle1.await; - // abort_handle.abort(); - // handle2.await; - // - // // println!("{:?}", status); - // - // >::Ok(()) - // }) + // spawn the root task + rt.block_on(async { + let (upgrade_receiver, watcher_handle) = start_upgrade_plan_watcher(&config)?; + let upstream_poller_handle = UpstreamPoller::new(&config).start(); + let mut launcher = DaemonLauncher::new(config, upgrade_receiver); + + if let Err(err) = launcher.run_loop(args.daemon_args).await { + error!("the daemon could not continue running: {err}"); + } else { + info!("the daemon has finished execution"); + } + + if !watcher_handle.is_finished() { + watcher_handle.abort(); + } + + if !upstream_poller_handle.is_finished() { + upstream_poller_handle.abort(); + } + + // TODO: add timeouts and error handling here + // TODO2: maybe we need to make those fuse futures? + watcher_handle.await; + upstream_poller_handle.await; + + Ok(()) + }) } diff --git a/tools/nymvisor/src/config/mod.rs b/tools/nymvisor/src/config/mod.rs index d68e0312a8..59572117ad 100644 --- a/tools/nymvisor/src/config/mod.rs +++ b/tools/nymvisor/src/config/mod.rs @@ -476,6 +476,8 @@ pub struct DaemonDebug { /// default: 10s /// Can be overridden with $DAEMON_SHUTDOWN_GRACE_PERIOD environmental variable. #[serde(with = "humantime_serde")] + // this is not deprecated, im just marking it as such so that clippy would yell at me because I still havent implemented it + #[deprecated] pub shutdown_grace_period: Duration, /// Set custom backup directory for daemon data. If not set, the daemon's home directory will be used instead. @@ -486,6 +488,8 @@ pub struct DaemonDebug { /// If enabled, `nymvisor` will perform upgrades directly without performing any backups. /// default: false /// Can be overridden with $DAEMON_UNSAFE_SKIP_BACKUP environmental variable. + // this is not deprecated, im just marking it as such so that clippy would yell at me because I still havent implemented it + #[deprecated] pub unsafe_skip_backup: bool, } diff --git a/tools/nymvisor/src/error.rs b/tools/nymvisor/src/error.rs index aeee1e03e8..6bd16172c3 100644 --- a/tools/nymvisor/src/error.rs +++ b/tools/nymvisor/src/error.rs @@ -309,6 +309,12 @@ pub(crate) enum NymvisorError { #[source] source: io::Error, }, + + #[error("the daemon has reached the maximum number of startup failures ({failures})")] + DaemonMaximumStartupFailures { failures: usize }, + + #[error("the daemon restart on failure is disabled")] + DisabledRestartOnFailure, } impl From for NymvisorError { diff --git a/tools/nymvisor/src/tasks/launcher/backup.rs b/tools/nymvisor/src/tasks/launcher/backup.rs index 494939d30d..ead125f6b4 100644 --- a/tools/nymvisor/src/tasks/launcher/backup.rs +++ b/tools/nymvisor/src/tasks/launcher/backup.rs @@ -83,6 +83,15 @@ impl BackupBuilder { } } + fn finish(mut self) -> Result<(), NymvisorError> { + self.tar_builder + .finish() + .map_err(|source| NymvisorError::BackupTarFinalizationFailure { + path: self.backup_filepath, + source, + }) + } + pub(crate) fn backup_daemon_home>( mut self, daemon_home: P, @@ -105,13 +114,4 @@ impl BackupBuilder { } self.finish() } - - fn finish(mut self) -> Result<(), NymvisorError> { - self.tar_builder - .finish() - .map_err(|source| NymvisorError::BackupTarFinalizationFailure { - path: self.backup_filepath, - source, - }) - } } diff --git a/tools/nymvisor/src/tasks/launcher/mod.rs b/tools/nymvisor/src/tasks/launcher/mod.rs index af355fcea3..455881a7cc 100644 --- a/tools/nymvisor/src/tasks/launcher/mod.rs +++ b/tools/nymvisor/src/tasks/launcher/mod.rs @@ -27,13 +27,67 @@ pub(crate) struct DaemonLauncher { } impl DaemonLauncher { - pub(crate) fn new(config: Config) -> Self { - todo!() + pub(crate) fn new(config: Config, upgrade_plan_watcher: FileWatcherEventReceiver) -> Self { + DaemonLauncher { + config, + upgrade_plan_watcher, + } + } + + pub(crate) async fn run_loop(&mut self, args: Vec) -> Result<(), NymvisorError> { + let mut consecutive_startup_failure_count = 0; + loop { + let run_start = tokio::time::Instant::now(); + + let res = self.run_and_upgrade(args.clone()).await; + let run_duration = run_start.elapsed(); + info!( + "the daemon has run for {}", + humantime::format_duration(run_duration) + ); + + match res { + Ok(upgraded) => { + if upgraded { + if !self.config.daemon.debug.restart_after_upgrade { + info!("upgrade detected, DAEMON_RESTART_AFTER_UPGRADE is off. Verify new upgrade and start nymvisor again"); + return Ok(()); + } + // restart + } else { + return Ok(()); + } + } + Err(failure) => { + error!("daemon failed with the following error: {failure}"); + + if !self.config.daemon.debug.restart_on_failure { + return Err(NymvisorError::DisabledRestartOnFailure); + } + + if run_duration < self.config.daemon.debug.startup_period_duration { + consecutive_startup_failure_count += 1; + } else { + consecutive_startup_failure_count = 1; + } + + if consecutive_startup_failure_count + >= self.config.daemon.debug.max_startup_failures + { + return Err(NymvisorError::DaemonMaximumStartupFailures { + failures: consecutive_startup_failure_count, + }); + } + + sleep(self.config.daemon.debug.failure_restart_delay).await; + } + } + } } /// the full upgrade process process, i.e. run until upgrade, do backup and perform the upgrade. - /// returns a boolean indicating whether the process should get restarted - pub(crate) async fn run(&mut self, args: Vec) -> Result { + /// returns a boolean indicating whether an upgrade has been performed + async fn run_and_upgrade(&mut self, args: Vec) -> Result { let upgrade_available = self.wait_for_upgrade_or_termination(args.clone()).await?; if !upgrade_available { return Ok(false); diff --git a/tools/nymvisor/src/tasks/upgrade_plan_watcher.rs b/tools/nymvisor/src/tasks/upgrade_plan_watcher.rs index 7052b1f493..7ca896946b 100644 --- a/tools/nymvisor/src/tasks/upgrade_plan_watcher.rs +++ b/tools/nymvisor/src/tasks/upgrade_plan_watcher.rs @@ -6,6 +6,7 @@ use crate::error::NymvisorError; use async_file_watcher::{AsyncFileWatcher, FileWatcherEventReceiver, NotifyResult}; use futures::channel::mpsc; use tokio::task::JoinHandle; +use tracing::warn; pub(crate) fn start_upgrade_plan_watcher( config: &Config, @@ -14,7 +15,11 @@ pub(crate) fn start_upgrade_plan_watcher( let mut watcher = AsyncFileWatcher::new_file_changes_watcher(config.upgrade_plan_filepath(), events_sender)?; - let join_handle = tokio::spawn(async move { watcher.watch().await }); + let join_handle = tokio::spawn(async move { + let res = watcher.watch().await; + warn!("the upgrade plan watcher has stopped"); + res + }); Ok((events_receiver, join_handle)) } diff --git a/tools/nymvisor/src/tasks/upstream_poller.rs b/tools/nymvisor/src/tasks/upstream_poller.rs index 41fe1ee7ed..cfe06e065e 100644 --- a/tools/nymvisor/src/tasks/upstream_poller.rs +++ b/tools/nymvisor/src/tasks/upstream_poller.rs @@ -70,7 +70,7 @@ impl UpstreamPoller { } } - pub(crate) async fn start(mut self) -> JoinHandle<()> { + pub(crate) fn start(mut self) -> JoinHandle<()> { tokio::spawn(async move { self.run().await }) } } diff --git a/tools/nymvisor/src/upgrades/mod.rs b/tools/nymvisor/src/upgrades/mod.rs index ecdcf8fd5c..6c149382e1 100644 --- a/tools/nymvisor/src/upgrades/mod.rs +++ b/tools/nymvisor/src/upgrades/mod.rs @@ -25,6 +25,11 @@ pub(crate) async fn upgrade_binary(config: &Config) -> Result<(), NymvisorError> return Err(NymvisorError::NoQueuedUpgrades); }; + if next.manual { + let unused_variable = 42; + todo!() + } + let upgrade_name = next.name.clone(); let history_path = config.upgrade_history_filepath(); @@ -64,6 +69,9 @@ pub(crate) async fn upgrade_binary(config: &Config) -> Result<(), NymvisorError> upgrade_binary_path.display() ); download_upgrade_binary(config, &next).await?; + + let unused_variable = 42; + // TODO: checksum, etc. } let tmp_daemon = Daemon::new(upgrade_binary_path);