Robot
/
watchexec
mirror of https://github.com/watchexec/watchexec.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

More logging for CLI and Supervisor (#709)

This commit is contained in:
Félix Saparelli 2023-12-09 22:10:11 +13:00 committed by GitHub
parent 44d794c921
commit 03460a6181
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 321 additions and 190 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
crates/cli/src/config.rs
View File

@ -15,7 +15,7 @@ use miette::{miette, IntoDiagnostic, Report, Result};
use notify_rust::Notification;
use termcolor::{Color, ColorChoice, ColorSpec, StandardStream, WriteColor};
use tokio::{process::Command as TokioCommand, time::sleep};
use tracing::{debug, debug_span, error};
use tracing::{debug, debug_span, error, instrument, trace, trace_span, Instrument};
use watchexec::{
command::{Command, Program, Shell, SpawnOptions},
error::RuntimeError,
@ -32,7 +32,7 @@ use crate::{
};
use crate::{emits::events_to_simple_format, state::State};
#[derive(Clone, Copy)]
#[derive(Clone, Copy, Debug)]
struct OutputFlags {
quiet: bool,
colour: ColorChoice,
@ -189,12 +189,16 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
let command = command.clone();
let emit_file = emit_file.clone();
let workdir = workdir.clone();
Box::new(async move {
Box::new(
async move {
trace!(events=?action.events, "handling action");
let add_envs = add_envs.clone();
let command = command.clone();
let emit_file = emit_file.clone();
let workdir = workdir.clone();
trace!("set spawn hook for workdir and environment variables");
let job = action.get_or_create_job(id, move || command.clone());
let events = action.events.clone();
job.set_spawn_hook(move |command, _| {
@ -212,6 +216,7 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
let show_events = || {
if print_events {
trace!("print events to stderr");
for (n, event) in action.events.iter().enumerate() {
eprintln!("[EVENT {n}] {event}");
}
@ -219,6 +224,7 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
};
if once {
debug!("debug mode: run once and quit");
show_events();
if let Some(delay) = delay_run {
@ -247,6 +253,7 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
}
let signals: Vec<Signal> = action.signals().collect();
trace!(?signals, "received some signals");
// if we got a terminate or interrupt signal, quit
if signals.contains(&Signal::Terminate) || signals.contains(&Signal::Interrupt) {
@ -265,6 +272,7 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
match mode {
ClearMode::Clear => {
clearscreen::clear().ok();
debug!("cleared screen");
}
ClearMode::Reset => {
for cs in [
@ -276,6 +284,7 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
] {
cs.clear().ok();
}
debug!("hard-reset screen");
}
}
}
@ -283,6 +292,7 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
show_events();
if let Some(delay) = delay_run {
trace!("delaying run by sleeping inside the job");
job.run_async(move |_| {
Box::new(async move {
sleep(delay).await;
@ -290,6 +300,7 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
});
}
trace!("querying job state via run_async");
job.run_async({
let job = job.clone();
move |context| {
@ -298,6 +309,7 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
Box::new(async move {
let innerjob = job.clone();
if is_running {
trace!(?on_busy, "job is running, decide what to do");
match on_busy {
OnBusyUpdate::DoNothing => {}
OnBusyUpdate::Signal => {
@ -346,9 +358,14 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
}
}
} else {
trace!("job is not running, start it");
job.start();
job.run(move |context| {
setup_process(innerjob.clone(), context.command.clone(), outflags)
setup_process(
innerjob.clone(),
context.command.clone(),
outflags,
)
});
}
})
@ -356,12 +373,15 @@ pub fn make_config(args: &Args, state: &State) -> Result<Config> {
});
action
})
}
.instrument(trace_span!("action handler")),
)
});
Ok(config)
}
#[instrument(level = "debug")]
fn interpret_command_args(args: &Args) -> Result<Arc<Command>> {
let mut cmd = args.command.clone();
if cmd.is_empty() {
@ -422,6 +442,7 @@ fn interpret_command_args(args: &Args) -> Result<Arc<Command>> {
}))
}
#[instrument(level = "trace")]
fn setup_process(job: Job, command: Arc<Command>, outflags: OutputFlags) {
if outflags.toast {
Notification::new()
@ -452,6 +473,7 @@ fn setup_process(job: Job, command: Arc<Command>, outflags: OutputFlags) {
});
}
#[instrument(level = "trace")]
fn end_of_process(state: &CommandState, outflags: OutputFlags) {
let CommandState::Finished {
status,
@ -509,6 +531,7 @@ fn end_of_process(state: &CommandState, outflags: OutputFlags) {
}
}
#[instrument(level = "trace")]
fn emit_events_to_command(
command: &mut TokioCommand,
events: Arc<[Event]>,

2
crates/supervisor/CHANGELOG.md
View File

@ -2,6 +2,8 @@
## Next (YYYY-MM-DD)
- Add `trace`-level logging to Job task.
## v1.0.1 (2023-11-29)
- Deps: watchexec-events 2.0.1

5
crates/supervisor/src/job/state.rs
View File

@ -3,6 +3,7 @@ use std::{sync::Arc, time::Instant};
#[cfg(not(test))]
use command_group::{tokio::ErasedChild, AsyncCommandGroup};
use tokio::process::Command as TokioCommand;
use tracing::trace;
use watchexec_events::ProcessEnd;
use crate::command::Command;
@ -74,9 +75,12 @@ impl CommandState {
mut spawnable: TokioCommand,
) -> std::io::Result<bool> {
if let Self::Running { .. } = self {
trace!("command running, not spawning again");
return Ok(false);
}
trace!(?command, "spawning command");
#[cfg(test)]
let child = super::TestChild::new(command)?;
@ -96,6 +100,7 @@ impl CommandState {
#[must_use]
pub(crate) fn reset(&mut self) -> Self {
trace!(?self, "resetting command state");
match self {
Self::Pending => Self::Pending,
Self::Finished {

471
crates/supervisor/src/job/task.rs
View File

@ -1,6 +1,7 @@
use std::{future::Future, mem::take, sync::Arc, time::Instant};
use tokio::{process::Command as TokioCommand, select, task::JoinHandle};
use tracing::{instrument, trace, trace_span, Instrument};
use watchexec_signals::Signal;
use crate::{
@ -21,6 +22,7 @@ use super::{
///
/// The job task immediately starts in the background: it does not need polling.
#[must_use]
#[instrument(level = "trace")]
pub fn start_job(command: Arc<Command>) -> (Job, JoinHandle<()>) {
let (sender, mut receiver) = priority::new();
@ -42,121 +44,100 @@ pub fn start_job(command: Arc<Command>) -> (Job, JoinHandle<()>) {
let mut on_end: Vec<Flag> = Vec::new();
let mut on_end_restart: Option<Flag> = None;
enum Loop {
Normally,
Skip,
Break,
}
'main: loop {
select! {
result = command_state.wait(), if command_state.is_running() => {
#[cfg(test)] eprintln!("[{:?}] waited: {result:?}", Instant::now());
trace!(?result, ?command_state, "got wait result");
match async {
#[cfg(test)] eprintln!("[{:?}] waited: {result:?}", Instant::now());
match result {
Err(err) => {
let fut = error_handler.call(sync_io_error(err));
fut.await;
match result {
Err(err) => {
let fut = error_handler.call(sync_io_error(err));
fut.await;
return Loop::Skip;
}
Ok(true) => {
trace!(existing=?stop_timer, "erasing stop timer");
stop_timer = None;
trace!(count=%on_end.len(), "raising all pending end flags");
for done in take(&mut on_end).into_iter() {
done.raise();
}
if let Some(flag) = on_end_restart.take() {
trace!("continuing a graceful restart");
let mut spawnable = command.to_spawnable();
previous_run = Some(command_state.reset());
spawn_hook
.call(
&mut spawnable,
&JobTaskContext {
command: command.clone(),
current: &command_state,
previous: previous_run.as_ref(),
},
)
.await;
if let Err(err) = command_state.spawn(command.clone(), spawnable).await {
let fut = error_handler.call(sync_io_error(err));
fut.await;
return Loop::Skip;
}
trace!("raising graceful restart's flag");
flag.raise();
}
}
Ok(false) => {
trace!("child wasn't running, ignoring wait result");
}
}
Loop::Normally
}.instrument(trace_span!("handle wait result")).await {
Loop::Normally => {}
Loop::Skip => {
trace!("skipping to next event");
continue 'main;
}
Ok(true) => {
stop_timer = None;
for done in take(&mut on_end).into_iter() {
done.raise();
}
if let Some(flag) = on_end_restart.take() {
let mut spawnable = command.to_spawnable();
previous_run = Some(command_state.reset());
spawn_hook
.call(
&mut spawnable,
&JobTaskContext {
command: command.clone(),
current: &command_state,
previous: previous_run.as_ref(),
},
)
.await;
if let Err(err) = command_state.spawn(command.clone(), spawnable).await {
let fut = error_handler.call(sync_io_error(err));
fut.await;
continue 'main;
}
flag.raise();
}
Loop::Break => {
trace!("breaking out of main loop");
break 'main;
}
Ok(false) => {}
}
}
Some(ControlMessage { control, done }) = receiver.recv(&mut stop_timer) => {
macro_rules! try_with_handler {
($erroring:expr) => {
match $erroring {
Err(err) => {
let fut = error_handler.call(sync_io_error(err));
fut.await;
done.raise();
continue 'main;
match async {
trace!(?control, ?command_state, "got control message");
#[cfg(test)] eprintln!("[{:?}] control: {control:?}", Instant::now());
macro_rules! try_with_handler {
($erroring:expr) => {
match $erroring {
Err(err) => {
let fut = error_handler.call(sync_io_error(err));
fut.await;
trace!("raising done flag for this control after error");
done.raise();
return Loop::Normally;
}
Ok(value) => value,
}
Ok(value) => value,
}
};
}
#[cfg(test)] eprintln!("[{:?}] control: {control:?}", Instant::now());
match control {
Control::Start => {
let mut spawnable = command.to_spawnable();
previous_run = Some(command_state.reset());
spawn_hook
.call(
&mut spawnable,
&JobTaskContext {
command: command.clone(),
current: &command_state,
previous: previous_run.as_ref(),
},
)
.await;
try_with_handler!(command_state.spawn(command.clone(), spawnable).await);
};
}
Control::Stop => {
if let CommandState::Running { child, started, .. } = &mut command_state {
try_with_handler!(child.kill().await);
let status = try_with_handler!(child.wait().await);
command_state = CommandState::Finished {
status: status.into(),
started: *started,
finished: Instant::now(),
};
for done in take(&mut on_end).into_iter() {
done.raise();
}
}
}
Control::GracefulStop { signal, grace } => {
if let CommandState::Running { child, .. } = &mut command_state {
try_with_handler!(signal_child(signal, child).await);
stop_timer.replace(Timer::stop(grace, done));
continue 'main;
}
}
Control::TryRestart => {
if let CommandState::Running { child, started, .. } = &mut command_state {
try_with_handler!(child.kill().await);
let status = try_with_handler!(child.wait().await);
command_state = CommandState::Finished {
status: status.into(),
started: *started,
finished: Instant::now(),
};
previous_run = Some(command_state.reset());
for done in take(&mut on_end).into_iter() {
done.raise();
}
match control {
Control::Start => {
let mut spawnable = command.to_spawnable();
previous_run = Some(command_state.reset());
spawn_hook
.call(
&mut spawnable,
@ -169,104 +150,211 @@ pub fn start_job(command: Arc<Command>) -> (Job, JoinHandle<()>) {
.await;
try_with_handler!(command_state.spawn(command.clone(), spawnable).await);
}
}
Control::TryGracefulRestart { signal, grace } => {
if let CommandState::Running { child, .. } = &mut command_state {
try_with_handler!(signal_child(signal, child).await);
Control::Stop => {
if let CommandState::Running { child, started, .. } = &mut command_state {
trace!("stopping child");
try_with_handler!(child.kill().await);
trace!("waiting on child");
let status = try_with_handler!(child.wait().await);
stop_timer.replace(Timer::restart(grace, done.clone()));
on_end_restart = Some(done);
continue 'main;
trace!(?status, "got child end status");
command_state = CommandState::Finished {
status: status.into(),
started: *started,
finished: Instant::now(),
};
trace!(count=%on_end.len(), "raising all pending end flags");
for done in take(&mut on_end).into_iter() {
done.raise();
}
} else {
trace!("child isn't running, skip");
}
}
}
Control::ContinueTryGracefulRestart => {
if let CommandState::Running { child, started, .. } = &mut command_state {
try_with_handler!(child.kill().await);
let status = try_with_handler!(child.wait().await);
Control::GracefulStop { signal, grace } => {
if let CommandState::Running { child, .. } = &mut command_state {
try_with_handler!(signal_child(signal, child).await);
command_state = CommandState::Finished {
status: status.into(),
started: *started,
finished: Instant::now(),
};
trace!(?grace, "setting up graceful stop timer");
stop_timer.replace(Timer::stop(grace, done));
return Loop::Skip;
} else {
trace!("child isn't running, skip");
}
}
Control::TryRestart => {
if let CommandState::Running { child, started, .. } = &mut command_state {
trace!("stopping child");
try_with_handler!(child.kill().await);
trace!("waiting on child");
let status = try_with_handler!(child.wait().await);
for done in take(&mut on_end).into_iter() {
trace!(?status, "got child end status");
command_state = CommandState::Finished {
status: status.into(),
started: *started,
finished: Instant::now(),
};
previous_run = Some(command_state.reset());
trace!(count=%on_end.len(), "raising all pending end flags");
for done in take(&mut on_end).into_iter() {
done.raise();
}
let mut spawnable = command.to_spawnable();
spawn_hook
.call(
&mut spawnable,
&JobTaskContext {
command: command.clone(),
current: &command_state,
previous: previous_run.as_ref(),
},
)
.await;
try_with_handler!(command_state.spawn(command.clone(), spawnable).await);
} else {
trace!("child isn't running, skip");
}
}
Control::TryGracefulRestart { signal, grace } => {
if let CommandState::Running { child, .. } = &mut command_state {
try_with_handler!(signal_child(signal, child).await);
trace!(?grace, "setting up graceful stop timer");
stop_timer.replace(Timer::restart(grace, done.clone()));
trace!("setting up graceful restart flag");
on_end_restart = Some(done);
return Loop::Skip;
} else {
trace!("child isn't running, skip");
}
}
Control::ContinueTryGracefulRestart => {
trace!("continuing a graceful try-restart");
if let CommandState::Running { child, started, .. } = &mut command_state {
trace!("stopping child forcefully");
try_with_handler!(child.kill().await);
trace!("waiting on child");
let status = try_with_handler!(child.wait().await);
trace!(?status, "got child end status");
command_state = CommandState::Finished {
status: status.into(),
started: *started,
finished: Instant::now(),
};
trace!(count=%on_end.len(), "raising all pending end flags");
for done in take(&mut on_end).into_iter() {
done.raise();
}
}
let mut spawnable = command.to_spawnable();
previous_run = Some(command_state.reset());
spawn_hook
.call(
&mut spawnable,
&JobTaskContext {
command: command.clone(),
current: &command_state,
previous: previous_run.as_ref(),
},
)
.await;
try_with_handler!(command_state.spawn(command.clone(), spawnable).await);
}
Control::Signal(signal) => {
if let CommandState::Running { child, .. } = &mut command_state {
try_with_handler!(signal_child(signal, child).await);
} else {
trace!("child isn't running, skip");
}
}
Control::Delete => {
trace!("raising done flag immediately");
done.raise();
return Loop::Break;
}
Control::NextEnding => {
if !matches!(command_state, CommandState::Finished { .. }) {
trace!("queue end flag");
on_end.push(done);
return Loop::Skip;
} else {
trace!("child is finished, raise done flag immediately");
done.raise();
return Loop::Normally;
}
}
let mut spawnable = command.to_spawnable();
previous_run = Some(command_state.reset());
spawn_hook
.call(
&mut spawnable,
&JobTaskContext {
command: command.clone(),
current: &command_state,
previous: previous_run.as_ref(),
},
)
Control::SyncFunc(f) => {
f(&JobTaskContext {
command: command.clone(),
current: &command_state,
previous: previous_run.as_ref(),
});
}
Control::AsyncFunc(f) => {
Box::into_pin(f(&JobTaskContext {
command: command.clone(),
current: &command_state,
previous: previous_run.as_ref(),
}))
.await;
try_with_handler!(command_state.spawn(command.clone(), spawnable).await);
}
Control::Signal(signal) => {
if let CommandState::Running { child, .. } = &mut command_state {
try_with_handler!(signal_child(signal, child).await);
}
}
Control::Delete => {
done.raise();
break 'main;
}
Control::NextEnding => {
if !matches!(command_state, CommandState::Finished { .. }) {
on_end.push(done);
continue 'main;
Control::SetSyncErrorHandler(f) => {
trace!("setting sync error handler");
error_handler = ErrorHandler::Sync(f);
}
Control::SetAsyncErrorHandler(f) => {
trace!("setting async error handler");
error_handler = ErrorHandler::Async(f);
}
Control::UnsetErrorHandler => {
trace!("unsetting error handler");
error_handler = ErrorHandler::None;
}
Control::SetSyncSpawnHook(f) => {
trace!("setting sync spawn hook");
spawn_hook = SpawnHook::Sync(f);
}
Control::SetAsyncSpawnHook(f) => {
trace!("setting async spawn hook");
spawn_hook = SpawnHook::Async(f);
}
Control::UnsetSpawnHook => {
trace!("unsetting spawn hook");
spawn_hook = SpawnHook::None;
}
}
Control::SyncFunc(f) => {
f(&JobTaskContext {
command: command.clone(),
current: &command_state,
previous: previous_run.as_ref(),
});
}
Control::AsyncFunc(f) => {
Box::into_pin(f(&JobTaskContext {
command: command.clone(),
current: &command_state,
previous: previous_run.as_ref(),
}))
.await;
}
trace!("raising control done flag");
done.raise();
Control::SetSyncErrorHandler(f) => {
error_handler = ErrorHandler::Sync(f);
}
Control::SetAsyncErrorHandler(f) => {
error_handler = ErrorHandler::Async(f);
}
Control::UnsetErrorHandler => {
error_handler = ErrorHandler::None;
}
Control::SetSyncSpawnHook(f) => {
spawn_hook = SpawnHook::Sync(f);
}
Control::SetAsyncSpawnHook(f) => {
spawn_hook = SpawnHook::Async(f);
}
Control::UnsetSpawnHook => {
spawn_hook = SpawnHook::None;
}
Loop::Normally
}.instrument(trace_span!("handle control message")).await {
Loop::Normally => {}
Loop::Skip => {
trace!("skipping to next event (without raising done flag)");
continue 'main;
}
Loop::Break => {
trace!("breaking out of main loop");
break 'main;
}
done.raise();
}
}
}
}
trace!("raising job done flag");
done.raise();
}),
)
@ -281,11 +369,18 @@ macro_rules! sync_async_callbox {
}
impl $name {
#[instrument(level = "trace", skip(self, $($argname),*))]
pub async fn call(&self, $($argname: $argtype),*) {
match self {
$name::None => (),
$name::Sync(f) => f($($argname),*),
$name::Async(f) => Box::into_pin(f($($argname),*)).await,
$name::Sync(f) => {
::tracing::trace!("calling sync {:?}", stringify!($name));
f($($argname),*)
}
$name::Async(f) => {
::tracing::trace!("calling async {:?}", stringify!($name));
Box::into_pin(f($($argname),*)).await
}
}
}
}
@ -334,22 +429,28 @@ pub type AsyncErrorHandler = Arc<
sync_async_callbox!(ErrorHandler, SyncErrorHandler, AsyncErrorHandler, (error: SyncIoError));
#[cfg_attr(not(windows), allow(clippy::needless_pass_by_ref_mut))] // needed for start_kill()
#[instrument(level = "trace")]
async fn signal_child(
signal: Signal,
#[cfg(test)] child: &mut super::TestChild,
#[cfg(not(test))] child: &mut command_group::tokio::ErasedChild,
) -> std::io::Result<()> {
#[cfg(unix)]
child.signal(
signal
{
let sig = signal
.to_nix()
.or_else(|| Signal::Terminate.to_nix())
.expect("UNWRAP: guaranteed for Signal::Terminate default"),
)?;
.expect("UNWRAP: guaranteed for Signal::Terminate default");
trace!(signal=?sig, "sending signal");
child.signal(sig)?;
}
#[cfg(windows)]
if signal == Signal::ForceStop {
trace!("starting kill, without waiting");
child.start_kill()?;
} else {
trace!(?signal, "ignoring unsupported signal");
}
Ok(())