dora-cli 1.0.0-rc.2

`dora` goal is to be a low latency, composable, and distributed data flow.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

use std::io::Write;

use clap::Args;
use eyre::Context;
use tabwriter::TabWriter;

use crate::{
    command::{Executable, default_tracing},
    common::{CoordinatorOptions, query_running_dataflows},
};

use super::query_connected_daemons;

/// Show the current status of the cluster.
///
/// Displays connected daemons and active dataflow count.
///
/// Examples:
///
///   dora cluster status
#[derive(Debug, Args)]
#[clap(verbatim_doc_comment)]
pub struct Status {
    #[clap(flatten)]
    coordinator: CoordinatorOptions,
}

impl Executable for Status {
    fn execute(self) -> eyre::Result<()> {
        default_tracing()?;
        let session = self.coordinator.connect()?;

        let daemons = query_connected_daemons(&session)?;
        let dataflows = query_running_dataflows(&session)?;
        let active_count = dataflows.get_active().len();

        if daemons.is_empty() {
            println!("No daemons connected.");
        } else {
            let mut tw = TabWriter::new(std::io::stdout().lock());
            tw.write_all(b"DAEMON ID\tLAST HEARTBEAT\tRESTARTS\tHEALTH KILLS\tINPUT TIMEOUTS\tCB RECOVERIES\n")
                .context("write")?;
            for d in &daemons {
                let (restarts, kills, timeouts, recoveries) = d
                    .ft_stats
                    .as_ref()
                    .map(|s| {
                        (
                            s.restarts.to_string(),
                            s.health_check_kills.to_string(),
                            s.input_timeouts.to_string(),
                            s.circuit_breaker_recoveries.to_string(),
                        )
                    })
                    .unwrap_or_else(|| ("-".into(), "-".into(), "-".into(), "-".into()));
                tw.write_all(
                    format!(
                        "{}\t{}ms ago\t{}\t{}\t{}\t{}\n",
                        d.daemon_id, d.last_heartbeat_ago_ms, restarts, kills, timeouts, recoveries,
                    )
                    .as_bytes(),
                )
                .context("write")?;
            }
            tw.flush().context("flush")?;
        }

        println!("\nActive dataflows: {active_count}");
        Ok(())
    }
}