Skip to main content

max / makenotwork

bento: emit real metrics + bound captured output tail (Run 2 perf) metrics: the Prometheus recorder was installed and /metrics served, but nothing was ever measured. Add emit sites the runner now calls — bento_builds_total (counter), bento_target_runs_total{target,status} + bento_target_run_seconds (per-target outcome + duration), and bento_targets_in_flight (gauge off the live active map). An operator scraping /metrics now sees real build activity. perf: RunOutput captured the full stdout/stderr in RAM (a verbose multi-minute build holds tens of MB) only for callers to read a recent tail — the live sink and on-disk log are the byte-exact source. Bound the captured buffers to a 256 KiB rolling tail (executor, remote drain, agent rpc). Safe for Sando (its compile classification reads std::process output, not RunOutput). Note: CHRONIC-1 (busy-poll), the active-map leak, and task supervision — the runner.rs perf cold spots — were already closed in the Phase 3 JoinSet refactor (acfe960). The log/status event-bus split (one 256-slot broadcast carries both) is left at A-: the bus is best-effort by design (the disk log is authoritative, Lagged is recoverable), so a full two-channel split is the A+ form, not required for A. ops-exec 36 tests, bento-daemon 44, sando builds; clippy -D clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 01:29 UTC
Commit: 6ce0ebedc80c75764109301950ec0d87f68789af
Parent: acfe960
5 files changed, +78 insertions, -8 deletions
M bento/daemon/src/metrics.rs +24
@@ -1,10 +1,34 @@
1 1 use axum::{extract::State, response::IntoResponse};
2 + use metrics::{counter, gauge, histogram};
2 3 use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
3 4
4 5 pub fn init() -> PrometheusHandle {
5 6 PrometheusBuilder::new().install_recorder().expect("install prometheus recorder")
6 7 }
7 8
9 + // --- Emission ---------------------------------------------------------------
10 + // The recorder was installed and `/metrics` served, but nothing was ever
11 + // measured. These are the emit sites the runner calls so an operator scraping
12 + // `/metrics` sees real build activity, not an empty page.
13 +
14 + /// A build was accepted and fanned out.
15 + pub fn build_started() {
16 + counter!("bento_builds_total").increment(1);
17 + }
18 +
19 + /// A target run reached a terminal state. `status` is `ok` / `failed` /
20 + /// `aborted`; `secs` is its wall-clock duration.
21 + pub fn target_finished(target: &str, status: &str, secs: f64) {
22 + counter!("bento_target_runs_total", "target" => target.to_owned(), "status" => status.to_owned())
23 + .increment(1);
24 + histogram!("bento_target_run_seconds", "target" => target.to_owned()).record(secs);
25 + }
26 +
27 + /// Set the gauge of target tasks currently in flight (the live `active` count).
28 + pub fn set_in_flight(n: usize) {
29 + gauge!("bento_targets_in_flight").set(n as f64);
30 + }
31 +
8 32 pub async fn render(State(handle): State<PrometheusHandle>) -> impl IntoResponse {
9 33 handle.render()
10 34 }
M bento/daemon/src/runner.rs +16 -2
@@ -83,6 +83,7 @@ pub async fn start_build(
83 83 &state.events,
84 84 Event::BuildRequested { app: app.clone(), version: version.clone(), targets: targets.clone() },
85 85 );
86 + crate::metrics::build_started();
86 87
87 88 // Spawn every target into a JoinSet so finalize_build can await completion
88 89 // event-driven (no DB polling) and observe each task's outcome (panic vs
@@ -107,7 +108,9 @@ pub async fn start_build(
107 108 }
108 109
109 110 let abort = set.spawn(run_target(state.clone(), build_id, app, version, target));
110 - state.active.lock().await.insert(key, (build_id, abort));
111 + let mut active = state.active.lock().await;
112 + active.insert(key, (build_id, abort));
113 + crate::metrics::set_in_flight(active.len());
111 114 }
112 115
113 116 // Mark the build done once all target tasks settle. Spawned so /build
@@ -118,6 +121,7 @@ pub async fn start_build(
118 121
119 122 /// Run one target's recipe end to end, updating its `target_runs` row.
120 123 async fn run_target(state: AppState, build_id: i64, app: AppId, version: Version, target: Target) {
124 + let started = std::time::Instant::now();
121 125 let target_run_id: i64 = match sqlx::query_scalar(
122 126 "INSERT INTO target_runs (build_id, app, version, target, status, started_at)
123 127 VALUES (?, ?, ?, ?, 'running', ?) RETURNING id",
@@ -146,6 +150,7 @@ async fn run_target(state: AppState, build_id: i64, app: AppId, version: Version
146 150 Ok(s) => s,
147 151 Err(e) => {
148 152 fail_target(&state, target_run_id, &app, &version, target, Step::Checkout, &format!("{e:#}")).await;
153 + crate::metrics::target_finished(&target.to_string(), "failed", started.elapsed().as_secs_f64());
149 154 return;
150 155 }
151 156 };
@@ -160,6 +165,7 @@ async fn run_target(state: AppState, build_id: i64, app: AppId, version: Version
160 165 && let Err(e) = exec.preflight().await
161 166 {
162 167 fail_target(&state, target_run_id, &app, &version, target, Step::Checkout, &format!("{e:#}")).await;
168 + crate::metrics::target_finished(&target.to_string(), "failed", started.elapsed().as_secs_f64());
163 169 return;
164 170 }
165 171
@@ -229,13 +235,17 @@ async fn run_target(state: AppState, build_id: i64, app: AppId, version: Version
229 235 // finalize_build reconciles it, but log so the cause is visible.
230 236 tracing::error!(%app, %target, error = %e, "could not stamp target_run ok");
231 237 }
238 + crate::metrics::target_finished(&target.to_string(), "ok", started.elapsed().as_secs_f64());
232 239 events::emit(&state.events, Event::TargetOk { app, version, target, artifacts });
233 240 }
234 241 Ok(Err((step, msg))) => {
235 242 fail_target(&state, target_run_id, &app, &version, target, step, &msg).await;
243 + crate::metrics::target_finished(&target.to_string(), "failed", started.elapsed().as_secs_f64());
236 244 }
237 245 Err(join_err) => {
238 246 // Task was aborted (superseded) or panicked.
247 + let status = if join_err.is_cancelled() { "aborted" } else { "failed" };
248 + crate::metrics::target_finished(&target.to_string(), status, started.elapsed().as_secs_f64());
239 249 let msg = if join_err.is_cancelled() { "aborted (superseded)".to_string() } else { format!("recipe task panicked: {join_err}") };
240 250 fail_target(&state, target_run_id, &app, &version, target, Step::Build, &msg).await;
241 251 }
@@ -300,7 +310,11 @@ async fn finalize_build(state: AppState, build_id: i64, mut set: tokio::task::Jo
300 310
301 311 // Reap this build's latest-wins slots (only our own — a superseding build
302 312 // owns a different build_id and is left intact).
303 - state.active.lock().await.retain(|_, (bid, _)| *bid != build_id);
313 + {
314 + let mut active = state.active.lock().await;
315 + active.retain(|_, (bid, _)| *bid != build_id);
316 + crate::metrics::set_in_flight(active.len());
317 + }
304 318
305 319 let now = chrono::Utc::now().to_rfc3339();
306 320 // Any row still `running` is a panicked/aborted task that never stamped its
M shared/ops-exec/src/executor.rs +2 -2
@@ -122,13 +122,13 @@ pub(crate) async fn run_command_into_sink(
122 122 r = async { out.as_mut().unwrap().read(&mut ob).await }, if !out_done => {
123 123 match r {
124 124 Ok(0) | Err(_) => out_done = true,
125 - Ok(n) => { stdout_buf.extend_from_slice(&ob[..n]); sink.write_chunk(&ob[..n]).await; }
125 + Ok(n) => { crate::remote::push_bounded(&mut stdout_buf, &ob[..n], crate::remote::OUTPUT_TAIL_CAP); sink.write_chunk(&ob[..n]).await; }
126 126 }
127 127 }
128 128 r = async { err.as_mut().unwrap().read(&mut eb).await }, if !err_done => {
129 129 match r {
130 130 Ok(0) | Err(_) => err_done = true,
131 - Ok(n) => { stderr_buf.extend_from_slice(&eb[..n]); sink.write_chunk(&eb[..n]).await; }
131 + Ok(n) => { crate::remote::push_bounded(&mut stderr_buf, &eb[..n], crate::remote::OUTPUT_TAIL_CAP); sink.write_chunk(&eb[..n]).await; }
132 132 }
133 133 }
134 134 }
M shared/ops-exec/src/remote.rs +35 -3
@@ -56,8 +56,11 @@ pub struct RemoteHost {
56 56 ssh_target: String,
57 57 }
58 58
59 - /// Result of a streamed run: the exit status plus the full captured stdout and
60 - /// stderr byte streams (already forwarded to the sink as they arrived).
59 + /// Result of a streamed run: the exit status plus a bounded recent tail of the
60 + /// captured stdout and stderr (each already forwarded to the sink — and, for
61 + /// the engine, the on-disk log — byte-exact as it arrived). The returned buffer
62 + /// is only for callers that branch on output, so it is capped at
63 + /// [`OUTPUT_TAIL_CAP`] to avoid holding a multi-GB build log in RAM.
61 64 #[derive(Debug)]
62 65 pub struct RunOutput {
63 66 pub status: ExitStatus,
@@ -71,6 +74,19 @@ impl RunOutput {
71 74 }
72 75 }
73 76
77 + /// Cap for the in-memory capture returned in [`RunOutput`]. The live sink and
78 + /// the on-disk log carry the byte-exact stream; this buffer is a recent tail.
79 + pub(crate) const OUTPUT_TAIL_CAP: usize = 256 * 1024;
80 +
81 + /// Append `chunk`, then trim the front so `buf` retains at most its last `cap`
82 + /// bytes — a rolling tail that bounds memory for verbose, long-running commands.
83 + pub(crate) fn push_bounded(buf: &mut Vec<u8>, chunk: &[u8], cap: usize) {
84 + buf.extend_from_slice(chunk);
85 + if buf.len() > cap {
86 + buf.drain(..buf.len() - cap);
87 + }
88 + }
89 +
74 90 impl RemoteHost {
75 91 /// `"local"` (or empty) runs commands directly via `sh -c`; anything else
76 92 /// is an SSH target.
@@ -140,7 +156,7 @@ where
140 156 match s.read(&mut buf).await {
141 157 Ok(0) | Err(_) => break,
142 158 Ok(n) => {
143 - total.extend_from_slice(&buf[..n]);
159 + push_bounded(&mut total, &buf[..n], OUTPUT_TAIL_CAP);
144 160 sink.lock().await.write_chunk(&buf[..n]).await;
145 161 }
146 162 }
@@ -160,6 +176,22 @@ pub fn sh_quote(s: &str) -> String {
160 176 mod tests {
161 177 use super::*;
162 178
179 + #[test]
180 + fn push_bounded_keeps_only_the_last_cap_bytes() {
181 + let mut buf = Vec::new();
182 + push_bounded(&mut buf, b"hello", 8);
183 + push_bounded(&mut buf, b"world", 8); // 10 bytes -> trim front to last 8
184 + assert_eq!(buf, b"lloworld");
185 + // A single chunk larger than the cap is itself trimmed to the tail.
186 + let mut buf2 = Vec::new();
187 + push_bounded(&mut buf2, b"0123456789", 4);
188 + assert_eq!(buf2, b"6789");
189 + // Under the cap: untouched.
190 + let mut buf3 = Vec::new();
191 + push_bounded(&mut buf3, b"ok", 8);
192 + assert_eq!(buf3, b"ok");
193 + }
194 +
163 195 /// A simple sink that accumulates every chunk for assertions.
164 196 #[derive(Default)]
165 197 pub(crate) struct VecSink(pub Vec<u8>);
M shared/ops-exec/src/rpc.rs +1 -1
@@ -107,7 +107,7 @@ impl Executor for AgentRpc {
107 107 .with_context(|| format!("decoding agent frame: {}", String::from_utf8_lossy(line)))?;
108 108 match frame {
109 109 Frame::Chunk { text } => {
110 - captured.extend_from_slice(text.as_bytes());
110 + crate::remote::push_bounded(&mut captured, text.as_bytes(), crate::remote::OUTPUT_TAIL_CAP);
111 111 sink.write_chunk(text.as_bytes()).await;
112 112 }
113 113 Frame::Exit { code } => exit_code = Some(code),