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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113

use super::{Executable, default_tracing};
use crate::common::{
    connect_and_check_version, handle_dataflow_result, long_context, query_running_dataflows, rpc,
};
use dora_core::topics::{DORA_COORDINATOR_PORT_CONTROL_DEFAULT, LOCALHOST};
use dora_message::{
    cli_to_coordinator::CoordinatorControlClient, coordinator_to_cli::StopDataflowReply,
};
use duration_str::parse;
use eyre::Context;
use std::net::IpAddr;
use std::time::Duration;
use uuid::Uuid;

#[derive(Debug, clap::Args)]
/// Stop a running dataflow. If no id or name is provided, you will be able to choose between the running dataflows.
///
/// You could specify the strategy to stop the dataflow with `--grace-duration` or `--force`.
pub struct Stop {
    /// UUID of the dataflow that should be stopped
    uuid: Option<Uuid>,
    /// Name of the dataflow that should be stopped
    #[clap(long)]
    name: Option<String>,
    /// Kill the dataflow if it doesn't stop after the given duration
    ///
    /// Specifically, it does the following:
    /// 1. Sends `Event::Stop` to all nodes of the dataflow.
    /// 2. After DURATION, performs a soft kill (sending SIGTERM, or Ctrl-Break on Windows).
    /// 3. If the dataflow is still running after DURATION * 0.5, terminates all its processes.
    #[clap(
        long,
        value_name = "DURATION",
        group = "strategy",
        verbatim_doc_comment
    )]
    #[arg(value_parser = parse)]
    grace_duration: Option<Duration>,
    /// Force stop the dataflow by immediately terminating all its processes
    #[clap(short, long, action, group = "strategy")]
    force: bool,
    /// Address of the dora coordinator
    #[clap(long, value_name = "IP", default_value_t = LOCALHOST)]
    coordinator_addr: IpAddr,
    /// Port number of the coordinator control server
    #[clap(long, value_name = "PORT", default_value_t = DORA_COORDINATOR_PORT_CONTROL_DEFAULT)]
    coordinator_port: u16,
}

impl Executable for Stop {
    async fn execute(self) -> eyre::Result<()> {
        default_tracing()?;
        let client = connect_and_check_version(self.coordinator_addr, self.coordinator_port)
            .await
            .wrap_err("could not connect to dora coordinator")?;
        match (self.uuid, self.name) {
            (Some(uuid), _) => stop_dataflow(uuid, self.grace_duration, self.force, &client).await,
            (None, Some(name)) => {
                stop_dataflow_by_name(name, self.grace_duration, self.force, &client).await
            }
            (None, None) => {
                stop_dataflow_interactive(self.grace_duration, self.force, &client).await
            }
        }
    }
}

async fn stop_dataflow_interactive(
    grace_duration: Option<Duration>,
    force: bool,
    client: &CoordinatorControlClient,
) -> eyre::Result<()> {
    let list = query_running_dataflows(client)
        .await
        .wrap_err("failed to query running dataflows")?;
    let active = list.get_active();
    if active.is_empty() {
        eprintln!("No dataflows are running");
    } else {
        let selection = inquire::Select::new("Choose dataflow to stop:", active).prompt()?;
        stop_dataflow(selection.uuid, grace_duration, force, client).await?;
    }

    Ok(())
}

async fn stop_dataflow(
    uuid: Uuid,
    grace_duration: Option<Duration>,
    force: bool,
    client: &CoordinatorControlClient,
) -> Result<(), eyre::ErrReport> {
    let StopDataflowReply { uuid, result } = rpc(
        "stop dataflow",
        client.stop(long_context(), uuid, grace_duration, force),
    )
    .await?;
    handle_dataflow_result(result, Some(uuid))
}

async fn stop_dataflow_by_name(
    name: String,
    grace_duration: Option<Duration>,
    force: bool,
    client: &CoordinatorControlClient,
) -> Result<(), eyre::ErrReport> {
    let StopDataflowReply { uuid, result } = rpc(
        "stop dataflow by name",
        client.stop_by_name(long_context(), name, grace_duration, force),
    )
    .await?;
    handle_dataflow_result(result, Some(uuid))
}