dora-cli 0.5.0

`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

use clap::ValueEnum;
use dora_message::{common::NodeErrorCause, coordinator_to_cli::DataflowResult};

#[derive(Clone, Copy, Debug, ValueEnum)]
pub enum OutputFormat {
    Table,
    Json,
}

impl std::fmt::Display for OutputFormat {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            OutputFormat::Table => write!(f, "table"),
            OutputFormat::Json => write!(f, "json"),
        }
    }
}

pub struct FormatDataflowError<'a>(pub &'a DataflowResult);

impl std::fmt::Display for FormatDataflowError<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        writeln!(f)?;
        let failed = self
            .0
            .node_results
            .iter()
            .filter_map(|(id, r)| r.as_ref().err().map(|e| (id, e)));
        let total_failed = failed.clone().count();

        let mut non_cascading: Vec<_> = failed
            .clone()
            .filter(|(_, e)| !matches!(e.cause, NodeErrorCause::Cascading { .. }))
            .collect();
        non_cascading.sort_by_key(|(_, e)| e.timestamp);
        // try to print earliest non-cascading error
        let hidden = if !non_cascading.is_empty() {
            let printed = non_cascading.len();
            for (id, err) in non_cascading {
                writeln!(f, "Node `{id}` failed: {err}")?;
            }
            total_failed - printed
        } else {
            // no non-cascading errors -> print earliest cascading
            let mut all: Vec<_> = failed.collect();
            all.sort_by_key(|(_, e)| e.timestamp);
            if let Some((id, err)) = all.first() {
                write!(f, "Node `{id}` failed: {err}")?;
                total_failed - 1
            } else {
                write!(f, "unknown error")?;
                0
            }
        };

        if hidden > 1 {
            write!(
                f,
                "\n\nThere are {hidden} consequential errors. Check the `out/{}` folder for full details.",
                self.0.uuid
            )?;
        }

        Ok(())
    }
}