olai_stack_runtime/
instances.rs

1//! [`Instances`]: running revisions from the [`Store`] as compose projects.
2
3use std::collections::{BTreeMap, BTreeSet};
4use std::path::{Path, PathBuf};
5use std::sync::Arc;
6use std::time::Duration;
7
8use futures::StreamExt;
9use olai_stack_state::{Desired, Instance, InstanceId, Revision, Store};
10use olai_stack_topology::{
11    Catalog, LOCAL_CREDENTIALS, LaunchEnvError, LocalCredentials, LocalSecrets, MaterializedOutput,
12    Plan, PlanError, RenderError, RuntimePlacement, Secrets, Vantage, baseline_catalog_with,
13};
14use serde::Serialize;
15
16use crate::effective::{EffectiveEnv, EnvLayers, Layer};
17use crate::engine::{ComposeEngine, DownOptions, EngineStream, Progress, UpOptions};
18use crate::events::{EngineEvent, EventAction, EventFilter};
19use crate::ports::{BusyPort, probe_host_ports};
20use crate::secrets::{SecretError, SecretProviders};
21use crate::source::{Source, SourceCtx, SourceError};
22use crate::status::Health;
23
24/// The directory under the state directory instances are materialized into.
25pub const INSTANCES_DIR: &str = "instances";
26
27/// Keys the placement reserves in an instance's environment: no layer overrides them.
28pub const PLACEMENT_KEYS: &[&str] = &[
29    "COMPOSE_PROJECT_NAME",
30    "TRESTLE_INSTANCE",
31    "TRESTLE_SLOT",
32    "TRESTLE_HOST_LABEL",
33];
34
35/// What can go wrong running an instance.
36#[derive(Debug, thiserror::Error)]
37pub enum InstanceError {
38    /// The store failed.
39    #[error(transparent)]
40    Store(#[from] olai_stack_state::Error),
41    /// The engine failed.
42    #[error(transparent)]
43    Engine(#[from] crate::Error),
44    /// A module manifest the revision embeds doesn't parse.
45    #[error("module manifest `{path}` of the revision: {reason}")]
46    Module {
47        /// The path the revision lists it under.
48        path: String,
49        /// The parse error.
50        reason: String,
51    },
52    /// The revision doesn't plan.
53    #[error(transparent)]
54    Plan(#[from] PlanError),
55    /// The plan doesn't render.
56    #[error(transparent)]
57    Render(#[from] RenderError),
58    /// The rendered output would need a credential on disk.
59    #[error(transparent)]
60    LaunchEnv(#[from] LaunchEnvError),
61    /// A secret couldn't be resolved.
62    #[error(transparent)]
63    Secret(#[from] SecretError),
64    /// A source failed, or the instance names one that isn't configured.
65    #[error(transparent)]
66    Source(#[from] SourceError),
67    /// A local credential the environment sets isn't valid.
68    #[error("local credential: {0}")]
69    LocalCredential(String),
70    /// Host ports the instance would bind are taken.
71    #[error("host ports are already in use: {}", ports(.0))]
72    PortsBusy(Vec<BusyPort>),
73    /// Another instance of the same revision is running (they would share a compose project).
74    #[error("instance {running} already runs revision {revision} of this environment")]
75    RevisionRunning {
76        /// The revision.
77        revision: u32,
78        /// The instance running it.
79        running: InstanceId,
80    },
81    /// Writing the materialized tree failed.
82    #[error("{path}: {source}")]
83    Io {
84        /// The file or directory.
85        path: PathBuf,
86        /// The I/O error.
87        #[source]
88        source: std::io::Error,
89    },
90}
91
92fn ports(busy: &[BusyPort]) -> String {
93    busy.iter()
94        .map(|b| format!("{} ({})", b.port, b.owner))
95        .collect::<Vec<_>>()
96        .join(", ")
97}
98
99type Result<T, E = InstanceError> = std::result::Result<T, E>;
100
101/// What happened to one service of an instance.
102#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
103pub struct InstanceEvent {
104    /// The instance.
105    pub instance: InstanceId,
106    /// The compose service.
107    pub service: String,
108    /// What happened (see [`EventAction`]).
109    pub action: EventAction,
110}
111
112/// What [`Instances::reconcile`] found.
113#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)]
114pub struct Reconciled {
115    /// Instances found running and marked as such.
116    pub adopted: Vec<InstanceId>,
117    /// Instances the store wanted running whose project is gone, now marked stopped.
118    pub stale: Vec<InstanceId>,
119    /// Instance-shaped projects no store instance accounts for.
120    pub orphans: Vec<String>,
121    /// Orphans that were taken down (with [`ReconcileOptions::remove_orphans`]).
122    pub removed: Vec<String>,
123}
124
125/// How [`Instances::reconcile`] treats what it finds.
126#[derive(Clone, Copy, Debug, Default)]
127pub struct ReconcileOptions {
128    /// Take down orphaned projects rather than only report them.
129    pub remove_orphans: bool,
130}
131
132/// What [`Instances::materialize`] produced.
133#[derive(Debug)]
134pub struct Materialized {
135    /// Where the tree was written.
136    pub dir: PathBuf,
137    /// The plan.
138    pub plan: Plan,
139    /// The resolved environment.
140    pub env: EffectiveEnv,
141    /// Everything Compose needs in its process environment: the effective environment plus
142    /// the credentials moved out of the tree. Holds secrets; never print it.
143    launch: BTreeMap<String, String>,
144}
145
146impl Materialized {
147    /// The variable names the instance launches with (values withheld: they hold secrets).
148    pub fn launch_var_names(&self) -> impl Iterator<Item = &str> {
149        self.launch.keys().map(String::as_str)
150    }
151}
152
153/// Runs the store's instances on an engine. Each instance runs one revision in its own compose
154/// project, on its own slot and under its own host label, so two revisions of one environment
155/// run side by side:
156///
157/// - project `<workspace>_<env>_r<revision>`: trestle names can't contain `_`, so a project
158///   name parses back to its parts and [`reconcile`](Self::reconcile) tells instance projects
159///   from any other;
160/// - host label `<workspace>-<env>-r<revision>` (`demo-dev-r2.localhost`);
161/// - the slot the store leases the instance, host-wide.
162///
163/// The rendered tree lives in `<state>/instances/<id>/` (owner-only) and is a cache,
164/// re-rendered from the revision on every start. It holds no credential: they reach Compose as
165/// process environment only (see [`MaterializedOutput::into_launch_env`]).
166pub struct Instances<E> {
167    store: Arc<Store>,
168    engine: E,
169    root: PathBuf,
170    providers: SecretProviders,
171    sources: BTreeMap<String, Arc<dyn Source>>,
172    probe_ports: bool,
173    wait_timeout: Option<Duration>,
174}
175
176/// The compose project of revision `revision` of `env` in `workspace`.
177pub fn instance_project(workspace: &str, env: &str, revision: u32) -> String {
178    format!("{workspace}_{env}_r{revision}")
179}
180
181/// `(workspace, env, revision)` of an instance project name, if it is one.
182pub fn parse_instance_project(project: &str) -> Option<(&str, &str, u32)> {
183    let mut parts = project.split('_');
184    let (workspace, env, revision) = (parts.next()?, parts.next()?, parts.next()?);
185    if parts.next().is_some() {
186        return None;
187    }
188    let revision = revision.strip_prefix('r')?.parse().ok()?;
189    let valid = olai_stack_topology::workspace::is_valid_name;
190    (valid(workspace) && valid(env)).then_some((workspace, env, revision))
191}
192
193impl<E: ComposeEngine> Instances<E> {
194    /// Run `store`'s instances on `engine`, materializing them under `state_dir`
195    /// (`<state_dir>/instances/<id>/`). Secrets resolve with the `env` provider only until
196    /// [`with_providers`](Self::with_providers) says otherwise.
197    pub fn new(store: Arc<Store>, engine: E, state_dir: impl Into<PathBuf>) -> Self {
198        Instances {
199            store,
200            engine,
201            root: state_dir.into().join(INSTANCES_DIR),
202            providers: SecretProviders::new().with(crate::secrets::EnvProvider),
203            sources: BTreeMap::new(),
204            probe_ports: true,
205            wait_timeout: None,
206        }
207    }
208
209    /// Resolve secrets with `providers` (and no others).
210    pub fn with_providers(mut self, providers: SecretProviders) -> Self {
211        self.providers = providers;
212        self
213    }
214
215    /// Make `source` available to instances that name it.
216    pub fn with_source(mut self, source: impl Source + 'static) -> Self {
217        self.sources
218            .insert(source.name().to_owned(), Arc::new(source));
219        self
220    }
221
222    /// Don't check host ports before starting (tests on a shared host, a fake engine).
223    pub fn without_port_probe(mut self) -> Self {
224        self.probe_ports = false;
225        self
226    }
227
228    /// Give up waiting for health after `timeout` on start.
229    pub fn with_wait_timeout(mut self, timeout: Duration) -> Self {
230        self.wait_timeout = Some(timeout);
231        self
232    }
233
234    /// The engine.
235    pub fn engine(&self) -> &E {
236        &self.engine
237    }
238
239    /// The store.
240    pub fn store(&self) -> &Store {
241        &self.store
242    }
243
244    /// Where the instance's tree is materialized.
245    pub fn dir(&self, id: &InstanceId) -> PathBuf {
246        self.root.join(&id.0)
247    }
248
249    /// Where the instance runs: its compose project, host label, mesh and slot. Leases it a
250    /// slot if it has none.
251    pub fn placement(&self, id: &InstanceId) -> Result<RuntimePlacement> {
252        let (instance, workspace, env) = self.describe(id)?;
253        let slot = match instance.slot {
254            Some(slot) => slot,
255            None => self.store.lease_slot(id)?.slot,
256        };
257        Ok(RuntimePlacement {
258            slot,
259            project_name: instance_project(&workspace.name, &env.name, instance.revision),
260            mesh_network: workspace
261                .mesh
262                .then(|| format!("trestle-{}-mesh", workspace.name)),
263            host_label: format!("{}-{}-r{}", workspace.name, env.name, instance.revision),
264        })
265    }
266
267    fn describe(
268        &self,
269        id: &InstanceId,
270    ) -> Result<(
271        Instance,
272        olai_stack_state::Workspace,
273        olai_stack_state::Environment,
274    )> {
275        let instance = self.store.instance(id)?;
276        let env = self.store.environment(&instance.environment)?;
277        let workspace = self.store.workspace_by_id(&env.workspace)?;
278        Ok((instance, workspace, env))
279    }
280
281    /// The layers of the instance's environment that don't depend on its plan.
282    async fn base_layers(
283        &self,
284        id: &InstanceId,
285    ) -> Result<(EnvLayers, Revision, RuntimePlacement)> {
286        let (instance, workspace, env) = self.describe(id)?;
287        let revision = self
288            .store
289            .revision(&instance.environment, instance.revision)?;
290        let placement = self.placement(id)?;
291        let defaults: BTreeMap<String, String> = LOCAL_CREDENTIALS
292            .iter()
293            .map(|c| (c.env_var(), c.default.to_owned()))
294            .collect();
295        let reserved: BTreeMap<String, String> = [
296            ("COMPOSE_PROJECT_NAME", placement.project_name.clone()),
297            ("TRESTLE_INSTANCE", id.0.clone()),
298            ("TRESTLE_SLOT", placement.slot.to_string()),
299            ("TRESTLE_HOST_LABEL", placement.host_label.clone()),
300        ]
301        .into_iter()
302        .map(|(k, v)| (k.to_owned(), v))
303        .collect();
304        let mut layers = EnvLayers::new()
305            .with_secrets(Layer::CatalogDefaults, &defaults)
306            .with_plain(Layer::Placement, &reserved)
307            .with_vars(Layer::WorkspaceVars, &workspace.vars)
308            .with_vars(Layer::EnvironmentVars, &env.vars)
309            .with_vars(Layer::RevisionVars, revision.vars())
310            .with_vars(Layer::InstanceOverrides, &instance.overrides);
311        let ctx = SourceCtx {
312            instance: id.clone(),
313            workspace: workspace.name.clone(),
314            environment: env.name.clone(),
315            revision: instance.revision,
316        };
317        for name in &instance.sources {
318            let source = self.sources.get(name).ok_or_else(|| SourceError {
319                source_name: name.clone(),
320                reason: "no such source is configured".into(),
321            })?;
322            let mut overlay = source.overlay(&ctx).await?;
323            if overlay.is_expired(std::time::SystemTime::now()) {
324                overlay = source.refresh(&ctx, &overlay).await?;
325            }
326            layers = layers.with_overlay(name, &overlay);
327        }
328        Ok((layers, revision, placement))
329    }
330
331    /// Plan the instance's revision in its placement under `env` (its secrets and local
332    /// credentials come from there).
333    fn plan(
334        &self,
335        revision: &Revision,
336        env_name: &str,
337        placement: &RuntimePlacement,
338        env: &EffectiveEnv,
339    ) -> Result<Plan> {
340        let secrets: Secrets = env.launch_vars().into_iter().collect();
341        let credentials = LocalCredentials::from_secrets(&secrets)
342            .map_err(|e| InstanceError::LocalCredential(e.to_string()))?;
343        let mut catalog = baseline_catalog_with(&credentials);
344        for (path, text) in revision.modules() {
345            let custom =
346                Catalog::from_manifests([text.as_str()]).map_err(|e| InstanceError::Module {
347                    path: path.clone(),
348                    reason: e.to_string(),
349                })?;
350            catalog = catalog.merge(custom);
351        }
352        let mut ctx = revision.manifest().context.clone();
353        ctx.env_name = env_name.to_owned();
354        placement.apply(&mut ctx);
355        // Credentials go to Compose as process env, never inline in a file.
356        ctx.local_secrets = LocalSecrets::Separate;
357        ctx.secrets = Some(secrets);
358        Ok(catalog.plan(&revision.manifest().selection, &ctx)?)
359    }
360
361    /// The instance's environment, every variable with its provenance (see [`EffectiveEnv`]).
362    pub async fn effective_env(&self, id: &InstanceId) -> Result<EffectiveEnv> {
363        Ok(self.resolve(id).await?.0)
364    }
365
366    async fn resolve(&self, id: &InstanceId) -> Result<(EffectiveEnv, Plan)> {
367        let (layers, revision, placement) = self.base_layers(id).await?;
368        let env_name = self.store.environment(revision.environment())?.name;
369        // The plan's own variables are layers too, but planning needs the credentials the
370        // other layers resolve to: plan with those, then add the plan's layers and resolve
371        // again.
372        let first = layers.resolve(&self.providers)?;
373        let plan = self.plan(&revision, &env_name, &placement, &first)?;
374        // The plan's variables can carry a credential under a key of their own (a module's
375        // password variable, say): those are secrets too.
376        let secrets: Vec<&str> = first
377            .vars()
378            .values()
379            .filter(|r| r.value.is_secret() && !r.value.expose().is_empty())
380            .map(|r| r.value.expose())
381            .collect();
382        let split = |env: &olai_stack_topology::InjectedEnv| {
383            let (secret, plain): (BTreeMap<_, _>, BTreeMap<_, _>) = env
384                .iter()
385                .map(|(k, v)| (k.to_owned(), v.to_owned()))
386                .partition(|(_, v)| secrets.iter().any(|s| v.contains(s)));
387            (secret, plain)
388        };
389        let (injected_secret, injected) = split(&plan.env);
390        let (client_secret, client) = split(plan.client_env(Vantage::Host));
391        let env = layers
392            .with_plain(Layer::RevisionManifest, &injected)
393            .with_secrets(Layer::RevisionManifest, &injected_secret)
394            .with_plain(Layer::ClientEnv, &client)
395            .with_secrets(Layer::ClientEnv, &client_secret)
396            .resolve(&self.providers)?;
397        Ok((env, plan))
398    }
399
400    /// Render the instance into its directory, from scratch, and return what it launches
401    /// with. The tree holds no credential.
402    pub async fn materialize(&self, id: &InstanceId) -> Result<Materialized> {
403        let (env, plan) = self.resolve(id).await?;
404        let (output, launch_env) = plan.materialize()?.into_launch_env()?;
405        let dir = self.dir(id);
406        write_tree(&dir, &output)?;
407        let mut launch = env.launch_vars();
408        launch.extend(launch_env.into_vars());
409        Ok(Materialized {
410            dir,
411            plan,
412            env,
413            launch,
414        })
415    }
416
417    /// Start the instance: lease it a slot (if it has none), check its host ports, render it
418    /// afresh and `up` it, waiting for health. The store records it as wanted running first,
419    /// so a crash mid-start is reconciled as running.
420    pub async fn start(&self, id: &InstanceId) -> Result<Materialized> {
421        let instance = self.store.instance(id)?;
422        let project = self.placement(id)?.project_name;
423        for other in self.store.instances()? {
424            if other.id != *id
425                && other.environment == instance.environment
426                && other.revision == instance.revision
427                && other.desired == Desired::Running
428            {
429                return Err(InstanceError::RevisionRunning {
430                    revision: instance.revision,
431                    running: other.id,
432                });
433            }
434        }
435        let materialized = self.materialize(id).await?;
436        if self.probe_ports {
437            let running = self
438                .engine
439                .ls()
440                .await?
441                .into_iter()
442                .any(|p| p.name == project && p.is_running());
443            // A running project holds its own ports.
444            if !running {
445                let busy = probe_host_ports(&materialized.plan);
446                if !busy.is_empty() {
447                    return Err(InstanceError::PortsBusy(busy));
448                }
449            }
450        }
451        self.store.set_desired(id, Desired::Running)?;
452        let opts = UpOptions {
453            wait: true,
454            wait_timeout: self.wait_timeout,
455            env: materialized.launch.clone(),
456            progress: Progress::Quiet,
457        };
458        self.engine.up(&materialized.dir, &opts).await?;
459        Ok(materialized)
460    }
461
462    /// Stop the instance (`down`, with its volumes when `volumes`), keeping its slot.
463    pub async fn stop(&self, id: &InstanceId, volumes: bool) -> Result<()> {
464        // Compose interpolates the files on `down` too, so it needs the launch env again.
465        let materialized = self.materialize(id).await?;
466        let opts = DownOptions {
467            volumes,
468            env: materialized.launch,
469            progress: Progress::Quiet,
470        };
471        self.engine.down(&materialized.dir, &opts).await?;
472        self.store.set_desired(id, Desired::Stopped)?;
473        Ok(())
474    }
475
476    /// Stop the instance if it runs, drop its tree, and delete it from the store (releasing
477    /// its slot).
478    pub async fn remove(&self, id: &InstanceId, volumes: bool) -> Result<()> {
479        let project = self.placement(id)?.project_name;
480        if self.engine.ls().await?.iter().any(|p| p.name == project) {
481            self.stop(id, volumes).await?;
482        }
483        let dir = self.dir(id);
484        if dir.exists() {
485            std::fs::remove_dir_all(&dir)
486                .map_err(|source| InstanceError::Io { path: dir, source })?;
487        }
488        self.store.delete_instance(id)?;
489        Ok(())
490    }
491
492    /// The health of each of the instance's services (empty when it doesn't run).
493    pub async fn health(&self, id: &InstanceId) -> Result<BTreeMap<String, Health>> {
494        let project = self.placement(id)?.project_name;
495        Ok(self
496            .engine
497            .ps(&project)
498            .await?
499            .into_iter()
500            .map(|s| (s.service, s.health))
501            .collect())
502    }
503
504    /// The instance's events from now on: its containers' lifecycle and health changes.
505    pub async fn events(&self, id: &InstanceId) -> Result<EngineStream<InstanceEvent>> {
506        let project = self.placement(id)?.project_name;
507        let id = id.clone();
508        Ok(self
509            .engine
510            .events(&EventFilter::Project(project))
511            .await?
512            .map(move |e: crate::Result<EngineEvent>| {
513                e.map(|e| InstanceEvent {
514                    instance: id.clone(),
515                    service: e.service,
516                    action: e.action,
517                })
518            })
519            .boxed())
520    }
521
522    /// Match what Docker runs against the store, after a restart of the runtime: a running
523    /// instance project is adopted (its instance marked running), an instance wanted running
524    /// whose project is gone is marked stopped, and an instance-shaped project the store
525    /// doesn't know is an orphan (taken down with [`ReconcileOptions::remove_orphans`]).
526    pub async fn reconcile(&self, opts: ReconcileOptions) -> Result<Reconciled> {
527        let projects = self.engine.ls().await?;
528        let mut by_project: BTreeMap<String, InstanceId> = BTreeMap::new();
529        let instances = self.store.instances()?;
530        for instance in &instances {
531            let env = self.store.environment(&instance.environment)?;
532            let ws = self.store.workspace_by_id(&env.workspace)?;
533            by_project.insert(
534                instance_project(&ws.name, &env.name, instance.revision),
535                instance.id.clone(),
536            );
537        }
538        let mut out = Reconciled::default();
539        let mut present = BTreeSet::new();
540        for project in &projects {
541            if parse_instance_project(&project.name).is_none() {
542                continue;
543            }
544            present.insert(project.name.clone());
545            match by_project.get(&project.name) {
546                Some(id) if project.is_running() => {
547                    // Several instances of one revision share a project name; the first
548                    // wanted running (or the first) is the one that runs it.
549                    self.store.set_desired(id, Desired::Running)?;
550                    out.adopted.push(id.clone());
551                }
552                Some(_) => {}
553                None => {
554                    out.orphans.push(project.name.clone());
555                    if opts.remove_orphans && self.remove_orphan(&project.config_files).await? {
556                        out.removed.push(project.name.clone());
557                    }
558                }
559            }
560        }
561        for instance in instances {
562            if instance.desired != Desired::Running {
563                continue;
564            }
565            let project = by_project
566                .iter()
567                .find(|(_, id)| **id == instance.id)
568                .map(|(p, _)| p.clone())
569                .unwrap_or_default();
570            if !present.contains(&project) {
571                self.store.set_desired(&instance.id, Desired::Stopped)?;
572                out.stale.push(instance.id);
573            }
574        }
575        Ok(out)
576    }
577
578    /// Take down the project started from `config_files`; `false` when Compose didn't say
579    /// where that was (or it is gone), so there is nothing to run `down` in.
580    async fn remove_orphan(&self, config_files: &str) -> Result<bool> {
581        // Compose lists the files it started the project from; `down` needs their directory.
582        let Some(dir) = config_files
583            .split(',')
584            .next()
585            .filter(|f| !f.is_empty())
586            .map(Path::new)
587            .and_then(Path::parent)
588            .filter(|d| d.is_dir())
589        else {
590            return Ok(false);
591        };
592        let opts = DownOptions {
593            progress: Progress::Quiet,
594            ..DownOptions::default()
595        };
596        self.engine.down(dir, &opts).await?;
597        Ok(true)
598    }
599}
600
601/// Write `output` into `dir`, replacing whatever was there, with the directory owner-only.
602fn write_tree(dir: &Path, output: &MaterializedOutput) -> Result<()> {
603    let io = |path: &Path| {
604        let path = path.to_owned();
605        move |source| InstanceError::Io { path, source }
606    };
607    if dir.exists() {
608        std::fs::remove_dir_all(dir).map_err(io(dir))?;
609    }
610    std::fs::create_dir_all(dir).map_err(io(dir))?;
611    #[cfg(unix)]
612    {
613        use std::os::unix::fs::PermissionsExt as _;
614        std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)).map_err(io(dir))?;
615    }
616    output.write_to(dir).map_err(io(dir))
617}
618
619#[cfg(test)]
620mod tests {
621    use super::*;
622
623    #[test]
624    fn instance_projects_parse_back() {
625        let name = instance_project("demo-x", "prod", 3);
626        assert_eq!(name, "demo-x_prod_r3");
627        assert_eq!(parse_instance_project(&name), Some(("demo-x", "prod", 3)));
628        for other in ["demo-dev", "a_b", "a_b_c", "a_b_r", "A_b_r1", "a_b_r1_x"] {
629            assert_eq!(parse_instance_project(other), None, "{other}");
630        }
631    }
632}
olai_stack_runtime/instances.rs

olai_stack_runtime/
instances.rs
