pbs-client/src/task_log.rs

   1 use std::sync::{
   2     atomic::{AtomicUsize, Ordering},
   3     Arc,
   4 };
   5
   6 use anyhow::{bail, Error};
   7 use futures::*;
   8 use serde_json::{json, Value};
   9 use tokio::signal::unix::{signal, SignalKind};
  10
  11 use proxmox_router::cli::format_and_print_result;
  12
  13 use pbs_api_types::percent_encoding::percent_encode_component;
  14
  15 use super::HttpClient;
  16
  17 /// Display task log on console
  18 ///
  19 /// This polls the task API and prints the log to the console. It also
  20 /// catches interrupt signals, and sends an abort request to the task if the
  21 /// user presses CTRL-C and `forward_interrupt` is true. Two interrupts cause an
  22 /// immediate end of the loop. The task may still run in that case.
  23 pub async fn display_task_log(
  24     client: &HttpClient,
  25     upid_str: &str,
  26     strip_date: bool,
  27     forward_interrupt: bool,
  28 ) -> Result<(), Error> {
  29     let mut signal_stream = signal(SignalKind::interrupt())?;
  30     let abort_count = Arc::new(AtomicUsize::new(0));
  31     let abort_count2 = Arc::clone(&abort_count);
  32
  33     let abort_future = async move {
  34         while signal_stream.recv().await.is_some() {
  35             log::info!("got shutdown request (SIGINT)");
  36             let prev_count = abort_count2.fetch_add(1, Ordering::SeqCst);
  37             if prev_count >= 1 {
  38                 log::info!("forced exit (task still running)");
  39                 break;
  40             }
  41         }
  42         Ok::<_, Error>(())
  43     };
  44
  45     let request_future = async move {
  46         let mut start = 1;
  47         let limit = 500;
  48
  49         let upid_encoded = percent_encode_component(upid_str);
  50
  51         loop {
  52             let abort = abort_count.load(Ordering::Relaxed);
  53             if abort > 0 {
  54                 if forward_interrupt {
  55                     let path = format!("api2/json/nodes/localhost/tasks/{upid_encoded}");
  56                     let _ = client.delete(&path, None).await?;
  57                 } else {
  58                     return Ok(());
  59                 }
  60             }
  61
  62             let param = json!({ "start": start, "limit": limit, "test-status": true });
  63
  64             let path = format!("api2/json/nodes/localhost/tasks/{upid_encoded}/log");
  65             let result = client.get(&path, Some(param)).await?;
  66
  67             let active = result["active"].as_bool().unwrap();
  68             let total = result["total"].as_u64().unwrap();
  69             let data = result["data"].as_array().unwrap();
  70
  71             let lines = data.len();
  72
  73             for item in data {
  74                 let n = item["n"].as_u64().unwrap();
  75                 let t = item["t"].as_str().unwrap();
  76                 if n != start {
  77                     bail!("got wrong line number in response data ({n} != {start}");
  78                 }
  79                 if strip_date && t.len() > 27 && &t[25..27] == ": " {
  80                     let line = &t[27..];
  81                     println!("{line}");
  82                 } else {
  83                     println!("{t}");
  84                 }
  85                 start += 1;
  86             }
  87
  88             if start > total {
  89                 if active {
  90                     tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await;
  91                 } else {
  92                     break;
  93                 }
  94             } else if lines != limit {
  95                 bail!("got wrong number of lines from server ({lines} != {limit})");
  96             }
  97         }
  98
  99         let status_path = format!("api2/json/nodes/localhost/tasks/{upid_encoded}/status");
 100         let task_result = &client.get(&status_path, None).await?["data"];
 101         if task_result["status"].as_str() == Some("stopped") {
 102             match task_result["exitstatus"].as_str() {
 103                 None => bail!("task stopped with unknown status"),
 104                 Some(status) if status == "OK" || status.starts_with("WARNINGS") => (),
 105                 Some(status) => bail!("task failed (status {status})"),
 106             }
 107         }
 108
 109         Ok(())
 110     };
 111
 112     futures::select! {
 113         request = request_future.fuse() => request?,
 114         abort = abort_future.fuse() => abort?,
 115     };
 116
 117     Ok(())
 118 }
 119
 120 /// Display task result (upid), or view task log - depending on output format
 121 ///
 122 /// In case of a task log of a running task, this will forward interrupt signals
 123 /// to the task and potentially abort it!
 124 pub async fn view_task_result(
 125     client: &HttpClient,
 126     result: Value,
 127     output_format: &str,
 128 ) -> Result<(), Error> {
 129     let data = &result["data"];
 130     if output_format == "text" {
 131         if let Some(upid) = data.as_str() {
 132             display_task_log(client, upid, true, true).await?;
 133         }
 134     } else {
 135         format_and_print_result(data, output_format);
 136     }
 137
 138     Ok(())
 139 }