isothrum_runner/
run.rs

1//! One manifest opened under one run name: the operations an embedder calls. `operation` is what
2//! every operation does around its own work, `up` starts a run, `existing` holds the operations on
3//! a run that exists (down, park, resume, check, provision).
4
5mod attach;
6mod existing;
7mod operation;
8mod seal;
9#[cfg(test)]
10mod tests;
11mod up;
12
13pub use attach::AttachReport;
14
15use std::collections::BTreeMap;
16use std::path::Path;
17
18use isoloom_manifest::{Manifest, RunVar};
19
20use crate::Result;
21use crate::driver::AnyDriver;
22use crate::error::Error;
23use crate::events::Emitter;
24use crate::events::WarningCode;
25use crate::executor::{Cx, Stale};
26use crate::hooks::{Edge, RunView};
27use crate::manifest::LoadedManifest;
28use crate::name::RunName;
29use crate::options::RunOptions;
30use crate::outputs;
31use crate::params::params_for_existing;
32use crate::runner::Runner;
33use crate::scope::Scope;
34use crate::secrets::Secrets;
35use crate::state::Op;
36use crate::status::{self, OutputValue, Status};
37
38/// What a read-only query runs with: the run's record, its provider, and its params.
39type QueryPrelude = (Option<crate::store::RunRecord>, Option<String>, BTreeMap<String, String>);
40
41/// A run: a manifest under a name, on a runner.
42pub struct Running<'r> {
43    pub(crate) runner: &'r Runner,
44    loaded: LoadedManifest,
45    name: RunName,
46    pub(crate) driver: AnyDriver,
47    /// Values of the manifest's connection slots, given once ([`Running::with_connection`]).
48    connection: Secrets,
49}
50
51impl std::fmt::Debug for Running<'_> {
52    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53        f.debug_struct("Running")
54            .field("name", &self.name)
55            .field("manifest", &self.loaded.path)
56            .finish_non_exhaustive()
57    }
58}
59
60impl<'r> Running<'r> {
61    pub(crate) fn new(runner: &'r Runner, loaded: LoadedManifest, name: RunName) -> Result<Running<'r>> {
62        if !loaded.manifest.hosts.contains(&runner.config.host) {
63            let target = serde_json::to_value(loaded.manifest.target).unwrap_or_default();
64            return Err(Error::Unsupported {
65                what: format!("target {target} on this host"),
66            });
67        }
68        let driver = AnyDriver::for_manifest(&loaded.manifest)?;
69        Ok(Running {
70            runner,
71            loaded,
72            name,
73            driver,
74            connection: Secrets::new(),
75        })
76    }
77
78    /// The run with `secrets`, the values of its connection slots (an ESXi host's password),
79    /// delivered to every operation: status and park, which take no options, too. Kept in memory
80    /// only, never stored; a value an operation's options give wins.
81    #[must_use]
82    pub fn with_connection(mut self, secrets: Secrets) -> Running<'r> {
83        self.connection = secrets;
84        self
85    }
86
87    /// `opts` with the connection's values its secrets lack.
88    pub(crate) fn connected(&self, opts: &RunOptions) -> RunOptions {
89        let mut out = opts.clone();
90        for (name, value) in self.connection.iter() {
91            if out.secrets.get(name).is_none() {
92                out.secrets.insert(name, value.clone());
93            }
94        }
95        out
96    }
97
98    /// The manifest.
99    #[must_use]
100    pub fn manifest(&self) -> &Manifest {
101        &self.loaded.manifest
102    }
103
104    /// The run's name.
105    #[must_use]
106    pub fn name(&self) -> &RunName {
107        &self.name
108    }
109
110    /// The manifest's file.
111    #[must_use]
112    pub fn path(&self) -> &Path {
113        &self.loaded.path
114    }
115
116    /// The manifest's folder (`${run.dir}`).
117    pub(crate) fn dir(&self) -> &Path {
118        &self.loaded.dir
119    }
120
121    /// The manifest's SHA-256.
122    pub(crate) fn loaded_sha256(&self) -> &str {
123        &self.loaded.sha256
124    }
125
126    /// What a read-only query runs with: the run's record, the provider it was started on, and
127    /// its params (recorded params replayed; a fresh run's defaults otherwise). No lock, no event.
128    pub(crate) fn query_prelude(&self) -> Result<QueryPrelude> {
129        let record = self.runner.store.record(&self.name)?;
130        let provider = record.as_ref().and_then(|r| r.provider.clone());
131        let params = params_for_existing(
132            self.manifest(),
133            record.as_ref().map(|r| &r.params),
134            &BTreeMap::new(),
135            &self.scope(provider.clone()),
136        )?;
137        Ok((record, provider, params))
138    }
139
140    /// The run as it stands: read-only, never locked, and never an event.
141    ///
142    /// # Errors
143    ///
144    /// Fails when the engine can't be asked (the status step fails or runs out of time) or the
145    /// run's record can't be read.
146    pub async fn status(&self) -> Result<Status> {
147        let (record, provider, params) = self.query_prelude()?;
148        let silent = |_: &crate::events::Event| {};
149        let events = Emitter::new(self.name.as_str(), Op::Check, self.runner.config.clock.as_ref(), &silent);
150        let secrets = self.connection.clone();
151        let mut cx = self.cx(params, &secrets, &events, provider.clone())?;
152        if let Some(r) = &record {
153            Self::bind_record(&mut cx, r);
154        }
155        self.driver.prepare(&cx)?;
156        self.relay(&mut cx).await?;
157        let observed = self.observe(&mut cx).await?;
158        cx.vars.set_published(observed.published.clone());
159        cx.vars.set_hosts(observed.hosts.clone());
160        let state = match self.runner.op_of(&self.name).and_then(Op::during) {
161            Some(during) => during,
162            None => status::derive_state(&observed.machines, record.as_ref()),
163        };
164        let outputs = outputs::read(self.manifest(), &cx.vars, &observed.outputs, &observed.sensitive);
165        outputs::bind(&mut cx.vars, &outputs);
166        let message = (state == crate::RunState::Ready)
167            .then(|| outputs::message(self.manifest(), &cx.vars, &outputs))
168            .flatten();
169        Ok(Status {
170            state,
171            provider,
172            machines: observed.machines,
173            networks: self.networks(&cx),
174            outputs,
175            message,
176        })
177    }
178
179    /// The run's outputs, read live.
180    ///
181    /// # Errors
182    ///
183    /// Fails as [`Running::status`] does.
184    pub async fn outputs(&self) -> Result<Vec<OutputValue>> {
185        Ok(self.status().await?.outputs)
186    }
187
188    /// What `params`, `secrets` and `events` give this run's steps.
189    pub(crate) fn cx<'a>(
190        &'a self,
191        params: BTreeMap<String, String>,
192        secrets: &'a Secrets,
193        events: &'a Emitter<'a>,
194        provider: Option<String>,
195    ) -> Result<Cx<'a>> {
196        let state_dir = self.runner.store.run_dir(&self.name)?;
197        let mut vars = crate::vars::Vars::default();
198        vars.set_run(RunVar::Name, self.name.as_str());
199        vars.set_run(RunVar::Dir, self.dir().display().to_string());
200        vars.set_run(RunVar::StateDir, state_dir.display().to_string());
201        vars.set_run(RunVar::KnownHosts, state_dir.join("known_hosts").display().to_string());
202        vars.set_run(RunVar::SshKey, crate::ssh::key_path(&self.runner.config.home).display().to_string());
203        if let Some(key) = crate::ssh::public_key(&self.runner.config.home) {
204            vars.set_run(RunVar::SshPublicKey, key);
205        }
206        if let Some(p) = &provider {
207            vars.set_run(RunVar::Provider, p.clone());
208        }
209        vars.set_params(params);
210        let stale = self.manifest().recovery.as_ref().map(|r| Stale::new(&r.stale_markers)).unwrap_or_default();
211        Ok(Cx {
212            run: self,
213            vars,
214            secrets,
215            events,
216            scope: self.scope(provider),
217            stale,
218            processes: std::sync::Arc::clone(&self.runner.config.processes),
219            ssh_jump: None,
220            expires_at: None,
221            relayed: None,
222        })
223    }
224
225    /// `cx` for an operation that acts on the run (never a status read): on a provider whose
226    /// management commands need elevation (Hyper-V on Windows), its tools run through one
227    /// elevated session, kept in the run's state folder.
228    pub(crate) fn acting<'a>(&'a self, mut cx: Cx<'a>) -> Result<Cx<'a>> {
229        if cx.scope.host == isoloom_manifest::HostOs::Windows && self.driver.elevates(&cx.scope) {
230            let dir = self.runner.store.run_dir(&self.name)?.join("elevated");
231            let own = std::env::vars_os().filter_map(|(k, v)| Some((k.into_string().ok()?, v.into_string().ok()?)));
232            let inner = std::sync::Arc::clone(&self.runner.config.processes);
233            cx.processes = std::sync::Arc::new(crate::process::Elevated::new(inner, dir, own));
234        }
235        Ok(cx)
236    }
237
238    /// Starts the macOS relays `cx`'s run needs (Vagrant on an ESXi host), for the operation.
239    pub(crate) async fn relay(&self, cx: &mut Cx<'_>) -> Result<()> {
240        let config = &self.runner.config;
241        let vagrant = self.manifest().driver == isoloom_manifest::Driver::Vagrant;
242        let params = cx.vars.params();
243        let Some(host) = crate::relay::esxi_host(config.host == isoloom_manifest::HostOs::Macos, vagrant, cx.scope.provider.as_deref(), params) else {
244            return Ok(());
245        };
246        let port = params.get("ESXI_HOSTPORT").and_then(|p| p.parse().ok()).unwrap_or(22);
247        let relayed = crate::relay::prepare(host, port, &config.home, config.path.as_deref()).await;
248        cx.relayed = Some(relayed.map_err(Error::io(config.home.join("relay"), "set up"))?);
249        Ok(())
250    }
251
252    /// Binds what `record` keeps of the run into `cx`: its slot (`${run.slot}`), its SSH jump and
253    /// its expiry.
254    pub(crate) fn bind_record(cx: &mut Cx<'_>, record: &crate::store::RunRecord) {
255        if let Some(slot) = record.slot {
256            cx.vars.set_run(RunVar::Slot, slot.to_string());
257        }
258        cx.ssh_jump.clone_from(&record.ssh_jump);
259        cx.expires_at = record.expires_at;
260    }
261
262    /// This host, with `provider`.
263    pub(crate) fn scope(&self, provider: Option<String>) -> Scope {
264        Scope {
265            host: self.runner.config.host,
266            provider,
267        }
268    }
269
270    /// Reads what the engine gives back (published ports, outputs) and binds it, so a template
271    /// can name it (`${outputs.ip}`): the run's outputs.
272    pub(crate) async fn read_outputs(&self, cx: &mut Cx<'_>) -> Result<Vec<status::OutputValue>> {
273        let observed = self.observe(cx).await?;
274        Ok(self.bind_observed(cx, observed))
275    }
276
277    /// The run as its engine reports it: the driver's machines, then, when the run's containers
278    /// run inside one of its VMs (`docker-vm`'s guest), those containers as read in the VM.
279    pub(crate) async fn observe(&self, cx: &mut Cx<'_>) -> Result<crate::driver::Observed> {
280        let mut observed = self.driver.observe(cx).await?;
281        if let Some(guest) = &self.manifest().guest {
282            crate::in_vm::observe(cx, guest, &mut observed).await;
283        }
284        Ok(observed)
285    }
286
287    /// Binds what `observed` gives back (published ports, outputs) into `cx`: the run's outputs.
288    pub(crate) fn bind_observed(&self, cx: &mut Cx<'_>, observed: crate::driver::Observed) -> Vec<status::OutputValue> {
289        cx.vars.set_published(observed.published);
290        cx.vars.set_hosts(observed.hosts);
291        let outputs = outputs::read(self.manifest(), &cx.vars, &observed.outputs, &observed.sensitive);
292        // A value the engine or the manifest keeps secret is masked in every later event.
293        cx.events.hide(outputs.iter().filter(|o| o.sensitive).filter_map(|o| o.value.clone()));
294        outputs::bind(&mut cx.vars, &outputs);
295        outputs
296    }
297
298    /// The run's networks, their native names filled.
299    fn networks(&self, cx: &Cx<'_>) -> Vec<status::NetworkView> {
300        status::networks(self.manifest(), |t| cx.vars.render(t, "networks.native").ok())
301    }
302
303    /// Runs the embedder's hooks at `edge`; each failure is a warning.
304    pub(crate) async fn hooks(&self, cx: &Cx<'_>, edge: Edge, outputs: &[OutputValue]) {
305        if self.runner.hooks.is_empty() {
306            return;
307        }
308        let networks = self.networks(cx);
309        let view = RunView {
310            name: &self.name,
311            manifest: self.manifest(),
312            networks: &networks,
313            outputs,
314        };
315        for failure in edge.run(&self.runner.hooks, &view).await {
316            cx.events.warning(WarningCode::HookFailed, &failure);
317        }
318    }
319}
isothrum_runner/run.rs

isothrum_runner/
run.rs
