GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

basic draft of all tasks

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2023-11-07 17:47:51 +00:00
parent 70d3b784f4
commit 2e077ca946
17 changed files with 423 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -7609,6 +7609,8 @@ dependencies = [
"nix 0.27.1",
"nym-bin-common",
"nym-config",
"nym-task",
"reqwest",
"serde",
"serde_json",
"thiserror",
common/task/src/signal.rs
-1
View File
@@ -1,7 +1,6 @@
use crate::{manager::SentError, TaskManager};
#[cfg(unix)]
#[deprecated]
pub async fn wait_for_signal() {
use tokio::signal::unix::{signal, SignalKind};
let mut sigterm = signal(SignalKind::terminate()).expect("Failed to setup SIGTERM channel");
tools/nymvisor/Cargo.toml
+5
View File
@@ -20,6 +20,7 @@ humantime = "2.1.0"
humantime-serde = "1.1.1"
lazy_static = { workspace = true }
nix = { version = "0.27.1", features = ["signal"] }
reqwest = { workspace = true, features = ["json"] }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
time = { workspace = true, features = [ "serde-human-readable" ] }
@@ -31,3 +32,7 @@ url = { workspace = true, features = ["serde"] }
async-file-watcher = { path = "../../common/async-file-watcher" }
nym-bin-common = { path = "../../common/bin-common", features = ["output_format", "basic_tracing"] }
nym-config = { path = "../../common/config" }
nym-task = { path = "../../common/task"}
[dev-dependencies]
tokio = { workspace = true, features = ["full"] }
tools/nymvisor/src/cli/init.rs
+13 -4
View File
@@ -32,6 +32,12 @@ pub(crate) struct Args {
#[arg(long)]
upstream_base_upgrade_url: Option<Url>,
/// Specifies the rate of polling the upstream url for upgrade information.
/// default: 1h
/// Can be overridden with $NYMVISOR_UPSTREAM_POLLING_RATE
#[arg(long, value_parser = humantime::parse_duration)]
upstream_polling_rate: Option<Duration>,
/// If enabled, this will disable `nymvisor` logs (but not the underlying process)
/// Can be overridden with $NYMVISOR_DISABLE_LOGS environmental variable.
#[arg(long)]
@@ -127,6 +133,9 @@ impl Args {
if let Some(upstream) = &self.upstream_base_upgrade_url {
config.nymvisor.debug.upstream_base_upgrade_url = upstream.clone()
}
if let Some(polling_rate) = self.upstream_polling_rate {
config.nymvisor.debug.upstream_polling_rate = polling_rate
}
if self.disable_nymvisor_logs {
config.nymvisor.debug.disable_logs = self.disable_nymvisor_logs;
}
@@ -352,22 +361,22 @@ fn setup_initial_upgrade_plan(
let existing_plan = UpgradePlan::try_load(&plan_path)?;
if let (Some(current_info), Some(existing_info)) = (
&genesis_info.binary_details,
&existing_plan.current.binary_details,
&existing_plan.current().binary_details,
) {
if current_info != existing_info {
// if possible, compare the actual full details
return Err(NymvisorError::PreexistingUpgradePlan {
path: plan_path,
current_name: genesis_info.name,
existing_name: existing_plan.current.name,
existing_name: existing_plan.current().name.clone(),
});
}
} else if genesis_info.name != existing_plan.current.name {
} else if genesis_info.name != existing_plan.current().name {
// otherwise just check the upgrade name
return Err(NymvisorError::PreexistingUpgradePlan {
path: plan_path,
current_name: genesis_info.name,
existing_name: existing_plan.current.name,
existing_name: existing_plan.current().name.clone(),
});
}
tools/nymvisor/src/cli/run.rs
+41 -35
View File
@@ -41,40 +41,46 @@ pub(crate) fn execute(args: Args) -> Result<(), NymvisorError> {
.build()
.expect("failed to create the runtime");
// spawn the root task
rt.block_on(async {
println!("run");
// we have three tasks only:
// - one for managing the daemon launcher
// - the other one for watching the upgrade plan file
// - 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 and terminate the nymvisor
let daemon = Daemon::from_config(&config);
let interrupt_notify = Arc::new(Notify::new());
let running = daemon.execute_async(args.daemon_args, Arc::clone(&interrupt_notify))?;
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);
<Result<_, NymvisorError>>::Ok(())
})
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);
//
// <Result<_, NymvisorError>>::Ok(())
// })
}
tools/nymvisor/src/config/mod.rs
+18 -6
View File
@@ -21,6 +21,7 @@ pub(crate) const DEFAULT_FAILURE_RESTART_DELAY: Duration = Duration::from_secs(1
pub(crate) const DEFAULT_STARTUP_PERIOD: Duration = Duration::from_secs(120);
pub(crate) const DEFAULT_MAX_STARTUP_FAILURES: usize = 10;
pub(crate) const DEFAULT_SHUTDOWN_GRACE_PERIOD: Duration = Duration::from_secs(10);
pub(crate) const DEFAULT_UPSTREAM_POLLING_RATE: Duration = Duration::from_secs(60 * 60);
pub(crate) const DEFAULT_BASE_UPSTREAM_UPGRADE_INFO_SOURCE: &str =
"https://nymtech.net/.wellknown/";
@@ -73,7 +74,7 @@ pub fn default_global_data_directory() -> PathBuf {
.join(DEFAULT_DATA_DIR)
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)]
#[serde(deny_unknown_fields)]
pub struct Config {
// additional metadata holding on-disk location of this config file
@@ -273,6 +274,11 @@ impl Config {
.join(&self.daemon.name)
}
// e.g. $HOME/.nym/nymvisors/data/nym-api/current/upgrade-info.json
pub fn current_upgrade_info_filepath(&self) -> PathBuf {
self.current_daemon_dir().join(UPGRADE_INFO_FILENAME)
}
// e.g. $HOME/.nym/nymvisors/data/nym-api/upgrades/<upgrade-name>/upgrade-info.json
// or $HOME/.nym/nymvisors/data/nym-api/genesis/upgrade-info.json
pub fn upgrade_info_filepath<S: AsRef<str>>(&self, upgrade_name: S) -> PathBuf {
@@ -307,7 +313,7 @@ impl Config {
}
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)]
#[serde(deny_unknown_fields)]
pub struct Nymvisor {
/// ID specifies the human readable ID of this particular nymvisor instance.
@@ -319,7 +325,7 @@ pub struct Nymvisor {
pub debug: NymvisorDebug,
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)]
#[serde(deny_unknown_fields)]
pub struct NymvisorDebug {
/// Sets the base url of the upstream source for obtaining upgrade information for the deaemon.
@@ -328,6 +334,12 @@ pub struct NymvisorDebug {
/// Can be overridden with $NYMVISOR_UPSTREAM_BASE_UPGRADE_URL environmental variable.
pub upstream_base_upgrade_url: Url,
/// Specifies the rate of polling the upstream url for upgrade information.
/// default: 1h
/// Can be overridden with $NYMVISOR_UPSTREAM_POLLING_RATE
#[serde(with = "humantime_serde")]
pub upstream_polling_rate: Duration,
/// If set to true, this will disable `nymvisor` logs (but not the underlying process)
/// default: false
/// Can be overridden with $NYMVISOR_DISABLE_LOGS environmental variable.
@@ -340,7 +352,6 @@ pub struct NymvisorDebug {
pub upgrade_data_directory: Option<PathBuf>,
}
#[allow(clippy::derivable_impls)]
impl Default for NymvisorDebug {
fn default() -> Self {
NymvisorDebug {
@@ -349,13 +360,14 @@ impl Default for NymvisorDebug {
upstream_base_upgrade_url: DEFAULT_BASE_UPSTREAM_UPGRADE_INFO_SOURCE
.parse()
.expect("default upstream url was malformed"),
upstream_polling_rate: DEFAULT_UPSTREAM_POLLING_RATE,
disable_logs: false,
upgrade_data_directory: None,
}
}
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)]
#[serde(deny_unknown_fields)]
pub struct Daemon {
/// The name of the managed binary itself (e.g. nym-api, nym-mixnode, nym-gateway, etc.)
@@ -373,7 +385,7 @@ pub struct Daemon {
pub debug: DaemonDebug,
}
#[derive(Debug, Deserialize, PartialEq, Serialize)]
#[derive(Debug, Deserialize, PartialEq, Serialize, Clone)]
#[serde(deny_unknown_fields)]
pub struct DaemonDebug {
/// Override url to the upstream source for upgrade plans for this daeamon.
tools/nymvisor/src/daemon/mod.rs
+33 -11
View File
@@ -17,7 +17,21 @@ use std::task::{Context, Poll};
use std::time::Duration;
use tokio::sync::Notify;
use tokio::time::{sleep, Sleep};
use tracing::{debug, error, info, instrument, warn};
use tracing::{debug, info, instrument, warn};
pub(crate) struct InterruptHandle(Arc<Notify>);
impl InterruptHandle {
pub(crate) fn interrupt_daemon(&self) {
self.0.notify_one()
}
}
impl Drop for InterruptHandle {
fn drop(&mut self) {
self.interrupt_daemon();
}
}
#[derive(Debug)]
pub(crate) struct Daemon {
@@ -76,11 +90,7 @@ impl Daemon {
todo!()
}
pub(crate) fn execute_async<I, S>(
&self,
args: I,
interrupt_handle: Arc<Notify>,
) -> Result<ExecutingDaemon, NymvisorError>
pub(crate) fn execute_async<I, S>(&self, args: I) -> Result<ExecutingDaemon, NymvisorError>
where
I: IntoIterator<Item = S>,
S: AsRef<OsStr>,
@@ -93,7 +103,7 @@ impl Daemon {
.spawn()
.map_err(|source| NymvisorError::DaemonIoFailure { source })?;
ExecutingDaemon::new(self.kill_timeout, interrupt_handle, child)
ExecutingDaemon::new(self.kill_timeout, child)
}
}
@@ -103,6 +113,7 @@ pub(crate) struct ExecutingDaemon {
child_id: i32,
kill_timeout_duration: Duration,
interrupt_sent: bool,
interrupt_handle: Option<Arc<Notify>>,
// interrupted: Option<Pin<Box<Notified<'static>>>>,
interrupted: Pin<Box<dyn Future<Output = ()> + Send + Sync>>,
@@ -114,15 +125,17 @@ pub(crate) struct ExecutingDaemon {
impl ExecutingDaemon {
fn new(
kill_timeout_duration: Duration,
interrupt_notify: Arc<Notify>,
mut child: tokio::process::Child,
) -> Result<ExecutingDaemon, NymvisorError> {
if let Some(id) = child.id() {
let interrupt_handle = Arc::new(Notify::new());
let notified_handle = Arc::clone(&interrupt_handle);
Ok(ExecutingDaemon {
child_id: id as i32,
kill_timeout_duration,
interrupt_sent: false,
interrupted: Box::pin(async move { interrupt_notify.notified().await }),
interrupt_handle: Some(interrupt_handle),
interrupted: Box::pin(async move { notified_handle.notified().await }),
kill_timeout: None,
child_future: Box::pin(async move { child.wait().await }),
})
@@ -137,6 +150,15 @@ impl ExecutingDaemon {
}
}
pub(crate) fn interrupt_handle(&mut self) -> InterruptHandle {
#[allow(clippy::expect_used)]
InterruptHandle(
self.interrupt_handle
.take()
.expect("the interrupt handle has already been obtained"),
)
}
fn signal_child(&self, signal: Signal) -> Result<(), NymvisorError> {
info!("sending {signal} to the daemon");
nix::sys::signal::kill(Pid::from_raw(self.child_id), signal)
@@ -145,13 +167,13 @@ impl ExecutingDaemon {
}
impl Future for ExecutingDaemon {
type Output = Result<Option<ExitStatus>, NymvisorError>;
type Output = Result<ExitStatus, NymvisorError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// 1. check if the child is done
if let Poll::Ready(result) = Pin::new(&mut self.child_future).poll(cx) {
return match result {
Ok(exit_status) => Poll::Ready(Ok(Some(exit_status))),
Ok(exit_status) => Poll::Ready(Ok(exit_status)),
Err(source) => Poll::Ready(Err(NymvisorError::DaemonIoFailure { source })),
};
}
tools/nymvisor/src/env.rs
+6
View File
@@ -15,6 +15,7 @@ pub mod vars {
pub const NYMVISOR_ID: &str = "NYMVISOR_ID";
pub const NYMVISOR_CONFIG_PATH: &str = "NYMVISOR_CONFIG_PATH";
pub const NYMVISOR_UPSTREAM_BASE_UPGRADE_URL: &str = "NYMVISOR_UPSTREAM_BASE_UPGRADE_URL";
pub const NYMVISOR_UPSTREAM_POLLING_RATE: &str = "NYMVISOR_UPSTREAM_POLLING_RATE";
pub const NYMVISOR_DISABLE_LOGS: &str = "NYMVISOR_DISABLE_LOGS";
pub const NYMVISOR_UPGRADE_DATA_DIRECTORY: &str = "NYMVISOR_UPGRADE_DATA_DIRECTORY";
@@ -45,6 +46,7 @@ pub(crate) struct Env {
pub(crate) nymvisor_id: Option<String>,
pub(crate) nymvisor_config_path: Option<PathBuf>,
pub(crate) nymvisor_upstream_base_upgrade_url: Option<Url>,
pub(crate) nymvisor_upstream_polling_rate: Option<Duration>,
pub(crate) nymvisor_disable_logs: Option<bool>,
pub(crate) nymvisor_upgrade_data_directory: Option<PathBuf>,
@@ -71,6 +73,9 @@ impl Env {
if let Some(upstream) = &self.nymvisor_upstream_base_upgrade_url {
config.nymvisor.debug.upstream_base_upgrade_url = upstream.clone()
}
if let Some(polling_rate) = self.nymvisor_upstream_polling_rate {
config.nymvisor.debug.upstream_polling_rate = polling_rate
}
if let Some(nymvisor_disable_logs) = self.nymvisor_disable_logs {
config.nymvisor.debug.disable_logs = nymvisor_disable_logs;
}
@@ -212,6 +217,7 @@ impl Env {
nymvisor_id: read_string(vars::NYMVISOR_ID)?,
nymvisor_config_path: read_pathbuf(vars::NYMVISOR_CONFIG_PATH)?,
nymvisor_upstream_base_upgrade_url: read_url(vars::NYMVISOR_UPSTREAM_BASE_UPGRADE_URL)?,
nymvisor_upstream_polling_rate: read_duration(vars::NYMVISOR_UPSTREAM_POLLING_RATE)?,
nymvisor_disable_logs: read_bool(vars::NYMVISOR_DISABLE_LOGS)?,
nymvisor_upgrade_data_directory: read_pathbuf(vars::NYMVISOR_UPGRADE_DATA_DIRECTORY)?,
daemon_name: read_string(vars::DAEMON_NAME)?,
tools/nymvisor/src/error.rs
+8
View File
@@ -10,6 +10,7 @@ use std::num::ParseIntError;
use std::path::PathBuf;
use std::process::ExitStatus;
use thiserror::Error;
use url::Url;
#[derive(Debug, Error)]
pub(crate) enum NymvisorError {
@@ -203,6 +204,13 @@ pub(crate) enum NymvisorError {
#[from]
source: NotifyError,
},
#[error("failed to query the upstream url ('{url}') - source")]
UpstreamQueryFailure {
url: Url,
#[source]
source: reqwest::Error,
},
}
impl From<ExitStatus> for NymvisorError {
tools/nymvisor/src/helpers.rs
+37
View File
@@ -0,0 +1,37 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use futures::future::Aborted;
use futures::stream::AbortHandle;
use std::future::Future;
use tokio::task::{JoinError, JoinHandle};
pub(crate) struct TaskHandle<T>
where
T: Send + 'static,
{
abort_handle: AbortHandle,
// join_handle: JoinHandle<F::Output>,
join_handle: JoinHandle<Result<T, Aborted>>,
}
impl<T> TaskHandle<T>
where
T: Send + 'static,
{
pub(crate) fn new(
abort_handle: AbortHandle,
join_handle: JoinHandle<Result<T, Aborted>>,
) -> Self {
TaskHandle {
abort_handle,
join_handle,
}
}
// TODO: change return type
pub(crate) async fn abort_and_finalize(self) -> Result<Result<T, Aborted>, JoinError> {
self.abort_handle.abort();
self.join_handle.await
}
}
tools/nymvisor/src/launcher/mod.rs
-32
View File
@@ -1,32 +0,0 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::config::Config;
pub(crate) struct DaemonLauncher {
config: Config,
}
impl DaemonLauncher {
pub(crate) fn new(config: Config) -> Self {
todo!()
}
// responsible for running until exit or until update is detected
pub(crate) fn run(&self, args: Vec<String>) {
/*
tokio select on:
- daemon terminating
- upgrade-plan.json changes
- https://nymtech.net/.wellknown/<DAEMON_NAME>/update-info.json changes
// todo: maybe move to a higher layer
- signals received (to propagate them to daemon before terminating to prevent creating zombie processes)
*/
todo!()
}
}
tools/nymvisor/src/main.rs
+2 -1
View File
@@ -12,7 +12,8 @@ pub(crate) mod config;
pub(crate) mod daemon;
pub(crate) mod env;
pub(crate) mod error;
pub(crate) mod launcher;
pub(crate) mod helpers;
pub(crate) mod tasks;
pub(crate) mod upgrades;
fn main() -> anyhow::Result<()> {
tools/nymvisor/src/tasks/launcher/mod.rs
+109
View File
@@ -0,0 +1,109 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::config::Config;
use crate::daemon::Daemon;
use crate::error::NymvisorError;
use crate::upgrades::{UpgradeInfo, UpgradePlan};
use async_file_watcher::FileWatcherEventReceiver;
use futures::future::{FusedFuture, OptionFuture};
use futures::{FutureExt, StreamExt};
use nym_task::signal::wait_for_signal;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;
use tokio::time::Sleep;
use tracing::{error, info, warn};
pub(crate) struct DaemonLauncher {
config: Config,
upgrade_plan_watcher: FileWatcherEventReceiver,
}
impl DaemonLauncher {
pub(crate) fn new(config: Config) -> Self {
todo!()
}
// responsible for running until exit or until update is detected
pub(crate) async fn run(&mut self, args: Vec<String>) -> Result<(), NymvisorError> {
todo!()
}
fn check_upgrade_plan_changes(&self) {
//
}
async fn wait_for_upgrade_or_termination(
&mut self,
args: Vec<String>,
) -> Result<bool, NymvisorError> {
let daemon = Daemon::from_config(&self.config);
let current_upgrade = UpgradeInfo::try_load(self.config.current_upgrade_info_filepath())?;
// see if there's already a queued up upgrade
let current_upgrade_plan = UpgradePlan::try_load(self.config.upgrade_plan_filepath())?;
let next = current_upgrade_plan.next_upgrade();
let mut running_daemon = daemon.execute_async(args)?;
let interrupt_handle = running_daemon.interrupt_handle();
// we need to fuse the daemon future so that we could check if it has already terminated
let mut fused_runner = running_daemon.fuse();
let mut upgrade_timeout: OptionFuture<_> = None.into();
upgrade_timeout =
Some(Box::pin(tokio::time::sleep(Duration::from_secs(123))).fuse()).into();
let sig_fut = wait_for_signal();
// note: this has to be in a loop because `upgrade_plan_watcher` might receive events that do not necessarily trigger the upgrade
loop {
tokio::select! {
daemon_res = &mut fused_runner => {
warn!("the daemon has terminated by itself - was it a short lived command?");
let exit_status = daemon_res?;
info!("it finished with the following exit status: {exit_status}");
return Ok(false)
}
_ = &mut self.upgrade_plan_watcher.next() => {
//
}
_ = &mut upgrade_timeout, if !upgrade_timeout.is_terminated() => {
break
}
}
}
// if the runner hasn't terminated by itself (which should be the almost every single time!)
// send the interrupt and wait for it to be done
if fused_runner.is_terminated() {
todo!("error case")
}
interrupt_handle.interrupt_daemon();
let res = fused_runner.await;
/*
tokio select on:
- daemon terminating
- upgrade-plan.json changes
- https://nymtech.net/.wellknown/<DAEMON_NAME>/update-info.json changes
// todo: maybe move to a higher layer
- signals received (to propagate them to daemon before terminating to prevent creating zombie processes)
*/
todo!()
}
async fn perform_backup(&self) {
todo!()
}
}
tools/nymvisor/src/tasks/mod.rs
+6
View File
@@ -0,0 +1,6 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub(crate) mod launcher;
pub(crate) mod upgrade_plan_watcher;
pub(crate) mod upstream_poller;
tools/nymvisor/src/tasks/upgrade_plan_watcher.rs
+25
View File
@@ -0,0 +1,25 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::config::Config;
use crate::error::NymvisorError;
use crate::helpers::TaskHandle;
use async_file_watcher::{AsyncFileWatcher, FileWatcherEventReceiver, NotifyResult};
use futures::channel::mpsc;
use futures::future::{AbortHandle, Abortable};
pub(crate) fn start_upgrade_plan_watcher(
config: &Config,
) -> Result<(FileWatcherEventReceiver, TaskHandle<NotifyResult<()>>), NymvisorError> {
let (events_sender, 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 join_handle =
tokio::spawn(async move { Abortable::new(watcher.watch(), abort_registration).await });
let task_handle = TaskHandle::new(abort_handle, join_handle);
Ok((events_receiver, task_handle))
}
tools/nymvisor/src/tasks/upstream_poller.rs
+81
View File
@@ -0,0 +1,81 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::config::Config;
use crate::error::NymvisorError;
use crate::helpers::TaskHandle;
use crate::upgrades::{UpgradeInfo, UpgradePlan};
use futures::future::{AbortHandle, Abortable};
use reqwest::get;
use tracing::{error, warn};
pub(crate) struct UpstreamPoller {
config: Config,
}
impl UpstreamPoller {
pub(crate) fn new(config: &Config) -> Self {
UpstreamPoller {
config: config.clone(),
}
}
/// Poll the upstream url to see if new upgrade has been published.
/// If so, save it to `upgrade-info.json` and update the `upgrade-plan.json`
async fn check_upstream(&self) -> Result<(), NymvisorError> {
let upgrade_info: UpgradeInfo = get(self.config.upstream_upgrade_url())
.await
.map_err(|source| NymvisorError::UpstreamQueryFailure {
url: self.config.upstream_upgrade_url(),
source,
})?
.json()
.await
.map_err(|source| NymvisorError::UpstreamQueryFailure {
url: self.config.upstream_upgrade_url(),
source,
})?;
let mut plan = UpgradePlan::try_load(self.config.upgrade_plan_filepath())?;
// if the current version is the same as the one announced by upstream, we're done
if upgrade_info.version == plan.current().version {
return Ok(());
}
if !plan.has_planned(&upgrade_info) {
if let Err(err) =
upgrade_info.save(self.config.upgrade_info_filepath(&upgrade_info.name))
{
error!("failed to save new upgrade info: {err}");
return Err(err);
}
if let Err(err) = plan.insert_new_upgrade(upgrade_info) {
error!("failed to insert new upgrade info into the current upgrade plan: {err}");
return Err(err);
}
}
Ok(())
}
pub(crate) async fn run(&mut self) {
let mut interval = tokio::time::interval(self.config.nymvisor.debug.upstream_polling_rate);
loop {
// note: first tick happens immediately
interval.tick().await;
if let Err(err) = self.check_upstream().await {
warn!("failed to check the upstream for new upgrade information: {err}. we will try to poll it again in {}", humantime::format_duration(interval.period()));
}
}
}
pub(crate) async fn start(mut self) -> TaskHandle<()> {
let (abort_handle, abort_registration) = AbortHandle::new_pair();
let join_handle =
tokio::spawn(async move { Abortable::new(self.run(), abort_registration).await });
TaskHandle::new(abort_handle, join_handle)
}
}
tools/nymvisor/src/upgrades/mod.rs
+37 -2
View File
@@ -1,6 +1,7 @@
// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::config::GENESIS_DIR;
use crate::error::NymvisorError;
use nym_bin_common::build_information::BinaryBuildInformationOwned;
use serde::{Deserialize, Serialize};
@@ -10,6 +11,7 @@ use std::fs::OpenOptions;
use std::path::{Path, PathBuf};
use std::{fs, io};
use time::OffsetDateTime;
use tracing::error;
use url::Url;
mod http_upstream;
@@ -22,9 +24,9 @@ pub struct UpgradePlan {
#[serde(skip)]
_save_path: Option<PathBuf>,
pub current: UpgradeInfo,
current: UpgradeInfo,
pub next: VecDeque<UpgradeInfo>,
next: VecDeque<UpgradeInfo>,
}
impl UpgradePlan {
@@ -73,10 +75,27 @@ impl UpgradePlan {
self.update_on_disk()
}
pub(crate) fn current(&self) -> &UpgradeInfo {
&self.current
}
pub(crate) fn next_upgrade(&self) -> Option<&UpgradeInfo> {
self.next.front()
}
pub(crate) fn has_planned(&self, upgrade: &UpgradeInfo) -> bool {
for planned in &self.next {
if planned.version == upgrade.version {
if planned.name != upgrade.name {
// TODO: should we maybe return a hard error here instead?
error!("we have already a planned upgrade for version {} under name '{}' which differs from provided '{}'", planned.version, planned.name, upgrade.name);
}
return true;
}
}
false
}
// pub(crate) fn update_current(&mut self) -> Result<(), NymvisorError> {
//
// }
@@ -171,6 +190,18 @@ pub struct UpgradeInfo {
impl UpgradeInfo {
pub(crate) fn save<P: AsRef<Path>>(&self, path: P) -> Result<(), NymvisorError> {
let path = path.as_ref();
// in case we're saving brand new upgrade info, make sure the parent directory exists
#[allow(clippy::expect_used)]
let parent = path
.parent()
.expect("attempted to save the upgrade info as the root of the fs");
fs::create_dir_all(parent).map_err(|source| NymvisorError::PathInitFailure {
path: parent.to_path_buf(),
source,
})?;
let file = OpenOptions::new()
.create(true)
.write(true)
@@ -201,6 +232,10 @@ impl UpgradeInfo {
source,
})
}
pub(crate) fn is_genesis(&self) -> bool {
self.name == GENESIS_DIR
}
}
#[derive(Serialize, Deserialize, Debug)]