add standalone clean subcommand

Signed-off-by: drindr <dreamchancn@qq.com>

Author: drindr

binaries/cli/src/command/clean.rs

@@ -0,0 +1,67 @@
+use super::{
+    Executable, default_tracing,
+    list::{OutputEntry, display_entries},
+};
+use crate::{
+    LOCALHOST,
+    common::{connect_and_check_version, rpc},
+    formatting::OutputFormat,
+};
+use clap::Args;
+use dora_core::topics::DORA_COORDINATOR_PORT_CONTROL_DEFAULT;
+use dora_message::{cli_to_coordinator::CliControlClient, tarpc};
+use eyre::Context;
+
+#[derive(Debug, Args)]
+/// Remove finished and failed dataflows from the list.
+pub struct CleanArgs {
+    /// Address of the dora coordinator
+    #[clap(long, value_name = "IP", default_value_t = LOCALHOST)]
+    pub coordinator_addr: std::net::IpAddr,
+    /// Port number of the coordinator control server
+    #[clap(long, value_name = "PORT", default_value_t = DORA_COORDINATOR_PORT_CONTROL_DEFAULT)]
+    pub coordinator_port: u16,
+    /// Output format
+    #[clap(long, value_name = "FORMAT", default_value_t = OutputFormat::Table)]
+    pub format: OutputFormat,
+}
+
+impl Executable for CleanArgs {
+    async fn execute(self) -> eyre::Result<()> {
+        default_tracing()?;
+
+        let client = connect_and_check_version(self.coordinator_addr, self.coordinator_port)
+            .await
+            .wrap_err("failed to connect to dora coordinator")?;
+
+        clean(&client, self.format).await
+    }
+}
+
+async fn clean(client: &CliControlClient, format: OutputFormat) -> eyre::Result<()> {
+    let list = rpc(
+        "clean finished dataflows",
+        client.clean(tarpc::context::current()),
+    )
+    .await?;
+
+    let entries: Vec<OutputEntry> = list
+        .0
+        .into_iter()
+        .map(|entry| OutputEntry {
+            uuid: entry.id.uuid,
+            name: entry.id.name.unwrap_or_default(),
+            status: entry.status,
+            nodes: 0,
+            cpu: 0.0,
+            memory: 0.0,
+        })
+        .collect();
+
+    if entries.is_empty() {
+        println!("No finished or failed dataflows to clean.");
+        return Ok(());
+    }
+
+    display_entries(&entries, format, false)
+}

binaries/cli/src/command/list.rs

@@ -37,8 +37,6 @@ pub struct ListArgs {
     /// Sort by field (memory, cpu)
     #[clap(long, value_name = "FIELD")]
     pub sort_by: Option<String>,
-    #[clap(long)]
-    pub clean: bool,
 }
 
 impl Executable for ListArgs {
@@ -49,26 +47,18 @@ impl Executable for ListArgs {
             .await
             .wrap_err("failed to connect to dora coordinator")?;
 
-        list(
-            &client,
-            self.format,
-            self.status,
-            self.name,
-            self.sort_by,
-            self.clean,
-        )
-        .await
+        list(&client, self.format, self.status, self.name, self.sort_by).await
     }
 }
 
 #[derive(Serialize)]
-struct OutputEntry {
-    uuid: Uuid,
-    name: String,
-    status: DataflowStatus,
-    nodes: usize,
-    cpu: f64,
-    memory: f64,
+pub(super) struct OutputEntry {
+    pub uuid: Uuid,
+    pub name: String,
+    pub status: DataflowStatus,
+    pub nodes: usize,
+    pub cpu: f64,
+    pub memory: f64,
 }
 
 #[derive(Default)]
@@ -78,23 +68,66 @@ struct DataflowMetrics {
     total_memory_mb: f64,
 }
 
+/// Render a list of [`OutputEntry`] values to stdout in either table or JSON format.
+///
+/// When `show_metrics` is `false` the Nodes, CPU and Memory columns are omitted.
+pub(super) fn display_entries(
+    entries: &[OutputEntry],
+    format: OutputFormat,
+    show_metrics: bool,
+) -> eyre::Result<()> {
+    match format {
+        OutputFormat::Table => {
+            let mut tw = TabWriter::new(std::io::stdout().lock());
+            if show_metrics {
+                tw.write_all(b"UUID\tName\tStatus\tNodes\tCPU\tMemory\n")?;
+            } else {
+                tw.write_all(b"UUID\tName\tStatus\n")?;
+            }
+            for entry in entries {
+                let status = match entry.status {
+                    DataflowStatus::Running => "Running",
+                    DataflowStatus::Finished => "Succeeded",
+                    DataflowStatus::Failed => "Failed",
+                };
+                if show_metrics {
+                    tw.write_all(
+                        format!(
+                            "{}\t{}\t{}\t{}\t{}\t{}\n",
+                            entry.uuid,
+                            entry.name,
+                            status,
+                            entry.nodes,
+                            format!("{:.1}%", entry.cpu),
+                            format!("{:.1} GB", entry.memory),
+                        )
+                        .as_bytes(),
+                    )?;
+                } else {
+                    tw.write_all(
+                        format!("{}\t{}\t{}\n", entry.uuid, entry.name, status).as_bytes(),
+                    )?;
+                }
+            }
+            tw.flush()?;
+        }
+        OutputFormat::Json => {
+            for entry in entries {
+                println!("{}", serde_json::to_string(entry)?);
+            }
+        }
+    }
+    Ok(())
+}
+
 async fn list(
     client: &CliControlClient,
     format: OutputFormat,
     status_filter: Option<String>,
     name_filter: Option<String>,
     sort_by: Option<String>,
-    clean: bool,
 ) -> Result<(), eyre::ErrReport> {
-    let list = if !clean {
-        query_running_dataflows(client).await
-    } else {
-        rpc(
-            "clean finished dataflows",
-            client.clean(tarpc::context::current()),
-        )
-        .await
-    }?;
+    let list = query_running_dataflows(client).await?;
 
     // Get node information via tarpc
     let node_infos = rpc(
@@ -197,39 +230,5 @@ async fn list(
         }
     }
 
-    match format {
-        OutputFormat::Table => {
-            let mut tw = TabWriter::new(std::io::stdout().lock());
-            // Header
-            tw.write_all(format!("UUID\tName\tStatus\tNodes\tCPU\tMemory\n").as_bytes())?;
-            for entry in entries {
-                let status = match entry.status {
-                    DataflowStatus::Running => "Running",
-                    DataflowStatus::Finished => "Succeeded",
-                    DataflowStatus::Failed => "Failed",
-                };
-
-                tw.write_all(
-                    format!(
-                        "{}\t{}\t{}\t{}\t{}\t{}\n",
-                        entry.uuid,
-                        entry.name,
-                        status,
-                        entry.nodes,
-                        format!("{:.1}%", entry.cpu),
-                        format!("{:.1} GB", entry.memory)
-                    )
-                    .as_bytes(),
-                )?;
-            }
-            tw.flush()?;
-        }
-        OutputFormat::Json => {
-            for entry in entries {
-                println!("{}", serde_json::to_string(&entry)?);
-            }
-        }
-    }
-
-    Ok(())
+    display_entries(&entries, format, true)
 }

binaries/cli/src/command/mod.rs

@@ -1,4 +1,5 @@
 mod build;
+mod clean;
 mod completion;
 mod coordinator;
 mod daemon;
@@ -23,6 +24,7 @@ pub use build::{build, build_async};
 pub use run::{Run, run, run_func};
 
 use build::Build;
+use clean::CleanArgs;
 use completion::Completion;
 use coordinator::Coordinator;
 use daemon::Daemon;
@@ -60,6 +62,8 @@ pub enum Command {
     Stop(Stop),
     #[clap(alias = "ps")]
     List(ListArgs),
+    /// Remove finished and failed dataflows from the list
+    Clean(CleanArgs),
     // Planned for future releases:
     // Dashboard,
     #[command(allow_missing_positional = true)]
@@ -119,6 +123,7 @@ impl Executable for Command {
             Command::Start(args) => args.execute().await,
             Command::Stop(args) => args.execute().await,
             Command::List(args) => args.execute().await,
+            Command::Clean(args) => args.execute().await,
             Command::Logs(args) => args.execute().await,
             Command::Inspect(args) => args.execute().await,
             Command::Daemon(args) => args.execute().await,