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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

use std::path::PathBuf;

use clap::Args;

use crate::command::{Executable, default_tracing};

use super::config::ClusterConfig;
use super::{
    format_daemon_port_arg, format_labels_arg, format_zenoh_peer_arg, print_summary,
    record_ssh_result, run_ssh, ssh_target,
};

/// Install dora-daemon as a systemd service on each machine.
///
/// SSH-es into each machine, writes a systemd unit file, and enables the service.
///
/// Examples:
///
///   dora cluster install cluster.yml
#[derive(Debug, Args)]
#[clap(verbatim_doc_comment)]
pub struct Install {
    /// Path to the cluster configuration file
    #[clap(value_name = "PATH", value_hint = clap::ValueHint::FilePath)]
    config: PathBuf,
}

impl Executable for Install {
    fn execute(self) -> eyre::Result<()> {
        default_tracing()?;
        let config = ClusterConfig::load(&self.config)?;

        let mut failures = Vec::new();

        let zenoh_peer_arg = format_zenoh_peer_arg(config.zenoh_peer.as_deref());
        for machine in &config.machines {
            let target = ssh_target(machine);
            let labels_arg = format_labels_arg(&machine.labels);
            let daemon_port_arg = format_daemon_port_arg(machine.daemon_port);
            let service_name = format!("dora-daemon-{}", machine.id);

            let unit = format!(
                r#"[Unit]
Description=Dora Daemon ({id})
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
ExecStart=dora daemon --machine-id {id} --coordinator-addr {addr} --coordinator-port {port}{daemon_port_arg}{zenoh_peer_arg}{labels} --quiet
Restart=on-failure
RestartSec=5

[Install]
WantedBy=multi-user.target
"#,
                id = machine.id,
                addr = config.coordinator.addr,
                port = config.coordinator.port,
                labels = labels_arg,
            );

            // Escape single quotes for shell
            let escaped_unit = unit.replace('\'', "'\\''");
            let cmd = format!(
                "echo '{escaped_unit}' | sudo tee /etc/systemd/system/{service_name}.service > /dev/null && sudo systemctl daemon-reload && sudo systemctl enable --now {service_name}"
            );

            println!("Installing {service_name} on {} ({target})", machine.id);
            let result = run_ssh(&target, machine.port, &cmd);
            record_ssh_result(
                &mut failures,
                &machine.id,
                result,
                &format!("{service_name} installed and started"),
            );
        }

        print_summary(
            "daemon(s) installed as systemd services",
            config.machines.len(),
            &failures,
        );

        if failures.is_empty() {
            Ok(())
        } else {
            eyre::bail!(
                "install failed on {}/{} machine(s)",
                failures.len(),
                config.machines.len()
            )
        }
    }
}