add a "kind" field to log messages, other fields — radicle-ci-broker

radicle-ci-broker

rad:zwTxygwuz5LDGBq255RA2CbNGrz8

Radicle CI broker

add a "kind" field to log messages, other fields

Merged liw opened 1 year ago

This will help with post-processing and filtering logs without having to rely on the text of the log message.

Activity Changes +144 -181

2 files changed +144 -181 cf395350 → b4f379ad

modified src/bin/cibtool.rs

@@ -43,7 +43,7 @@ use radicle_ci_broker::{

 mod cibtoolcmd;
 fn main() {
     let _logger = logger::open(logger::LogLevel::Info);
     logger::open(logger::LogLevel::Info);
     if let Err(e) = fallible_main() {
         eprintln!("ERROR: {}", e);
         let mut e = e.source();

modified src/logger.rs

@@ -1,20 +1,11 @@

 //! A logger abstraction for the CI broker.
 #[cfg(test)]
 use std::sync::Once;
 use std::{
     io::Write,
     path::Path,
     process::ExitStatus,
     sync::{Arc, Mutex},
     time::Duration,
 };
 use std::{path::Path, process::ExitStatus, time::Duration};
 use clap::ValueEnum;
 use radicle::{git::raw::Oid, identity::RepoId, node::Event};
 use tracing::{debug, error, info, trace, warn, Level};
 use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
 //use valuable::Valuable;
 use crate::{
     ci_event::CiEvent,

@@ -26,6 +17,33 @@ use crate::{

     run::Run,
 };
 // A unique type or kind for each log message.
 enum Kind {
     Debug,
     Startup,
     Shutdown,
     GotEvent,
     StartRun,
     AdapterMessage,
     FinishRun,
 }
 impl std::fmt::Display for Kind {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let s = match self {
             Self::Debug => "debug",
             Self::Startup => "startup",
             Self::Shutdown => "shutdown",
             Self::GotEvent => "got_event",
             Self::StartRun => "start_run",
             Self::FinishRun => "finish_run",
             Self::AdapterMessage => "adapter_message",
         };
         write!(f, "{s}")
     }
 }
 // We define our own type for log levels so that we can apply
 // clap::ValueEnum on it.
 #[derive(ValueEnum, Eq, PartialEq, Copy, Clone, Debug)]

@@ -63,60 +81,8 @@ impl From<LogLevel> for tracing::Level {

     }
 }
 // A custom type for log output. We need this to hide log messages
 // from output in successful tests. Cargo will hide standard output
 // and error by default, which is what we want, but it seems this only
 // applies to output via the `print!` family of macros. If we open a
 // new handle for stdout or stderr, output via that handle is not
 // captured by Cargo.
 //
 // Instead, we have our own log writer type that writes to stderr in
 // production mode, and uses `print!` in tests.
 #[derive(Default)]
 struct LogWriter {}
 #[cfg(test)]
 impl Write for LogWriter {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
         let s = String::from_utf8_lossy(buf);
         print!("{s}");
         Ok(buf.len())
     }
     fn flush(&mut self) -> std::io::Result<()> {
         Ok(())
     }
 }
 #[cfg(not(test))]
 impl Write for LogWriter {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
         let mut stderr = std::io::stderr();
         stderr.write_all(buf)?;
         stderr.flush()?;
         Ok(buf.len())
     }
     fn flush(&mut self) -> std::io::Result<()> {
         Ok(())
     }
 }
 pub struct Logger {
     minimum_log_level: Arc<Mutex<tracing::Level>>,
 }
 impl Logger {
     pub fn set_minimum_level(&self, level: LogLevel) {
         *self
             .minimum_log_level
             .lock()
             .expect("set minimum log level") = level.into();
     }
 }
 #[allow(clippy::unwrap_used)]
 pub fn open(level: LogLevel) -> Logger {
 pub fn open(level: LogLevel) {
     #[cfg(test)]
     let writer = fmt::TestWriter::new();

@@ -131,63 +97,43 @@ pub fn open(level: LogLevel) -> Logger {

         .with(filter_layer)
         .with(fmt_layer)
         .init();
     // Set the default log level. Trace is good for tests.
     let level = Arc::new(Mutex::new(LogLevel::Trace.into()));
     Logger {
         minimum_log_level: level,
     }
 }
 // Set up structured logging for tests.
 //
 // We have to keep the GlobalLoggerGuard we get when we initialize
 // `slog-scope` alive as long as we may need to do any logging. We can
 // only create that once per process. For tests, we do that here.
 //
 // We can't do the same thing for non-test processes, as that would
 // interfere with use of `radicle-ci-broker` as a library. Libraries
 // should not interfere with global state, unless they're specifically
 // intended to do that.
 //
 // This is for tests only: otherwise the default global logger is
 // used, and that's OK for a long-running process.
 // We use this to make sure we initialize the logger only once.
 #[cfg(test)]
 static INIT: Once = Once::new();
 // This is the actual logger for tests.
 #[cfg(test)]
 static mut LOGGER: Option<Logger> = None;
 #[cfg(test)]
 #[ctor::ctor]
 fn open_for_tests() {
     INIT.call_once(|| unsafe {
         LOGGER = Some(open(LogLevel::Trace));
     });
 }
 pub fn start_cib() {
     info!(version = env!("GIT_HEAD"), "CI broker starts");
     info!(
         kind = %Kind::Startup,
         version = env!("GIT_HEAD"),
         "CI broker starts"
     );
 }
 pub fn end_cib_successfully() {
     info!("CI broker ends successfully");
     info!(
         kind = %Kind::Shutdown,
         succss = true,
         "CI broker ends successfully"
     );
 }
 pub fn end_cib_in_error() {
     error!("CI broker ends in unrecoverable error");
     error!(
         kind = %Kind::Shutdown,
         succss = false,
         "CI broker ends in unrecoverable error"
     );
 }
 pub fn node_event_source_created(source: &NodeEventSource) {
     debug!(source = format!("{source:#?}"), "created node event source");
     debug!(
         kind = %Kind::Debug,
         source = format!("{source:#?}"),
         "created node event source"
     );
 }
 pub fn node_event_source_got_event(event: &Event) {
     debug!(
         kind = %Kind::GotEvent,
         node_event = ?event,
         "node event source received event"
     );

@@ -195,206 +141,215 @@ pub fn node_event_source_got_event(event: &Event) {

 pub fn node_event_source_eof(source: &NodeEventSource) {
     debug!(
         kind = %Kind::Debug,
         node_event_source = ?source,
         "node event source end of file on control socket"
     );
 }
 pub fn ci_event_source_created(source: &CiEventSource) {
     debug!(?source, "created CI event source");
     debug!(kind = %Kind::Debug, ?source, "created CI event source");
 }
 pub fn ci_event_source_got_events(events: &[CiEvent]) {
     debug!(?events, "CI event source received events");
     debug!(
         kind = %Kind::GotEvent,
         ?events,
         "CI event source received events"
     );
 }
 pub fn ci_event_source_disconnected() {
     info!("CI event source received disconnection");
     info!(kind = %Kind::Debug, "CI event source received disconnection");
 }
 pub fn ci_event_source_end() {
     info!("CI event source was notified end of events");
     info!(
         kind = %Kind::Debug,
         "CI event source was notified end of events"
     );
 }
 pub fn ci_event_source_eof(source: &CiEventSource) {
     info!(?source, "CI event source end of file");
     info!(kind = %Kind::Debug, ?source, "CI event source end of file");
 }
 pub fn loaded_config(config: &Config) {
     debug!("loaded configuration {config:#?}");
     debug!(kind = %Kind::Debug, "loaded configuration {config:#?}");
 }
 pub fn adapter_config(config: &Config) {
     debug!("adapter configuration {config:#?}");
     debug!(kind = %Kind::Debug, "adapter configuration {config:#?}");
 }
 pub fn queueproc_start() {
     debug!("start thread to process events until a shutdown event");
     debug!(kind = %Kind::Debug, "start thread to process events until a shutdown event");
 }
 pub fn queueproc_end() {
     debug!("thread to process events ends");
     debug!(kind = %Kind::Debug, "thread to process events ends");
 }
 pub fn queueproc_channel_disconnect() {
     info!("event notification channel disconnected");
     info!(kind = %Kind::Debug, "event notification channel disconnected");
 }
 pub fn queueproc_queue_length(len: usize) {
     trace!(?len, "event queue length");
     trace!(kind = %Kind::Debug, ?len, "event queue length");
 }
 pub fn queueproc_picked_event(id: &QueueId, event: &QueuedCiEvent) {
     info!("picked event from queue: {id}: {event:#?}");
     info!(kind = %Kind::GotEvent, "picked event from queue: {id}: {event:#?}");
 }
 pub fn queueproc_remove_event(id: &QueueId) {
     info!("remove event from queue: {id}");
     info!(kind = %Kind::Debug, "remove event from queue: {id}");
 }
 pub fn queueproc_action_run(rid: &RepoId, oid: &Oid) {
     info!("Action: run: {rid} {oid}");
     info!(kind = %Kind::Debug, "Action: run: {rid} {oid}");
 }
 pub fn queueproc_action_shutdown() {
     info!("Action: shutdown");
     info!(kind = %Kind::Debug, "Action: shutdown");
 }
 pub fn queueadd_start() {
     debug!("start thread to add events from node to event queue");
     debug!(kind = %Kind::Debug, "start thread to add events from node to event queue");
 }
 pub fn queueadd_control_socket_close() {
     info!("no more events from node control socket");
     info!(kind = %Kind::Debug, "no more events from node control socket");
 }
 pub fn queueadd_push_event(e: &CiEvent) {
     debug!("insert broker event into queue: {e:?}");
     debug!(kind = %Kind::GotEvent, event = ?e, "insert broker event into queue: {e:?}");
 }
 pub fn queueadd_end() {
     debug!("thread to process events ends");
     debug!(kind = %Kind::Debug, "thread to process events ends");
 }
 pub fn pages_directory_unset() {
     warn!("not writing HTML report pages as output directory has not been set");
     warn!(kind = %Kind::Debug, "not writing HTML report pages as output directory has not been set");
 }
 pub fn pages_interval(interval: Duration) {
     debug!(
         "wait about {} seconds to update HTML report pages again",
         interval.as_secs()
         kind = %Kind::Debug,
         interval = interval.as_secs(),
         "interval for waiting between HTML page updates",
     );
 }
 pub fn pages_disconnected() {
     info!("page updater: run notification channel disconnected");
     info!(kind = %Kind::Debug, "page updater: run notification channel disconnected");
 }
 pub fn pages_start() {
     debug!("start page updater thread");
     debug!(kind = %Kind::Debug, "start page updater thread");
 }
 pub fn pages_end() {
     debug!("end page updater thread");
     debug!(kind = %Kind::Debug, "end page updater thread");
 }
 pub fn event_disconnected() {
     info!("connection to node control socket broke");
     info!(kind = %Kind::Debug, "connection to node control socket broke");
 }
 pub fn event_end() {
     info!("no more node events from control socket: iterator ended");
     info!(kind = %Kind::Debug, "no more node events from control socket: iterator ended");
 }
 pub fn broker_db(filename: &Path) {
     info!("broker database: {}", filename.display());
     info!(kind = %Kind::Debug, filename = %filename.display(), "broker database");
 }
 pub fn broker_start_run(trigger: &Request) {
     info!("start CI run");
     debug!("trigger event: {trigger:#?}");
     info!(kind = %Kind::StartRun, "start CI run");
     debug!(kind = %Kind::Debug, event = ?trigger, "trigger event");
 }
 pub fn broker_end_run(run: &Run) {
     info!("Finish CI run");
     debug!("finished CI run: {run:#?}");
     info!(kind = %Kind::FinishRun, "Finish CI run");
     debug!(kind = %Kind::Debug, run = ?run, "finished CI run");
 }
 pub fn adapter_no_first_response() {
     error!("no first response message");
     error!(kind = %Kind::AdapterMessage, "no first response message");
 }
 pub fn adapter_no_second_response() {
     error!("no second response message");
     error!(kind = %Kind::AdapterMessage, "no second response message");
 }
 pub fn adapter_too_many_responses() {
     error!("too many response messages");
     error!(kind = %Kind::AdapterMessage, "too many response messages");
 }
 pub fn adapter_stderr_line(line: &str) {
     debug!(stderr_line = line, "adapter stderr");
     debug!(kind = %Kind::Debug, stderr_line = line, "adapter stderr");
 }
 pub fn adapter_result(exit: i32) {
     debug!(exit_code = exit, "adapter exit code");
     debug!(kind = %Kind::Debug, exit_code = exit, "adapter exit code");
 }
 pub fn adapter_did_not_exit_voluntarily() {
     warn!("adapter did not exit voluntarily: terminated for taking too long");
     warn!(kind = %Kind::Debug, "adapter did not exit voluntarily: terminated for taking too long");
 }
 pub fn adapter_did_not_exit() {
     warn!("adapter did not exit: probably killed by signal");
     warn!(kind = %Kind::Debug, "adapter did not exit: probably killed by signal");
 }
 pub fn timeoutcmd_request_termination(result: Result<(), std::sync::mpsc::SendError<()>>) {
     trace!("request termination of child process: {result:?}");
     trace!(kind = %Kind::Debug, "request termination of child process: {result:?}");
 }
 pub fn timeoutcmd_wait_word_from_nanny() {
     trace!("wait: wait for word from nanny");
     trace!(kind = %Kind::Debug, "wait: wait for word from nanny");
 }
 pub fn timeoutcmd_wait_got_word_from_nanny() {
     trace!("got word from nanny");
     trace!(kind = %Kind::Debug, "got word from nanny");
 }
 pub fn timeoutcmd_wait_on_nanny_to_end() {
     trace!("wait: wait on nanny thread to end");
     trace!(kind = %Kind::Debug, "wait: wait on nanny thread to end");
 }
 pub fn timeoutcmd_wait_on_stdin_writer_to_end() {
     trace!("wait: wait for stdin writer to terminate");
     trace!(kind = %Kind::Debug, "wait: wait for stdin writer to terminate");
 }
 pub fn timeoutcmd_wait_on_stdout_reader_to_end() {
     trace!("wait: wait for stdout reader to terminate");
     trace!(kind = %Kind::Debug, "wait: wait for stdout reader to terminate");
 }
 pub fn timeoutcmd_wait_on_stderr_reader_to_end() {
     trace!("wait: wait for stderr reader to terminate");
     trace!(kind = %Kind::Debug, "wait: wait for stderr reader to terminate");
 }
 pub fn timeoutcmd_wait_on_child_to_end() {
     trace!("wait: wait for child to terminate");
     trace!(kind = %Kind::Debug, "wait: wait for child to terminate");
 }
 pub fn timeoutcmd_wait_status(status: ExitStatus) {
     trace!("wait: wait status: {status:?}");
     trace!(kind = %Kind::Debug, ?status, "wait: wait status");
 }
 pub fn timeoutcmd_ok() {
     trace!("wait: return Ok result");
     trace!(kind = %Kind::Debug, "wait: return Ok result");
 }
 pub fn timeoutcmd_nanny_start() {
     trace!("nanny: start monitoring child");
     trace!(kind = %Kind::Debug, "nanny: start monitoring child");
 }
 pub fn timeoutcmd_nanny_terminated_as_requested(result: Result<(), std::io::Error>) {
     trace!("nanny: terminated child by request: {result:?}");
     trace!(kind = %Kind::Debug, ?result, "nanny: terminated child by request");
 }
 pub fn timeoutcmd_nanny_too_long(

@@ -404,43 +359,46 @@ pub fn timeoutcmd_nanny_too_long(

     result: Result<(), std::io::Error>,
 ) {
     trace!(
         "nanny: child {} has run for too long ({} ms > {} ms); terminated it: {:?}",
         id,
         elapsed.as_millis(),
         max.as_millis(),
         result,
         kind = %Kind::Debug,
         %id,
         elapsed_ms = elapsed.as_millis(),
         max_ms = max.as_millis(),
         ?result,
         "nanny: child has run for too long",
     );
 }
 pub fn timeoutcmd_nanny_child_died() {
     trace!("nanny: child has terminated");
     trace!(kind = %Kind::Debug, "nanny: child has terminated");
 }
 pub fn timeoutcmd_nanny_time_to_end() {
     trace!("nanny: tell other threads it's time to end");
     trace!(kind = %Kind::Debug, "nanny: tell other threads it's time to end");
 }
 pub fn timeoutcmd_nanny_ends() {
     trace!("nanny: ends");
     trace!(kind = %Kind::Debug, "nanny: ends");
 }
 pub fn timeoutcmd_line_reader_try_byte(name: &'static str) {
     trace!("line receiver {}: try to receive next byte", name);
     trace!(kind = %Kind::Debug, "line receiver {}: try to receive next byte", name);
 }
 pub fn timeoutcmd_line_reader_tried_byte(
     name: &'static str,
     result: Result<u8, std::sync::mpsc::TryRecvError>,
 ) {
     trace!("line receiver {}: tried to read line: {:?}", name, result);
     trace!(kind = %Kind::Debug, ?result, "line receiver {}: tried to read line", name);
 }
 pub fn timeoutcmd_line_reader_got_line(name: &'static str, line: &str) {
     trace!("line-receiver {}: received line={line:?}", name,);
     trace!(kind = %Kind::Debug, ?line, "line-receiver {}: received line", name);
 }
 pub fn timeoutcmd_line_reader_got_disconnected(name: &'static str) {
     trace!("line-receiver {}: disconnected", name);
     trace!(
         kind = %Kind::Debug,
         "line-receiver {}: disconnected", name);
 }
 pub fn timeoutcmd_line_reader_did_child_die(name: &'static str) {

@@ -449,6 +407,7 @@ pub fn timeoutcmd_line_reader_did_child_die(name: &'static str) {

 pub fn timeoutcmd_line_reader_child_died(name: &'static str) {
     trace!(
         kind = %Kind::Debug,
         "line receiver {}: OK: child has terminated, not returning line",
         name
     );

@@ -456,6 +415,7 @@ pub fn timeoutcmd_line_reader_child_died(name: &'static str) {

 pub fn timeoutcmd_line_reader_child_channel_disconnected(name: &'static str) {
     trace!(
         kind = %Kind::Debug,
         "line receiver {}: Disconnected: child has terminated, not returning line",
         name
     );

@@ -463,9 +423,10 @@ pub fn timeoutcmd_line_reader_child_channel_disconnected(name: &'static str) {

 pub fn timeoutcmd_nonblocking_try_byte(name: &'static str, count: usize) {
     trace!(
         "read_to_end {}: try to receive next byte ({} so far)",
         kind = %Kind::Debug,
         received_so_far = count,
         "read_to_end {}: try to receive next byte",
         name,
         count
     );
 }

@@ -475,15 +436,16 @@ pub fn timeoutcmd_nonblocking_tried_byte(

     byte: &[u8],
 ) {
     trace!(
         "read_to_end {}: tried to receive byte: result={:?} byte={:?}",
         kind = %Kind::Debug,
         ?result,
         ?byte,
         "read_to_end {}: tried to receive byte",
         name,
         result,
         byte
     );
 }
 pub fn timeoutcmd_nonblocking_eof(name: &'static str) {
     trace!("read_to_end {}: got end of file", name,);
     trace!(kind = %Kind::Debug, "read_to_end {}: got end of file", name,);
 }
 pub fn timeoutcmd_nonblocking_got_too_much(

@@ -492,27 +454,28 @@ pub fn timeoutcmd_nonblocking_got_too_much(

     byte: &[u8],
 ) {
     trace!(
         "read_to_end {}: received too much: result={:?} byte={:?}",
         kind = %Kind::Debug,
         ?result,
         ?byte,
         "read_to_end {}: received too much",
         name,
         result,
         byte
     );
 }
 pub fn timeoutcmd_nonblocking_read_error(name: &'static str, err: &std::io::Error) {
     trace!("read_to_end {}: read error: {}", name, err);
     trace!(kind = %Kind::Debug, ?err, "read_to_end {}: read error", name);
 }
 pub fn timeoutcmd_nonblocking_ends(name: &'static str) {
     trace!("read_to_end {}: ends", name,);
     trace!(kind = %Kind::Debug, "read_to_end {}: ends", name,);
 }
 pub fn debug(msg: &str) {
     debug!("{msg}");
     debug!(kind = %Kind::Debug, "{msg}");
 }
 pub fn debug2(msg: String) {
     debug!("{msg}");
     debug!(kind = %Kind::Debug, "{msg}");
 }
 pub fn error(msg: &str, e: &impl std::error::Error) {