consortium_ansible/
tasks.rs

1//! DagTask implementations for Ansible orchestration.
2//!
3//! Pipeline: build-env → copy-env → run-playbook → verify
4
5use std::process::Command;
6
7use consortium::dag::{DagContext, DagTask, TaskId, TaskOutcome};
8use consortium_nix::{build, copy};
9
10/// Build a hermetic ansible environment via nix.
11///
12/// Writes output: `build-ansible-env:{env_name}` → String (store path)
13pub struct NixBuildAnsibleEnvTask {
14    pub env_name: String,
15    pub flake_attr: String,
16}
17
18impl NixBuildAnsibleEnvTask {
19    pub fn new(env_name: &str, flake_uri: &str) -> Self {
20        Self {
21            env_name: env_name.to_string(),
22            flake_attr: format!("{}#ansibleEnvs.{}", flake_uri, env_name),
23        }
24    }
25}
26
27impl DagTask for NixBuildAnsibleEnvTask {
28    fn execute(&self, ctx: &DagContext) -> TaskOutcome {
29        match build::build_flake_attr(&self.flake_attr, None) {
30            Ok(path) => {
31                ctx.set_output(TaskId(format!("build-ansible-env:{}", self.env_name)), path);
32                TaskOutcome::Success
33            }
34            Err(e) => TaskOutcome::Failed(format!("build ansible env: {}", e)),
35        }
36    }
37
38    fn describe(&self) -> String {
39        format!("build ansible environment '{}'", self.env_name)
40    }
41}
42
43/// Copy the ansible environment to the control node.
44///
45/// Reads: `build-ansible-env:{env_name}` → store path
46/// Writes: `copy-ansible-env:{env_name}` → store path
47pub struct NixCopyAnsibleEnvTask {
48    pub env_name: String,
49    pub target_host: String,
50    pub target_user: String,
51}
52
53impl DagTask for NixCopyAnsibleEnvTask {
54    fn execute(&self, ctx: &DagContext) -> TaskOutcome {
55        let store_path: String =
56            match ctx.get_output(&TaskId(format!("build-ansible-env:{}", self.env_name))) {
57                Some(p) => p,
58                None => return TaskOutcome::Failed("no ansible env build output".into()),
59            };
60
61        let store_uri = format!("ssh-ng://{}@{}", self.target_user, self.target_host);
62        match copy::copy_closure(&store_path, &store_uri) {
63            Ok(()) => {
64                ctx.set_output(
65                    TaskId(format!("copy-ansible-env:{}", self.env_name)),
66                    store_path,
67                );
68                TaskOutcome::Success
69            }
70            Err(e) => TaskOutcome::Failed(format!("copy ansible env: {}", e)),
71        }
72    }
73
74    fn describe(&self) -> String {
75        format!(
76            "copy ansible env '{}' to {}",
77            self.env_name, self.target_host
78        )
79    }
80}
81
82/// Run an ansible playbook against a specific host.
83///
84/// Reads: `copy-ansible-env:{env_name}` → ansible store path
85pub struct AnsiblePlaybookTask {
86    pub host: String,
87    pub playbook: String,
88    pub env_name: String,
89    pub check_mode: bool,
90}
91
92impl AnsiblePlaybookTask {
93    pub fn new(host: &str, playbook: &str, env_name: &str) -> Self {
94        Self {
95            host: host.to_string(),
96            playbook: playbook.to_string(),
97            env_name: env_name.to_string(),
98            check_mode: false,
99        }
100    }
101
102    pub fn with_check(mut self, check: bool) -> Self {
103        self.check_mode = check;
104        self
105    }
106}
107
108impl DagTask for AnsiblePlaybookTask {
109    fn execute(&self, ctx: &DagContext) -> TaskOutcome {
110        let ansible_env: String =
111            match ctx.get_output(&TaskId(format!("copy-ansible-env:{}", self.env_name))) {
112                Some(p) => p,
113                None => return TaskOutcome::Failed("no ansible env in context".into()),
114            };
115
116        let ansible_bin = format!("{}/bin/ansible-playbook", ansible_env);
117
118        let mut cmd = Command::new(&ansible_bin);
119        cmd.args(["--limit", &self.host, &self.playbook]);
120
121        if self.check_mode {
122            cmd.arg("--check");
123        }
124
125        let output = match cmd.output() {
126            Ok(o) => o,
127            Err(e) => return TaskOutcome::Failed(format!("failed to run ansible: {}", e)),
128        };
129
130        if output.status.success() {
131            TaskOutcome::Success
132        } else {
133            let stderr = String::from_utf8_lossy(&output.stderr);
134            TaskOutcome::Failed(format!(
135                "playbook failed on {}: {}",
136                self.host,
137                stderr.trim()
138            ))
139        }
140    }
141
142    fn describe(&self) -> String {
143        format!("run {} on {}", self.playbook, self.host)
144    }
145}
146
147/// Optional post-playbook verification.
148pub struct AnsibleVerifyTask {
149    pub host: String,
150    pub check_command: String,
151}
152
153impl DagTask for AnsibleVerifyTask {
154    fn execute(&self, _ctx: &DagContext) -> TaskOutcome {
155        let output = Command::new("ssh")
156            .args([
157                "-oStrictHostKeyChecking=no",
158                "-oPasswordAuthentication=no",
159                "-oConnectTimeout=10",
160                &self.host,
161                &self.check_command,
162            ])
163            .output();
164
165        match output {
166            Ok(o) if o.status.success() => TaskOutcome::Success,
167            Ok(o) => {
168                let stderr = String::from_utf8_lossy(&o.stderr);
169                TaskOutcome::Failed(format!("verify failed on {}: {}", self.host, stderr.trim()))
170            }
171            Err(e) => TaskOutcome::Failed(format!("verify failed on {}: {}", self.host, e)),
172        }
173    }
174
175    fn describe(&self) -> String {
176        format!("verify {}", self.host)
177    }
178}
consortium_ansible/tasks.rs

consortium_ansible/
tasks.rs
