Skip to main content

max / makenotwork

10.1 KB · 256 lines History Blame Raw
1 //! The [`Executor`] trait: a handle scoped to one `(host, capability set)`.
2 //!
3 //! Callers hold executors as `Arc<dyn Executor>` (one per node, built from
4 //! topology) so the concrete transport — [`crate::LocalExec`],
5 //! [`crate::SshExec`], or `AgentRpc` — is chosen per host without the caller
6 //! caring which.
7
8 use crate::capability::CapabilitySet;
9 use crate::remote::{LogSink, RunOutput};
10 use crate::step::{ObserveKind, Step};
11 use anyhow::Result;
12 use async_trait::async_trait;
13 use std::path::Path;
14 use tokio::process::Command;
15
16 /// Options controlling an rsync push/pull. Default = a plain `-az --partial`
17 /// mirror that never prunes the destination (safe for artifact collection).
18 /// Sando's release-dir deploy opts in to `delete` + `chmod` to keep its exact
19 /// behavior.
20 #[derive(Clone, Debug)]
21 pub struct SyncOpts {
22 /// `--delete`: prune files on the destination that are gone from the
23 /// source (a true mirror). Off by default.
24 pub delete: bool,
25 /// `--chmod=<spec>`: force destination permissions per-file.
26 pub chmod: Option<String>,
27 /// `-z`: compress in flight. On by default, which is right for source trees
28 /// and release dirs. Turn it OFF for already-compressed payloads (a `.gz`
29 /// dump, a `.dmg`) — `-z` then burns CPU on both ends to save ~nothing.
30 pub compress: bool,
31 /// `--partial`: keep a partially-transferred file so a retry can resume it.
32 /// On by default (worth it for a multi-hundred-MB artifact over a flaky
33 /// link). Turn it OFF when a truncated leftover is *dangerous* rather than
34 /// merely useless — e.g. fetching a DB dump, where a resumed transfer could
35 /// splice two different dumps into one plausible-looking file.
36 pub partial: bool,
37 }
38
39 impl Default for SyncOpts {
40 fn default() -> Self {
41 Self {
42 delete: false,
43 chmod: None,
44 compress: true,
45 partial: true,
46 }
47 }
48 }
49
50 impl SyncOpts {
51 /// The Sando release-dir mirror: prune stale assets and force exec bits the
52 /// way `deploy.rs` always has.
53 pub fn release_mirror() -> Self {
54 Self {
55 delete: true,
56 chmod: Some("Du=rwx,Dgo=rx,Fu=rw,Fgo=r,F+X".into()),
57 ..Self::default()
58 }
59 }
60
61 /// For payloads that are already compressed (`.gz`/`.dmg`/`.zip`): skip
62 /// `-z` so the transfer doesn't re-compress compressed bytes.
63 pub fn precompressed() -> Self {
64 Self {
65 compress: false,
66 ..Self::default()
67 }
68 }
69 }
70
71 /// A read-only host observation. v1 can synthesize these from SSH-streamed
72 /// commands; a resident `ops-agent` is a drop-in upgrade (E3). The variants
73 /// mirror the executor spec.
74 #[derive(Clone, Debug, PartialEq)]
75 pub enum ObserveEvent {
76 ProcessExited {
77 unit: String,
78 code: i32,
79 },
80 ResourceSample {
81 cpu: f64,
82 rss: u64,
83 disk: u64,
84 },
85 JournalLine {
86 unit: String,
87 line: String,
88 },
89 HealthChanged {
90 check: String,
91 from: String,
92 to: String,
93 },
94 }
95
96 /// A live stream of [`ObserveEvent`]s from a host's observe plane.
97 pub type EventStream = tokio::sync::mpsc::Receiver<ObserveEvent>;
98
99 #[async_trait]
100 pub trait Executor: Send + Sync {
101 /// Run a typed step, streaming merged stdout/stderr into `sink`.
102 ///
103 /// Returns [`crate::CapabilityDenied`] (boxed into the error) if
104 /// `step.action` is outside this executor's grant — checked *before* the
105 /// command is dispatched.
106 async fn run_streaming(&self, step: &Step, sink: &mut dyn LogSink) -> Result<RunOutput>;
107
108 /// Pull ONE FILE a prior step produced back to the caller: `remote` names
109 /// the file on the host, `local` the destination path.
110 ///
111 /// Split from [`Executor::pull_dir`] because the transports genuinely
112 /// differ here, and conflating them is a silent foot-gun: the rsync
113 /// transports need a trailing slash for directory-contents semantics and
114 /// must NOT have one for a file, while [`crate::AgentRpc`] streams a single
115 /// file over HTTP and cannot do directories at all. One `pull` taking either
116 /// shape meant a config-level transport swap (`HostTransport::Agent` →
117 /// `Ssh`) could turn a working file pull into `rsync host:/x/App.dmg/`.
118 ///
119 /// Contract is file→file. Pointing this at a *directory* is not checked and
120 /// not an error on the rsync transports (`-a` implies `-r`, so the dir is
121 /// copied nested under `local`); `AgentRpc` will fail on it. Use
122 /// [`Executor::pull_dir`] when you mean a directory.
123 async fn pull_file(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()>;
124
125 /// Pull the CONTENTS OF A DIRECTORY back to the caller (rsync from the host
126 /// into `local`). See [`Executor::pull_file`] for why this is separate.
127 async fn pull_dir(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()>;
128
129 /// Pull every file matching a shell GLOB into the directory `local_dir`.
130 ///
131 /// The artifact-collection shape: a recipe knows `…/bundle/msi/*.msi`, not
132 /// the exact filenames. Matching zero files is an error — a collect that
133 /// silently gathers nothing is how an empty release ships.
134 ///
135 /// `remote_glob` is a `str`, not a `Path`, because it is a *pattern*: the
136 /// wildcard must survive to whatever expands it, and `Path` invites callers
137 /// to `join`/normalize it. Expansion differs per transport — the ssh
138 /// transport lets the REMOTE shell expand, the local one expands in-process
139 /// — so implementations must not assume a shell is involved.
140 async fn pull_glob(&self, remote_glob: &str, local_dir: &Path, opts: &SyncOpts) -> Result<()>;
141
142 /// Push the contents of a local directory to the host (rsync into `remote`).
143 ///
144 /// Directory-only: both rsync transports append a trailing slash, and
145 /// `AgentRpc` refuses by design (bulk data onto a host goes over
146 /// SshExec/rsync or a `git pull` step, not the agent's exec surface).
147 async fn push_dir(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()>;
148
149 /// Subscribe to the host's observe stream; `None` if no observe capability
150 /// (or, in v1, if no resident observer is wired — E3).
151 fn observe(&self) -> Option<EventStream> {
152 None
153 }
154
155 /// Optional readiness check, run before dispatching real work to this host.
156 /// The default is a no-op: a `LocalExec`/`SshExec` host is ready if it is
157 /// reachable, and the first command surfaces any failure. `AgentRpc`
158 /// overrides this to hit `/health`, so an agent-host build fails fast with a
159 /// clear "is ops-agent running in the Aqua session?" message instead of
160 /// erroring opaquely on the first `/run` dispatch.
161 async fn preflight(&self) -> Result<()> {
162 Ok(())
163 }
164
165 /// The capability set this executor was granted (introspection / audit).
166 fn capabilities(&self) -> &CapabilitySet;
167
168 /// Convenience: does this executor's grant cover `kind`?
169 fn can_observe(&self, kind: &ObserveKind) -> bool {
170 self.capabilities().permits_observe(kind)
171 }
172 }
173
174 /// Spawn `cmd`, draining stdout+stderr concurrently into the single
175 /// `&mut dyn LogSink`, and return the exit status plus full captured bytes.
176 ///
177 /// Unlike [`crate::remote::RemoteHost::run_streaming`] (which shares the sink
178 /// across two spawned tasks via `Arc<Mutex<_>>`), this drains both pipes in one
179 /// task with `select!` so it can take a borrowed `&mut dyn` sink — exactly the
180 /// shape the [`Executor`] trait exposes.
181 ///
182 /// `sentinel` must be the one [`crate::remote::RemoteHost::command_for`] handed
183 /// back with `cmd`: `Some` for a remote host, whose reported status is not
184 /// trustworthy (see [`crate::remote::RcSentinel`]), `None` for local. `host`
185 /// only names the target in errors.
186 pub(crate) async fn run_command_into_sink(
187 mut cmd: Command,
188 sink: &mut dyn LogSink,
189 sentinel: Option<crate::remote::RcSentinel>,
190 host: &str,
191 ) -> Result<RunOutput> {
192 use std::process::Stdio;
193 use tokio::io::AsyncReadExt;
194
195 cmd.stdout(Stdio::piped());
196 cmd.stderr(Stdio::piped());
197 cmd.kill_on_drop(true);
198 let mut child = cmd
199 .spawn()
200 .map_err(|e| anyhow::anyhow!("spawning command: {e}"))?;
201
202 let mut out = child.stdout.take();
203 let mut err = child.stderr.take();
204 let mut stdout_buf = Vec::new();
205 let mut stderr_buf = Vec::new();
206 let mut ob = [0u8; 4096];
207 let mut eb = [0u8; 4096];
208 let mut out_done = out.is_none();
209 let mut err_done = err.is_none();
210 // Only stdout carries the sentinel; stderr streams untouched.
211 let mut filter = crate::remote::RcFilter::new(sentinel.clone());
212
213 while !(out_done && err_done) {
214 tokio::select! {
215 r = async { out.as_mut().unwrap().read(&mut ob).await }, if !out_done => {
216 match r {
217 Ok(0) | Err(_) => out_done = true,
218 Ok(n) => {
219 let chunk = filter.feed(&ob[..n]);
220 if !chunk.is_empty() {
221 crate::remote::push_bounded(&mut stdout_buf, &chunk, crate::remote::OUTPUT_TAIL_CAP);
222 sink.write_chunk(&chunk).await;
223 }
224 }
225 }
226 }
227 r = async { err.as_mut().unwrap().read(&mut eb).await }, if !err_done => {
228 match r {
229 Ok(0) | Err(_) => err_done = true,
230 Ok(n) => { crate::remote::push_bounded(&mut stderr_buf, &eb[..n], crate::remote::OUTPUT_TAIL_CAP); sink.write_chunk(&eb[..n]).await; }
231 }
232 }
233 }
234 }
235
236 let (rest, code) = filter.finish();
237 if !rest.is_empty() {
238 crate::remote::push_bounded(&mut stdout_buf, &rest, crate::remote::OUTPUT_TAIL_CAP);
239 sink.write_chunk(&rest).await;
240 }
241
242 let status = child
243 .wait()
244 .await
245 .map_err(|e| anyhow::anyhow!("waiting on child: {e}"))?;
246 let status = match sentinel {
247 Some(_) => crate::remote::resolve_remote_status(host, status, code)?,
248 None => status,
249 };
250 Ok(RunOutput {
251 status,
252 stdout: stdout_buf,
253 stderr: stderr_buf,
254 })
255 }
256