isothrum_runner/
reap.rs

1//! The reaper's half in the runner, so an abandoned run can't keep billing:
2//!
3//! - [`Runner::expired`]: the recorded runs whose expiry passed and that may still hold
4//!   something. Destroying one is its `down`, which replays the params it was applied with (the
5//!   embedder gives the connection, as for any operation).
6//! - [`Runner::orphans`]: what a cloud still holds tagged past its expiry that no recorded run's
7//!   state covers, read through the cloud's own command line (its argv, never a shell). An orphan
8//!   is only listed, never deleted: no state says what it belongs to, so a person removes it.
9//!
10//! The sweep's schedule is the embedder's (the launcher sweeps every five minutes; the
11//! `isothrum reap` command once).
12
13mod tags;
14
15use std::collections::BTreeSet;
16use std::time::{Duration, SystemTime};
17
18use isoloom_manifest::Cloud;
19pub use tags::{Place, Tagged};
20
21use crate::Result;
22use crate::clock::unix_seconds;
23use crate::process::{ProcessRunner, ProcessSpec, capture};
24use crate::runner::Runner;
25use crate::state::RunState;
26use crate::store::RunRecord;
27
28/// How long a cloud's query may take.
29const QUERY_LIMIT: Duration = Duration::from_secs(120);
30
31/// What a cloud still holds past its expiry that no recorded run covers.
32#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
33#[non_exhaustive]
34pub struct Orphans {
35    /// The resources, as the cloud lists them.
36    pub resources: Vec<Tagged>,
37    /// Why a query couldn't be read: its command line missing or failing, or an answer of
38    /// another shape. The cloud wasn't checked there.
39    pub unread: Vec<String>,
40}
41
42impl Runner {
43    /// The recorded runs whose expiry passed at `now` and whose last operation didn't leave them
44    /// absent: the reaper's input. Reads files only.
45    ///
46    /// # Errors
47    ///
48    /// Fails with `io` when a record can't be read.
49    pub fn expired(&self, now: SystemTime) -> Result<Vec<RunRecord>> {
50        let now = unix_seconds(now);
51        let records = self.store.records()?;
52        Ok(records
53            .into_iter()
54            .filter(|r| r.state != RunState::Absent && r.expires_at.is_some_and(|at| at <= now))
55            .collect())
56    }
57
58    /// What `cloud` holds in `place` tagged with an expiry past `now` that no recorded run's
59    /// Terraform state covers. A command line that is missing or fails is said in
60    /// [`Orphans::unread`], never an error: the sweep goes on.
61    ///
62    /// # Errors
63    ///
64    /// Fails with `io` when a record or a run's state can't be read.
65    pub async fn orphans(&self, cloud: Cloud, place: &Place, now: SystemTime) -> Result<Orphans> {
66        let covered = self.covered(cloud)?;
67        let now = unix_seconds(now);
68        let mut out = Orphans::default();
69        for spec in tags::queries(cloud, place) {
70            let answer = query(self.config.processes.as_ref(), &spec).await.and_then(|text| tags::parse(cloud, &text));
71            match answer {
72                Ok(found) => out.resources.extend(found.into_iter().filter(|t| orphaned(t, now, &covered))),
73                Err(why) => out.unread.push(format!("`{}`: {why}", spec.line())),
74            }
75        }
76        Ok(out)
77    }
78
79    /// The environments the states of `cloud`'s recorded runs hold.
80    fn covered(&self, cloud: Cloud) -> Result<BTreeSet<String>> {
81        let mut out = BTreeSet::new();
82        for record in self.store.records()?.into_iter().filter(|r| r.cloud == Some(cloud)) {
83            let path = self.store.run_dir(&record.name)?.join("terraform.tfstate");
84            match std::fs::read_to_string(&path) {
85                Ok(text) => {
86                    if let Ok(state) = serde_json::from_str(&text) {
87                        tags::instances_in_state(&state, &mut out);
88                    }
89                }
90                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
91                Err(e) => return Err(crate::error::Error::io(&path, "read")(e)),
92            }
93        }
94        Ok(out)
95    }
96}
97
98/// Whether `t` is past its expiry at `now` and no recorded run covers its environment.
99fn orphaned(t: &Tagged, now: u64, covered: &BTreeSet<String>) -> bool {
100    t.expires_at.is_some_and(|at| at <= now) && t.instance.as_ref().is_none_or(|i| !covered.contains(i))
101}
102
103/// What `spec` prints, when it ran to a success within [`QUERY_LIMIT`]; else why not.
104async fn query(processes: &dyn ProcessRunner, spec: &ProcessSpec) -> std::result::Result<String, String> {
105    let ran = tokio::time::timeout(QUERY_LIMIT, capture(processes, spec)).await;
106    let out = ran
107        .map_err(|_| format!("no answer within {}s", QUERY_LIMIT.as_secs()))?
108        .map_err(|e| e.to_string())?;
109    if out.exit.is_some_and(crate::process::Exit::success) {
110        Ok(out.stdout.join("\n"))
111    } else {
112        Err(format!("failed: {}", out.stderr.join(" ").trim()))
113    }
114}
115
116#[cfg(test)]
117mod tests;
isothrum_runner/reap.rs

isothrum_runner/
reap.rs
