Skip to main content

max / makenotwork

11.5 KB · 285 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 /// `--mkpath`: create the destination's missing parent directories rather
38 /// than failing. Off by default, because for a hand-written destination a
39 /// missing parent is a typo and minting the tree hides it. Turn it on when
40 /// the caller computes the destination and is authoritative for it — Bento's
41 /// per-`(app, version, target)` artifact directory, which by definition does
42 /// not exist before the build that fills it.
43 pub mkpath: bool,
44 /// `--exclude=<pattern>`, one per entry. Empty by default: a sync normally
45 /// carries the whole tree, and silently dropping files from a release is the
46 /// last thing this should do casually.
47 ///
48 /// The case it exists for is a directory that holds a bundle *and* a
49 /// document about that bundle. Evidence names the digest of the bytes it
50 /// vouches for, so evidence copied in among them would change the digest it
51 /// names — the transport has to be able to leave it behind.
52 pub exclude: Vec<String>,
53 }
54
55 impl Default for SyncOpts {
56 fn default() -> Self {
57 Self {
58 delete: false,
59 chmod: None,
60 compress: true,
61 partial: true,
62 mkpath: false,
63 exclude: Vec::new(),
64 }
65 }
66 }
67
68 impl SyncOpts {
69 /// The Sando release-dir mirror: prune stale assets and force exec bits the
70 /// way `deploy.rs` always has.
71 pub fn release_mirror() -> Self {
72 Self {
73 delete: true,
74 chmod: Some("Du=rwx,Dgo=rx,Fu=rw,Fgo=r,F+X".into()),
75 ..Self::default()
76 }
77 }
78
79 /// For payloads that are already compressed (`.gz`/`.dmg`/`.zip`): skip
80 /// `-z` so the transfer doesn't re-compress compressed bytes.
81 pub fn precompressed() -> Self {
82 Self {
83 compress: false,
84 ..Self::default()
85 }
86 }
87
88 /// Depositing an already-compressed release artifact at a computed,
89 /// versioned destination: no `-z`, and create the destination tree, since
90 /// the first build of a version is the thing that makes its directory exist.
91 pub fn archive_deposit() -> Self {
92 Self {
93 compress: false,
94 mkpath: true,
95 ..Self::default()
96 }
97 }
98 }
99
100 /// A read-only host observation. v1 can synthesize these from SSH-streamed
101 /// commands; a resident `ops-agent` is a drop-in upgrade (E3). The variants
102 /// mirror the executor spec.
103 #[derive(Clone, Debug, PartialEq)]
104 pub enum ObserveEvent {
105 ProcessExited {
106 unit: String,
107 code: i32,
108 },
109 ResourceSample {
110 cpu: f64,
111 rss: u64,
112 disk: u64,
113 },
114 JournalLine {
115 unit: String,
116 line: String,
117 },
118 HealthChanged {
119 check: String,
120 from: String,
121 to: String,
122 },
123 }
124
125 /// A live stream of [`ObserveEvent`]s from a host's observe plane.
126 pub type EventStream = tokio::sync::mpsc::Receiver<ObserveEvent>;
127
128 #[async_trait]
129 pub trait Executor: Send + Sync {
130 /// Run a typed step, streaming merged stdout/stderr into `sink`.
131 ///
132 /// Returns [`crate::CapabilityDenied`] (boxed into the error) if
133 /// `step.action` is outside this executor's grant — checked *before* the
134 /// command is dispatched.
135 async fn run_streaming(&self, step: &Step, sink: &mut dyn LogSink) -> Result<RunOutput>;
136
137 /// Pull ONE FILE a prior step produced back to the caller: `remote` names
138 /// the file on the host, `local` the destination path.
139 ///
140 /// Split from [`Executor::pull_dir`] because the transports genuinely
141 /// differ here, and conflating them is a silent foot-gun: the rsync
142 /// transports need a trailing slash for directory-contents semantics and
143 /// must NOT have one for a file, while [`crate::AgentRpc`] streams a single
144 /// file over HTTP and cannot do directories at all. One `pull` taking either
145 /// shape meant a config-level transport swap (`HostTransport::Agent` →
146 /// `Ssh`) could turn a working file pull into `rsync host:/x/App.dmg/`.
147 ///
148 /// Contract is file→file. Pointing this at a *directory* is not checked and
149 /// not an error on the rsync transports (`-a` implies `-r`, so the dir is
150 /// copied nested under `local`); `AgentRpc` will fail on it. Use
151 /// [`Executor::pull_dir`] when you mean a directory.
152 async fn pull_file(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()>;
153
154 /// Pull the CONTENTS OF A DIRECTORY back to the caller (rsync from the host
155 /// into `local`). See [`Executor::pull_file`] for why this is separate.
156 async fn pull_dir(&self, remote: &Path, local: &Path, opts: &SyncOpts) -> Result<()>;
157
158 /// Pull every file matching a shell GLOB into the directory `local_dir`.
159 ///
160 /// The artifact-collection shape: a recipe knows `…/bundle/msi/*.msi`, not
161 /// the exact filenames. Matching zero files is an error — a collect that
162 /// silently gathers nothing is how an empty release ships.
163 ///
164 /// `remote_glob` is a `str`, not a `Path`, because it is a *pattern*: the
165 /// wildcard must survive to whatever expands it, and `Path` invites callers
166 /// to `join`/normalize it. Expansion differs per transport — the ssh
167 /// transport lets the REMOTE shell expand, the local one expands in-process
168 /// — so implementations must not assume a shell is involved.
169 async fn pull_glob(&self, remote_glob: &str, local_dir: &Path, opts: &SyncOpts) -> Result<()>;
170
171 /// Push the contents of a local directory to the host (rsync into `remote`).
172 ///
173 /// Directory-only: both rsync transports append a trailing slash, and
174 /// `AgentRpc` refuses by design (bulk data onto a host goes over
175 /// SshExec/rsync or a `git pull` step, not the agent's exec surface).
176 async fn push_dir(&self, local: &Path, remote: &Path, opts: &SyncOpts) -> Result<()>;
177
178 /// Subscribe to the host's observe stream; `None` if no observe capability
179 /// (or, in v1, if no resident observer is wired — E3).
180 fn observe(&self) -> Option<EventStream> {
181 None
182 }
183
184 /// Optional readiness check, run before dispatching real work to this host.
185 /// The default is a no-op: a `LocalExec`/`SshExec` host is ready if it is
186 /// reachable, and the first command surfaces any failure. `AgentRpc`
187 /// overrides this to hit `/health`, so an agent-host build fails fast with a
188 /// clear "is ops-agent running in the Aqua session?" message instead of
189 /// erroring opaquely on the first `/run` dispatch.
190 async fn preflight(&self) -> Result<()> {
191 Ok(())
192 }
193
194 /// The capability set this executor was granted (introspection / audit).
195 fn capabilities(&self) -> &CapabilitySet;
196
197 /// Convenience: does this executor's grant cover `kind`?
198 fn can_observe(&self, kind: &ObserveKind) -> bool {
199 self.capabilities().permits_observe(kind)
200 }
201 }
202
203 /// Spawn `cmd`, draining stdout+stderr concurrently into the single
204 /// `&mut dyn LogSink`, and return the exit status plus full captured bytes.
205 ///
206 /// Unlike [`crate::remote::RemoteHost::run_streaming`] (which shares the sink
207 /// across two spawned tasks via `Arc<Mutex<_>>`), this drains both pipes in one
208 /// task with `select!` so it can take a borrowed `&mut dyn` sink — exactly the
209 /// shape the [`Executor`] trait exposes.
210 ///
211 /// `sentinel` must be the one [`crate::remote::RemoteHost::command_for`] handed
212 /// back with `cmd`: `Some` for a remote host, whose reported status is not
213 /// trustworthy (see [`crate::remote::RcSentinel`]), `None` for local. `host`
214 /// only names the target in errors.
215 pub(crate) async fn run_command_into_sink(
216 mut cmd: Command,
217 sink: &mut dyn LogSink,
218 sentinel: Option<crate::remote::RcSentinel>,
219 host: &str,
220 ) -> Result<RunOutput> {
221 use std::process::Stdio;
222 use tokio::io::AsyncReadExt;
223
224 cmd.stdout(Stdio::piped());
225 cmd.stderr(Stdio::piped());
226 cmd.kill_on_drop(true);
227 let mut child = cmd
228 .spawn()
229 .map_err(|e| anyhow::anyhow!("spawning command: {e}"))?;
230
231 let mut out = child.stdout.take();
232 let mut err = child.stderr.take();
233 let mut stdout_buf = Vec::new();
234 let mut stderr_buf = Vec::new();
235 let mut ob = [0u8; 4096];
236 let mut eb = [0u8; 4096];
237 let mut out_done = out.is_none();
238 let mut err_done = err.is_none();
239 // Only stdout carries the sentinel; stderr streams untouched.
240 let mut filter = crate::remote::RcFilter::new(sentinel.clone());
241
242 while !(out_done && err_done) {
243 tokio::select! {
244 r = async { out.as_mut().unwrap().read(&mut ob).await }, if !out_done => {
245 match r {
246 Ok(0) | Err(_) => out_done = true,
247 Ok(n) => {
248 let chunk = filter.feed(&ob[..n]);
249 if !chunk.is_empty() {
250 crate::remote::push_bounded(&mut stdout_buf, &chunk, crate::remote::OUTPUT_TAIL_CAP);
251 sink.write_chunk(&chunk).await;
252 }
253 }
254 }
255 }
256 r = async { err.as_mut().unwrap().read(&mut eb).await }, if !err_done => {
257 match r {
258 Ok(0) | Err(_) => err_done = true,
259 Ok(n) => { crate::remote::push_bounded(&mut stderr_buf, &eb[..n], crate::remote::OUTPUT_TAIL_CAP); sink.write_chunk(&eb[..n]).await; }
260 }
261 }
262 }
263 }
264
265 let (rest, code) = filter.finish();
266 if !rest.is_empty() {
267 crate::remote::push_bounded(&mut stdout_buf, &rest, crate::remote::OUTPUT_TAIL_CAP);
268 sink.write_chunk(&rest).await;
269 }
270
271 let status = child
272 .wait()
273 .await
274 .map_err(|e| anyhow::anyhow!("waiting on child: {e}"))?;
275 let status = match sentinel {
276 Some(_) => crate::remote::resolve_remote_status(host, status, code)?,
277 None => status,
278 };
279 Ok(RunOutput {
280 status,
281 stdout: stdout_buf,
282 stderr: stderr_buf,
283 })
284 }
285