pbs-client/src/task_log.rs

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