Add event log export to (ND)JSON

Author: Kobzol

CHANGELOG.md

@@ -55,6 +55,13 @@
     $ hq submit --stdin bash
     ```
 
+   * You can now store HyperQueue events into a log file and later export them to JSON for further
+     processing. You can find more information in the
+     [documentation](https://it4innovations.github.io/hyperqueue/stable/jobs/directives/). 
+
+     *Note that this functionality is quite low-level, and it's designed primarily for
+     tool builders that use HyperQueue programmatically, not regular users. It is also currently
+     unstable.*
 
 ### Worker configuration
   * You can now select what should happen when a worker loses its connection to the server using the

Cargo.lock

@@ -47,6 +47,7 @@ termion = "1.5"
 indicatif = "0.16.2"
 textwrap = "0.14"
 async-compression = { version = "0.3.8", features = ["tokio", "gzip"] }
+flate2 = { version = "1.0.22", features = ["default"] }
 
 # Tako
 tako = { path = "../tako" }

crates/hyperqueue/Cargo.toml

@@ -8,6 +8,7 @@ use cli_table::ColorChoice;
 
 use clap::ArgSettings::HiddenShortHelp;
 use hyperqueue::client::commands::autoalloc::{command_autoalloc, AutoAllocOpts};
+use hyperqueue::client::commands::event::{command_event_log, EventLogOpts};
 use hyperqueue::client::commands::job::{
     cancel_job, output_job_cat, output_job_detail, output_job_list, output_job_tasks,
     JobCancelOpts, JobCatOpts, JobInfoOpts, JobListOpts, JobTasksOpts,
@@ -111,6 +112,8 @@ enum SubCommand {
     /// Automatic allocation management
     #[clap(name = "alloc")]
     AutoAlloc(AutoAllocOpts),
+    /// Event log management
+    EventLog(EventLogOpts),
     ///Commands for the dashboard
     Dashboard(DashboardOpts),
     /// Generate shell completion script
@@ -146,7 +149,7 @@ struct ServerStartOpts {
 
     /// Path to a log file where events will be stored.
     #[clap(long, hide(true))]
-    event_log_file: Option<PathBuf>,
+    event_log_path: Option<PathBuf>,
 }
 
 #[derive(Parser)]
@@ -297,7 +300,7 @@ async fn command_server_start(
         client_port: opts.client_port,
         worker_port: opts.worker_port,
         event_buffer_size: opts.event_store_size,
-        event_log_file: opts.event_log_file,
+        event_log_path: opts.event_log_path,
     };
 
     init_hq_server(gsettings, server_cfg).await
@@ -601,6 +604,7 @@ async fn main() -> hyperqueue::Result<()> {
         SubCommand::Dashboard(opts) => command_dashboard_start(&gsettings, opts).await,
         SubCommand::Log(opts) => command_log(&gsettings, opts),
         SubCommand::AutoAlloc(opts) => command_autoalloc(&gsettings, opts).await,
+        SubCommand::EventLog(opts) => command_event_log(opts),
         SubCommand::GenerateCompletion(opts) => generate_completion(opts),
     };
 

crates/hyperqueue/src/bin/hq.rs

@@ -42,13 +42,13 @@ enum AutoAllocCommand {
 }
 
 #[derive(Parser)]
-pub struct AddQueueOpts {
+struct AddQueueOpts {
     #[clap(subcommand)]
     subcmd: AddQueueCommand,
 }
 
 #[derive(Parser)]
-pub struct RemoveQueueOpts {
+struct RemoveQueueOpts {
     /// ID of the allocation queue that should be removed
     queue_id: DescriptorId,
 
@@ -59,7 +59,7 @@ pub struct RemoveQueueOpts {
 }
 
 #[derive(Parser)]
-pub enum AddQueueCommand {
+enum AddQueueCommand {
     /// Create a PBS allocation queue
     Pbs(SharedQueueOpts),
     /// Create a SLURM allocation queue
@@ -68,7 +68,7 @@ pub enum AddQueueCommand {
 
 #[derive(Parser)]
 #[clap(setting = clap::AppSettings::TrailingVarArg)]
-pub struct SharedQueueOpts {
+struct SharedQueueOpts {
     /// How many jobs should be waiting in the queue to be started
     #[clap(long, short, default_value = "4")]
     backlog: u32,
@@ -116,27 +116,27 @@ pub struct SharedQueueOpts {
 }
 
 #[derive(Parser)]
-pub struct DryRunOpts {
+struct DryRunOpts {
     #[clap(subcommand)]
     subcmd: DryRunCommand,
 }
 
 #[derive(Parser)]
-pub enum DryRunCommand {
+enum DryRunCommand {
     /// Try to create a PBS allocation
     Pbs(SharedQueueOpts),
     /// Try to create a SLURM allocation
     Slurm(SharedQueueOpts),
 }
 
 #[derive(Parser)]
-pub struct EventsOpts {
+struct EventsOpts {
     /// ID of the allocation queue
     queue: u32,
 }
 
 #[derive(Parser)]
-pub struct AllocationsOpts {
+struct AllocationsOpts {
     /// ID of the allocation queue
     queue: u32,
 

crates/hyperqueue/src/client/commands/autoalloc.rs

@@ -0,0 +1,71 @@
+mod output;
+
+use crate::client::commands::event::output::format_event;
+use crate::common::strutils::pluralize;
+use crate::event::log::EventLogReader;
+use anyhow::anyhow;
+use clap::{Parser, ValueHint};
+use std::io::{BufWriter, Write};
+use std::path::PathBuf;
+
+#[derive(Parser)]
+pub struct EventLogOpts {
+    /// Manage event log files.
+    #[clap(subcommand)]
+    command: EventCommand,
+}
+
+#[derive(Parser)]
+enum EventCommand {
+    /// Export events from a log file to NDJSON (line-delimited JSON).
+    /// Events will be exported to `stdout`, you can redirect it e.g. to a file.
+    Export(ExportOpts),
+}
+
+#[derive(Parser)]
+struct ExportOpts {
+    /// Path to a file containing the event log.
+    /// The file had to be created with `hq server start --event-log-path=<PATH>`.
+    #[clap(value_hint = ValueHint::FilePath)]
+    logfile: PathBuf,
+}
+
+pub fn command_event_log(opts: EventLogOpts) -> anyhow::Result<()> {
+    match opts.command {
+        EventCommand::Export(opts) => export_json(opts),
+    }
+}
+
+fn export_json(opts: ExportOpts) -> anyhow::Result<()> {
+    let file = EventLogReader::open(&opts.logfile).map_err(|error| {
+        anyhow!(
+            "Cannot open event log file at `{}`: {error:?}",
+            opts.logfile.display()
+        )
+    })?;
+
+    let stdout = std::io::stdout();
+    let stdout = stdout.lock();
+    let mut stdout = BufWriter::new(stdout);
+
+    let mut count = 0;
+    for event in file {
+        match event {
+            Ok(event) => {
+                writeln!(stdout, "{}", format_event(event))?;
+                count += 1;
+            }
+            Err(error) => {
+                log::error!(
+                    "Encountered an error while reading event log file: {error:?}.\n
+The file might have been incomplete."
+                )
+            }
+        }
+    }
+
+    log::info!("Outputted {count} {}", pluralize("event", count));
+
+    stdout.flush()?;
+    Ok(())
+}

crates/hyperqueue/src/client/commands/event/mod.rs

@@ -0,0 +1,32 @@
+use crate::client::output::json::format_datetime;
+use crate::event::events::MonitoringEventPayload;
+use crate::event::MonitoringEvent;
+use serde_json::json;
+use tako::messages::worker::WorkerOverview;
+
+pub fn format_event(event: MonitoringEvent) -> serde_json::Value {
+    json!({
+        "id": event.id,
+        "time": format_datetime(event.time),
+        "event": format_payload(event.payload)
+    })
+}
+
+fn format_payload(event: MonitoringEventPayload) -> serde_json::Value {
+    match event {
+        MonitoringEventPayload::WorkerConnected(id, ..) => json!({
+            "type": "worker-connected",
+            "id": id
+        }),
+        MonitoringEventPayload::WorkerLost(id, reason) => json!({
+            "type": "worker-lost",
+            "id": id,
+            "reason": reason
+        }),
+        MonitoringEventPayload::OverviewUpdate(WorkerOverview { id, hw_state, .. }) => json!({
+            "type": "worker-overview",
+            "id": id,
+            "hw_state": hw_state
+        }),
+    }
+}

crates/hyperqueue/src/client/commands/event/output.rs

@@ -1,4 +1,5 @@
 pub mod autoalloc;
+pub mod event;
 pub mod job;
 pub mod log;
 pub mod stats;

crates/hyperqueue/src/client/commands/mod.rs

@@ -472,6 +472,6 @@ fn format_duration(duration: Duration) -> serde_json::Value {
     let value = duration.as_secs() as f64 + duration.subsec_nanos() as f64 * 1e-9;
     json!(value)
 }
-fn format_datetime<T: Into<DateTime<Utc>>>(time: T) -> serde_json::Value {
+pub fn format_datetime<T: Into<DateTime<Utc>>>(time: T) -> serde_json::Value {
     json!(time.into())
 }

crates/hyperqueue/src/client/output/json.rs

@@ -1,10 +1,10 @@
 use std::borrow::Cow;
 
-/// Return the input string with an added "s" at the end if `count` is larger than one.
+/// Return the input string with an added "s" at the end if `count` is larger than one and non-zero.
 pub fn pluralize(value: &str, count: usize) -> Cow<str> {
-    if count > 1 {
-        Cow::Owned(format!("{}s", value))
-    } else {
+    if count == 1 {
         Cow::Borrowed(value)
+    } else {
+        Cow::Owned(format!("{}s", value))
     }
 }