diff --git a/Cargo.lock b/Cargo.lock index fc7e89fc..4855044f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2021,7 +2021,10 @@ name = "preloop-observability" version = "0.1.0" dependencies = [ "anyhow", + "chrono", "parking_lot", + "serde", + "serde_json", "tokio", "tracing", "tracing-subscriber", @@ -2038,6 +2041,7 @@ dependencies = [ "futures", "preloop-gha-parser", "preloop-gha-protocol", + "preloop-observability", "preloop-runner-server", "preloop-vm", "reqwest", diff --git a/crates/preloop-cli/src/main.rs b/crates/preloop-cli/src/main.rs index d5b0816b..42339662 100644 --- a/crates/preloop-cli/src/main.rs +++ b/crates/preloop-cli/src/main.rs @@ -478,12 +478,8 @@ enum Command { /// Show the expanded job DAG without executing. Plan(PlanArgs), - /// Show active and recent runs, or the live status of one run (prints a - /// single machine-readable status word for scripting). - Status { - #[arg(value_name = "RUN_ID")] - run_id: Option, - }, + /// Show operational status, queue health, and recent runs. + Status(StatusArgs), Logs(LogsArgs), @@ -708,6 +704,17 @@ struct PlanArgs { json: bool, } +#[derive(Debug, Parser)] +struct StatusArgs { + /// Print the raw status JSON (no prose) for jq/scripting. + #[arg(long)] + json: bool, + + /// Number of recent runs to show in the table. + #[arg(long, default_value = "20")] + limit: usize, +} + #[derive(Debug, Parser)] struct LogsArgs { /// Run ID. Defaults to the most recent run. @@ -736,7 +743,7 @@ struct ShellArgs { #[tokio::main] async fn main() -> anyhow::Result<()> { - // Unified observability init (Step 2): one handle for the process, shared + // Unified observability init: one handle for the process, shared // with `AppState` and `RunnerPoolConfig` later. `PRELOOP_LOG_FORMAT` now // controls pretty/json/auto (auto = pretty on TTY, JSON when piped), and // `RUST_LOG` defaults to `info` (the old `fmt::init()` default of ERROR hid @@ -746,11 +753,6 @@ async fn main() -> anyhow::Result<()> { let (observability, observability_runtime) = preloop_observability::Observability::from_config(obs_config); preloop_observability::ObservabilityRuntime::install_fmt_subscriber(observability.config()); - // Keep the handle alive; the pool/server will clone it later. Suppress - // unused warning until the wiring lands in Step 3. - let _observability = observability; - let _observability_runtime = observability_runtime; - let cli = Cli::parse(); // One config path for the whole process. `setup`/`doctor`/`secret` return // before `cmd_engine` runs, so pinning this only inside the engine let a @@ -765,51 +767,55 @@ async fn main() -> anyhow::Result<()> { } // Both run the daemon in this process, so neither may bootstrap another // one underneath itself. - match cli.command { + let result = match cli.command { Command::Version => { println!("preloop {}", env!("CARGO_PKG_VERSION")); - return Ok(()); + Ok(()) } - Command::Serve(args) => return cmd_engine(args).await, - Command::Engine => return cmd_engine(ServeArgs::default()).await, - Command::BuildGolden(args) => return cmd_build_golden(args).await, - Command::Update(args) => return update::run(args).await, + Command::Serve(args) => cmd_engine(args, observability.clone()).await, + Command::Engine => cmd_engine(ServeArgs::default(), observability.clone()).await, + Command::BuildGolden(args) => cmd_build_golden(args).await, + Command::Update(args) => update::run(args).await, // Local configuration commands must not spawn the engine. - Command::Setup(args) => return github_setup::cmd_setup(args).await, - Command::Doctor(args) => return github_setup::cmd_doctor(args).await, - Command::Secret(args) => return github_setup::cmd_secret(args).await, - Command::Server(args) => return server_install::run(args), + Command::Setup(args) => github_setup::cmd_setup(args).await, + Command::Doctor(args) => github_setup::cmd_doctor(args).await, + Command::Secret(args) => github_setup::cmd_secret(args).await, + Command::Server(args) => server_install::run(args), // Planning parses local workflow files only; do not bootstrap the // control-plane engine for a command that never contacts it. - Command::Plan(args) => return cmd_plan(args).await, - _ => {} - } - ensure_engine_running().await?; - - match cli.command { - Command::Run(args) => cmd_run(args).await, - Command::Plan(_) => unreachable!("plan is handled before engine bootstrap"), - Command::Status { run_id } => cmd_status(run_id).await, - Command::Logs(args) => cmd_logs(args).await, - Command::Cancel(args) => cmd_cancel(args).await, - Command::Shell(args) => cmd_shell(args).await, - Command::Debug(args) => { - debug_session::run(args, build_client(), server_url(), api_token()).await - } - Command::Dap(args) => dap_client::run(args, server_url(), api_token()).await, - Command::Push(args) => cmd_push(args).await, - Command::Update(_) - | Command::Serve(_) - | Command::Engine - | Command::BuildGolden(_) - | Command::Version - | Command::Setup(_) - | Command::Doctor(_) - | Command::Secret(_) - | Command::Server(_) => { - unreachable!("daemon commands handled before client startup") + Command::Plan(args) => cmd_plan(args).await, + _ => { + ensure_engine_running().await?; + match cli.command { + Command::Run(args) => cmd_run(args).await, + Command::Plan(_) => unreachable!("plan is handled before engine bootstrap"), + Command::Status(args) => cmd_status(args).await, + Command::Logs(args) => cmd_logs(args).await, + Command::Cancel(args) => cmd_cancel(args).await, + Command::Shell(args) => cmd_shell(args).await, + Command::Debug(args) => { + debug_session::run(args, build_client(), server_url(), api_token()).await + } + Command::Dap(args) => dap_client::run(args, server_url(), api_token()).await, + Command::Push(args) => cmd_push(args).await, + Command::Update(_) + | Command::Serve(_) + | Command::Engine + | Command::BuildGolden(_) + | Command::Version + | Command::Setup(_) + | Command::Doctor(_) + | Command::Secret(_) + | Command::Server(_) => { + unreachable!("daemon commands handled before client startup") + } + } } - } + }; + // Bounded 2s flush of buffered telemetry on every exit path; a clean + // shutdown must not drop the last flush window's records. + observability_runtime.shutdown().await; + result } fn systemd_socket_activation_requested() -> bool { @@ -903,6 +909,7 @@ async fn cmd_build_golden(args: BuildGoldenArgs) -> anyhow::Result<()> { next_job_runs_on: None, pending_registrations: None, preparing_signal: None, + pool_status: None, }; let payload = artifact_payload(&output, &config.base_image); RunnerPool::new(std::sync::Arc::new(SmolVmProvider::default()), config)? @@ -1261,7 +1268,10 @@ fn resolve_github_auth(args: &ServeArgs, state_dir: &std::path::Path) -> anyhow: Ok(()) } -async fn cmd_engine(args: ServeArgs) -> anyhow::Result<()> { +async fn cmd_engine( + args: ServeArgs, + observability: preloop_observability::Observability, +) -> anyhow::Result<()> { let home = preloop_home(); let state_dir = home.join("state"); let socket = home.join("preloop.sock"); @@ -1328,6 +1338,12 @@ async fn cmd_engine(args: ServeArgs) -> anyhow::Result<()> { std::time::SystemTime, >::new())); let pool_enabled = env_flag("PRELOOP_RUNNER_POOL_ENABLED", DEFAULT_RUNNER_POOL_ENABLED); + // One shared handle: the server and the pool both observe the same + // preparing/queue state, and the server exports via the process handle + // instead of a fresh no-op one. + let pool_status = std::sync::Arc::new(preloop_observability::status::PoolStatus::new( + preloop_observability::status::PoolSnapshot::default(), + )); let pool_config = local_runner_pool_config( &home, runner_url.clone(), @@ -1338,6 +1354,7 @@ async fn cmd_engine(args: ServeArgs) -> anyhow::Result<()> { pool_enabled, pool_preparing.clone(), pending_registrations.clone(), + pool_status.clone(), ); let pool_available = match &pool_config { Ok(_) => true, @@ -1366,6 +1383,8 @@ async fn cmd_engine(args: ServeArgs) -> anyhow::Result<()> { next_job_runs_on: Some(next_job_runs_on.clone()), pool_preparing: Some(pool_preparing.clone()), pending_registrations: pool_available.then_some(pending_registrations), + pool_status: Some(pool_status.clone()), + observability: Some(observability), require_job_assignments: env_flag("PRELOOP_REQUIRE_JOB_ASSIGNMENTS", false), state_dir, store_url: args.store.clone(), @@ -1459,26 +1478,96 @@ async fn engine_shutdown_signal() { } async fn wait_for_engine_socket(socket: &std::path::Path) -> anyhow::Result<()> { + // readyz probe: http://localhost/readyz (500ms timeout, 30s window) #[cfg(unix)] let client = reqwest::Client::builder().unix_socket(socket).build()?; #[cfg(not(unix))] let client = reqwest::Client::new(); let start = std::time::Instant::now(); + let mut last_reason: Option = None; while start.elapsed() < Duration::from_secs(30) { - if client - .get("http://localhost/healthz") + match client + .get("http://localhost/readyz") .timeout(Duration::from_millis(500)) .send() .await - .is_ok() { - return Ok(()); + Ok(resp) => { + if resp.status().is_success() { + return Ok(()); + } + // Non-2xx readyz: capture reason for timeout reporting. + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + let reason = extract_readyz_reason(&body) + .unwrap_or_else(|| format!("{status}: {}", truncate_reason(&body))); + last_reason = Some(reason); + } + Err(err) => { + last_reason = Some(err.to_string()); + } } tokio::time::sleep(Duration::from_millis(50)).await; } + if let Some(reason) = last_reason { + anyhow::bail!("local control plane did not become ready within 30 seconds: last readyz reason: {reason}") + } anyhow::bail!("local control plane did not become ready within 30 seconds") } +fn extract_readyz_reason(body: &str) -> Option { + if body.trim().is_empty() { + return None; + } + if let Ok(v) = serde_json::from_str::(body) { + for key in ["reason", "code", "message", "error", "status"] { + if let Some(s) = v.get(key).and_then(|x| x.as_str()) { + if !s.trim().is_empty() { + return Some(s.to_owned()); + } + } + } + // Nested { "ready": { "reason": ... } } or similar + if let Some(obj) = v.as_object() { + for (_, val) in obj { + if let Some(s) = val.as_str() { + if !s.trim().is_empty() && s.len() < 200 { + return Some(s.to_owned()); + } + } + if let Some(inner) = val.as_object() { + for k in ["reason", "code"] { + if let Some(s) = inner.get(k).and_then(|x| x.as_str()) { + return Some(s.to_owned()); + } + } + } + } + } + // Return truncated JSON if no specific field + return Some(truncate_reason(body)); + } + Some(truncate_reason(body)) +} + +fn truncate_reason(s: &str) -> String { + let t = s.trim(); + if t.len() > 300 { + // Byte slicing a `&str` panics when the cut lands inside a multi-byte + // character; the body is an arbitrary `/readyz` response (a localized + // proxy error page, for example), so cut on a character boundary. + let cut = t + .char_indices() + .map(|(index, _)| index) + .take_while(|index| *index <= 300) + .last() + .unwrap_or(0); + format!("{}…", &t[..cut]) + } else { + t.to_owned() + } +} + // Configuration assembly, not a public API: the parameter list mirrors the // inputs the pool genuinely needs. #[allow(clippy::too_many_arguments)] @@ -1494,6 +1583,7 @@ fn local_runner_pool_config( pending_registrations: std::sync::Arc< std::sync::RwLock>, >, + pool_status: std::sync::Arc, ) -> anyhow::Result { let control_bridge = home.join("control-bridge"); std::fs::create_dir_all(&control_bridge)?; @@ -1653,6 +1743,7 @@ fn local_runner_pool_config( next_job_runs_on: (!custom_base).then_some(next_job_runs_on), pending_registrations: Some(pending_registrations), preparing_signal: Some(preparing_signal), + pool_status: Some(pool_status), }) } @@ -2801,63 +2892,423 @@ fn plan_json(plan: &preloop_gha_protocol::JobPlan) -> serde_json::Value { }) } -async fn cmd_status(run_id: Option) -> anyhow::Result<()> { +async fn cmd_status(args: StatusArgs) -> anyhow::Result<()> { let client = build_client(); let url = server_url(); - if let Some(run_id) = run_id { - // Single-run mode: one machine-readable status word - // (success/failure/cancelled/skipped/in_progress/queued/pending) for - // scripts like the pre-push hook. Connection failures carry the - // engine-unreachable marker so the hook can fail open. - let mut request = client.get(format!("{url}/api/v1/runs/{run_id}")); - if let Some(token) = api_token() { - request = request.bearer_auth(token); - } - let response = request - .send() - .await - .with_context(engine_unreachable_context)?; - if !response.status().is_success() { - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - anyhow::bail!("server returned {status}: {body}"); + // Native bearer required for /api/v1/status (same as other native calls) + let mut status_req = client.get(format!("{url}/api/v1/status")); + if let Some(token) = api_token() { + status_req = status_req.bearer_auth(token); + } + let status_resp = status_req + .send() + .await + .with_context(engine_unreachable_context)?; + if !status_resp.status().is_success() { + let status = status_resp.status(); + let body = status_resp.text().await.unwrap_or_default(); + anyhow::bail!("server returned {status}: {body}"); + } + let status_text = status_resp.text().await?; + if args.json { + // Byte-for-byte, no prose: so jq works + print!("{}", status_text); + if !status_text.ends_with('\n') { + println!(); } - let run: serde_json::Value = response.json().await?; - println!( - "{}", - run.get("status") - .and_then(serde_json::Value::as_str) - .unwrap_or("unknown") - ); return Ok(()); } - let mut request = client.get(format!("{url}/api/v1/runs?limit=20")); + // Parse into the typed DTO first: a server-side schema change then fails + // loudly here instead of silently rendering every section as zeros and + // dashes. `--json` above stays raw for byte-identical `jq` output. + let _snapshot: preloop_observability::status::OperationalSnapshot = + serde_json::from_str(&status_text).context("parse status json")?; + let status: serde_json::Value = + serde_json::from_str(&status_text).context("parse status json")?; + + // Fetch recent runs for table (preserve existing behavior) + let mut runs_req = client.get(format!("{url}/api/v1/runs?limit={}", args.limit)); if let Some(token) = api_token() { - request = request.bearer_auth(token); + runs_req = runs_req.bearer_auth(token); + } + let runs: Vec = match runs_req.send().await { + Ok(r) if r.status().is_success() => r.json().await.unwrap_or_default(), + Ok(r) => { + let st = r.status(); + let body = r.text().await.unwrap_or_default(); + eprintln!("[warn] runs table unavailable: {st}: {body}"); + Vec::new() + } + Err(e) => { + eprintln!("[warn] runs table unavailable: {e}"); + Vec::new() + } + }; + + // --- Human rendering: 10 sections in order --- + render_status_human(&status, &runs, args.limit); + Ok(()) +} + +fn render_status_human(status: &serde_json::Value, runs: &[serde_json::Value], limit: usize) { + // Helpers to extract safely + let get_str = |v: &serde_json::Value, k: &str| -> Option { + v.get(k).and_then(|x| x.as_str()).map(|s| s.to_owned()) + }; + let get_f64 = + |v: &serde_json::Value, k: &str| -> Option { v.get(k).and_then(|x| x.as_f64()) }; + let get_u64 = + |v: &serde_json::Value, k: &str| -> Option { v.get(k).and_then(|x| x.as_u64()) }; + let get_bool = + |v: &serde_json::Value, k: &str| -> Option { v.get(k).and_then(|x| x.as_bool()) }; + + // 1. service + snapshot age + println!("== service =="); + let service = status.get("service").unwrap_or(&serde_json::Value::Null); + let version = + get_str(service, "version").unwrap_or_else(|| env!("CARGO_PKG_VERSION").to_owned()); + let instance = get_str(service, "instance_id") + .or_else(|| get_str(status, "instance_id")) + .unwrap_or_else(|| "-".to_owned()); + let uptime = get_f64(service, "uptime_seconds") + .or_else(|| get_f64(status, "uptime_seconds")) + .unwrap_or(0.0); + let shutdown = get_bool(service, "shutdown_requested") + .or_else(|| get_bool(status, "shutdown_requested")) + .unwrap_or(false); + let snapshot_age = get_f64(status, "snapshot_age_seconds").unwrap_or(0.0); + let observed_at = get_str(status, "observed_at").unwrap_or_else(|| "-".to_owned()); + let overall = get_str(status, "overall").unwrap_or_else(|| "unknown".to_owned()); + println!( + " version: {version} instance: {instance} uptime: {uptime:.0}s overall: {overall}" + ); + println!( + " observed_at: {observed_at} snapshot_age: {snapshot_age:.1}s shutdown: {shutdown}" + ); + if status.get("schema_version").is_some() { + println!(" schema_version: {}", status["schema_version"]); + } + + // 2. queue (ready/blocked + oldest) + println!("\n== queue =="); + let jobs = status.get("jobs").unwrap_or(&serde_json::Value::Null); + let ready = get_u64(jobs, "ready").unwrap_or(0); + let dep_blocked = get_u64(jobs, "dependency_blocked").unwrap_or(0); + let conc_blocked = get_u64(jobs, "concurrency_blocked").unwrap_or(0); + let pending_exp = get_u64(jobs, "pending_expansion").unwrap_or(0); + let expanding = get_u64(jobs, "expanding").unwrap_or(0); + let claimable = get_u64(jobs, "claimable").unwrap_or(0); + let unclaimable = get_u64(jobs, "unclaimable").unwrap_or(0); + let oldest = get_f64(jobs, "oldest_ready_seconds"); + println!(" ready: {ready} claimable: {claimable} unclaimable: {unclaimable} dependency_blocked: {dep_blocked} concurrency_blocked: {conc_blocked} pending_expansion: {pending_exp} expanding: {expanding}"); + match oldest { + Some(v) => println!(" oldest_ready: {v:.1}s"), + None => println!(" oldest_ready: -"), + } + // Also show runs queued/in_progress if present + if let Some(runs_obj) = status.get("runs") { + let q = get_u64(runs_obj, "queued").unwrap_or(0); + let ip = get_u64(runs_obj, "in_progress").unwrap_or(0); + let completed = get_u64(runs_obj, "completed").unwrap_or(0); + println!(" runs queued: {q} in_progress: {ip} completed: {completed}"); + } + + // 3. concurrency + scheduler + println!("\n== concurrency & scheduler =="); + let conc = status + .get("concurrency") + .unwrap_or(&serde_json::Value::Null); + let groups_active = get_u64(conc, "groups_active").unwrap_or(0); + let groups_contended = get_u64(conc, "groups_contended").unwrap_or(0); + let pending_holders = get_u64(conc, "pending_holders").unwrap_or(0); + let deepest = get_u64(conc, "deepest_group_pending").unwrap_or(0); + let qmax = get_u64(conc, "queue_max_pending").unwrap_or(0); + let overflow = get_u64(conc, "overflow_cancellations").unwrap_or(0); + println!(" groups active: {groups_active} contended: {groups_contended} pending_holders: {pending_holders} deepest_pending: {deepest} queue_max: {qmax} overflow_cancellations: {overflow}"); + let sched = status.get("scheduler").unwrap_or(&serde_json::Value::Null); + let enabled = get_bool(sched, "enabled").unwrap_or(false); + let schedules = get_u64(sched, "schedules").unwrap_or(0); + let last_scan = get_str(sched, "last_scan_at").unwrap_or_else(|| "-".to_owned()); + let next_fire = get_str(sched, "next_fire_at").unwrap_or_else(|| "-".to_owned()); + let fired = get_u64(sched, "fired").unwrap_or(0); + let skipped = get_u64(sched, "skipped_overlapping").unwrap_or(0); + let late = get_u64(sched, "late_fires").unwrap_or(0); + let max_delay = get_f64(sched, "max_fire_delay_seconds"); + println!(" scheduler enabled: {enabled} schedules: {schedules} fired: {fired} skipped_overlapping: {skipped} late_fires: {late}"); + println!( + " last_scan: {last_scan} next_fire: {next_fire} max_delay: {}", + max_delay + .map(|v| format!("{v:.1}s")) + .unwrap_or_else(|| "-".to_owned()) + ); + + // 4. pool + runners + println!("\n== pool & runners =="); + let pool = status.get("pool").unwrap_or(&serde_json::Value::Null); + let mode = get_str(pool, "mode").unwrap_or_else(|| "-".to_owned()); + let desired = get_u64(pool, "desired").unwrap_or(0); + let preparing = pool + .get("preparing") + .and_then(|x| x.as_bool()) + .unwrap_or(false); + let building = get_u64(pool, "building").unwrap_or(0); + let provisioning = get_u64(pool, "provisioning").unwrap_or(0); + let pool_idle = get_u64(pool, "idle").unwrap_or(0); + let pool_busy = get_u64(pool, "busy").unwrap_or(0); + let paused = get_u64(pool, "paused").unwrap_or(0); + let failures = get_u64(pool, "consecutive_provision_failures").unwrap_or(0); + println!(" pool mode: {mode} desired: {desired} idle: {pool_idle} busy: {pool_busy} building: {building} provisioning: {provisioning} paused: {paused} preparing: {preparing} provision_failures: {failures}"); + let runners = status.get("runners").unwrap_or(&serde_json::Value::Null); + let reg = get_u64(runners, "registered").unwrap_or(0); + let sessions = get_u64(runners, "sessions").unwrap_or(0); + let idle = get_u64(runners, "idle").unwrap_or(0); + let busy = get_u64(runners, "busy").unwrap_or(0); + let stale = get_u64(runners, "stale").unwrap_or(0); + let max_poll = get_f64(runners, "max_poll_age_seconds"); + let max_lease = get_f64(runners, "max_lease_age_seconds"); + println!(" runners registered: {reg} sessions: {sessions} idle: {idle} busy: {busy} stale: {stale}"); + println!( + " max_poll_age: {} max_lease_age: {}", + max_poll + .map(|v| format!("{v:.1}s")) + .unwrap_or_else(|| "-".to_owned()), + max_lease + .map(|v| format!("{v:.1}s")) + .unwrap_or_else(|| "-".to_owned()) + ); + + // 5. VM fleet stub + println!("\n== vm fleet =="); + let vms = status + .get("vms") + .unwrap_or(status.get("vm").unwrap_or(&serde_json::Value::Null)); + if vms.is_null() || vms.as_object().map(|m| m.is_empty()).unwrap_or(false) { + println!(" source: unavailable (host sampler not yet reporting)"); + println!(" capabilities: cpu=false memory=false sparse_disk=false"); + println!(" host_usage: -"); + } else { + let source = get_str(vms, "source").unwrap_or_else(|| "unknown".to_owned()); + let sample_age = get_f64(vms, "sample_age_seconds") + .map(|v| format!("{v:.1}s")) + .unwrap_or_else(|| "-".to_owned()); + println!(" source: {source} sample_age: {sample_age}"); + if let Some(caps) = vms.get("capabilities") { + let cap_str = caps + .as_object() + .map(|m| { + m.iter() + .map(|(k, v)| format!("{k}={}", v.as_bool().unwrap_or(false))) + .collect::>() + .join(" ") + }) + .unwrap_or_else(|| "-".to_owned()); + println!(" capabilities: {cap_str}"); + } + if let Some(cnt) = vms.get("count") { + let runner = get_u64(cnt, "runner").unwrap_or(0); + let golden = get_u64(cnt, "golden").unwrap_or(0); + let unavailable = get_u64(cnt, "unavailable").unwrap_or(0); + println!(" count runner: {runner} golden: {golden} unavailable: {unavailable}"); + } + if let Some(conf) = vms.get("configured") { + let vcpus = get_u64(conf, "vcpus") + .map(|v| v.to_string()) + .unwrap_or_else(|| "-".to_owned()); + let mem = get_u64(conf, "memory_bytes") + .map(|v| format!("{v}")) + .unwrap_or_else(|| "-".to_owned()); + let storage = get_u64(conf, "storage_bytes") + .map(|v| format!("{v}")) + .unwrap_or_else(|| "-".to_owned()); + println!(" configured vcpus: {vcpus} memory: {mem} storage: {storage}"); + } + if let Some(usage) = vms.get("host_usage") { + let cores = get_f64(usage, "cpu_cores") + .map(|v| format!("{v:.1}")) + .unwrap_or_else(|| "-".to_owned()); + let mem = get_u64(usage, "memory_bytes") + .map(|v| format!("{v}")) + .unwrap_or_else(|| "-".to_owned()); + println!(" host_usage cpu_cores: {cores} memory: {mem}"); + } + if let Some(top) = vms.get("top_consumers").and_then(|x| x.as_array()) { + if !top.is_empty() { + println!(" top_consumers ({}):", top.len().min(5)); + for c in top.iter().take(5) { + let name = c + .get("machine_name") + .and_then(|x| x.as_str()) + .unwrap_or("?"); + let role = c.get("role").and_then(|x| x.as_str()).unwrap_or("?"); + let activity = c.get("activity").and_then(|x| x.as_str()).unwrap_or("?"); + println!(" - {name} ({role}/{activity})"); + } + } else { + println!(" top_consumers: -"); + } + } } - let response = request.send().await?; - if !response.status().is_success() { - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - anyhow::bail!("server returned {status}: {body}"); + + // 6. store/storage/GitHub/debug/telemetry + println!("\n== store / storage / github / debug / telemetry =="); + let store = status.get("store").unwrap_or(&serde_json::Value::Null); + let backend = get_str(store, "backend").unwrap_or_else(|| "-".to_owned()); + let cfail = get_u64(store, "consecutive_failures").unwrap_or(0); + println!(" store backend: {backend} consecutive_failures: {cfail}"); + let storage = status.get("storage").unwrap_or(&serde_json::Value::Null); + if !storage.is_null() { + let free = get_u64(storage, "state_fs_free_bytes") + .map(|v| format!("{v}")) + .unwrap_or_else(|| "-".to_owned()); + let ratio = get_f64(storage, "state_fs_free_ratio") + .map(|v| format!("{v:.2}")) + .unwrap_or_else(|| "-".to_owned()); + println!(" storage free: {free} ratio: {ratio}"); + if let Some(comps) = storage.get("components").and_then(|x| x.as_array()) { + for c in comps { + let store_name = c.get("store").and_then(|x| x.as_str()).unwrap_or("?"); + let bytes = c.get("bytes").and_then(|x| x.as_u64()).unwrap_or(0); + println!(" {store_name}: {bytes} bytes"); + } + } + } else { + println!(" storage: -"); + } + let github = status.get("github").unwrap_or(&serde_json::Value::Null); + if !github.is_null() { + let configured = get_bool(github, "configured").unwrap_or(false); + let pending = get_u64(github, "pending_check_updates").unwrap_or(0); + println!(" github configured: {configured} pending_check_updates: {pending}"); + if let Some(rl) = github.get("rate_limit") { + let remaining = get_u64(rl, "remaining").unwrap_or(0); + let limit_rl = get_u64(rl, "limit").unwrap_or(0); + println!(" github rate_limit: {remaining}/{limit_rl} remaining"); + } + if let Some(exp) = github + .get("installation_token_expires_in_seconds") + .and_then(|x| x.as_u64()) + { + println!(" github token expires_in: {exp}s"); + } + } else { + println!(" github: -"); + } + let debug = status.get("debug").unwrap_or(&serde_json::Value::Null); + if !debug.is_null() { + let active = get_u64(debug, "active_sessions").unwrap_or(0); + let oldest = debug + .get("oldest_session_seconds") + .and_then(|x| x.as_f64()) + .map(|v| format!("{v:.0}s")) + .unwrap_or_else(|| "-".to_owned()); + println!(" debug active_sessions: {active} oldest: {oldest}"); + } + let tele = status.get("telemetry").unwrap_or(&serde_json::Value::Null); + if !tele.is_null() { + let enabled = get_bool(tele, "otlp_enabled").unwrap_or(false); + let dropped = get_u64(tele, "dropped_records").unwrap_or(0); + println!(" telemetry otlp_enabled: {enabled} dropped_records: {dropped}"); + } + + // 7. non-zero limits + println!("\n== limits (non-zero) =="); + let limits = status.get("limits").and_then(|x| x.as_array()); + let mut any_limit = false; + if let Some(arr) = limits { + for l in arr { + let dropped = l.get("dropped").and_then(|x| x.as_u64()).unwrap_or(0); + let rejected = l.get("rejected").and_then(|x| x.as_u64()).unwrap_or(0); + if dropped > 0 || rejected > 0 { + any_limit = true; + let name = l.get("limit").and_then(|x| x.as_str()).unwrap_or("?"); + let value = l + .get("value") + .and_then(|x| x.as_u64()) + .map(|v| v.to_string()) + .unwrap_or_else(|| "-".to_owned()); + println!(" {name}: value={value} dropped={dropped} rejected={rejected}"); + } + } } - let runs: Vec = response.json().await?; + if !any_limit { + println!(" (no limits with drops/rejects)"); + } + + // 8. stale tasks + println!("\n== tasks (stale/exited) =="); + let tasks = status.get("tasks").and_then(|x| x.as_array()); + let mut any_task = false; + if let Some(arr) = tasks { + for t in arr { + let state = t.get("state").and_then(|x| x.as_str()).unwrap_or("running"); + if state == "stale" || state == "exited" { + any_task = true; + let name = t.get("name").and_then(|x| x.as_str()).unwrap_or("?"); + let critical = t.get("critical").and_then(|x| x.as_bool()).unwrap_or(false); + let age = t + .get("heartbeat_age_seconds") + .and_then(|x| x.as_f64()) + .map(|v| format!("{v:.1}s")) + .unwrap_or_else(|| "-".to_owned()); + println!(" {name}: state={state} critical={critical} age={age}"); + } + } + } + if !any_task { + println!(" (all tasks healthy)"); + } + + // 9. conditions with one-line actions (≤5 exemplars) + println!("\n== conditions =="); + let conditions = status.get("conditions").and_then(|x| x.as_array()); + if let Some(arr) = conditions { + if arr.is_empty() { + println!(" (no conditions)"); + } else { + for c in arr { + let code = c.get("code").and_then(|x| x.as_str()).unwrap_or("?"); + let severity = c.get("severity").and_then(|x| x.as_str()).unwrap_or("info"); + let msg = c.get("message").and_then(|x| x.as_str()).unwrap_or(""); + let action = condition_action(code); + println!(" [{severity}] {code}: {msg} -> {action}"); + if let Some(exs) = c.get("exemplars").and_then(|x| x.as_array()) { + for ex in exs.iter().take(5) { + let ex_str = match ex { + serde_json::Value::String(s) => s.clone(), + other => other.to_string(), + }; + println!(" - {ex_str}"); + } + if exs.len() > 5 { + println!(" ... and {} more", exs.len() - 5); + } + } else if let Some(exs) = c.get("exemplar").and_then(|x| x.as_str()) { + println!(" - {exs}"); + } + } + } + } else { + println!(" (no conditions)"); + } + + // 10. recent runs table + println!("\n== recent runs (limit={}) ==", limit); if runs.is_empty() { println!("No runs found."); - return Ok(()); + return; } println!( "{:<38} {:<6} {:<12} {:<12} {:<10} WORKFLOW", "RUN ID", "#", "STATUS", "EVENT", "PUSH" ); println!("{}", "-".repeat(104)); - for run in &runs { + for run in runs { let run_id = run["run_id"].as_str().unwrap_or("?"); let run_number = run .get("run_number") .and_then(serde_json::Value::as_u64) .unwrap_or(0); - let status = run["status"].as_str().unwrap_or("?"); + let st = run["status"].as_str().unwrap_or("?"); let event = run .get("event") .and_then(serde_json::Value::as_str) @@ -2881,21 +3332,56 @@ async fn cmd_status(run_id: Option) -> anyhow::Result<()> { }) .unwrap_or("?"); let push = match run.get("push_state").and_then(|state| state.get("status")) { - Some(status) => { - let status = status.as_str().unwrap_or("?"); + Some(s) => { + let s = s.as_str().unwrap_or("?"); match run["push_state"]["pr_number"].as_u64() { - Some(number) => format!("{status} #{number}"), - None => status.to_owned(), + Some(n) => format!("{s} #{n}"), + None => s.to_owned(), } } None => "-".to_owned(), }; println!( "{:<38} {:<6} {:<12} {:<12} {:<10} {}", - run_id, run_number, status, event, push, workflow + run_id, run_number, st, event, push, workflow ); } - Ok(()) +} + +fn condition_action(code: &str) -> &'static str { + match code { + "queue_no_registered_runner" => "register a runner or enable the pool", + "queue_label_mismatch" => "add a runner with that label", + "concurrency_queue_overflow" => "raise concurrency queue max or reduce parallelism", + "concurrency_group_starved" => "check concurrency group that starves others", + "scheduler_scan_stale" => "check scheduler scan heartbeat (restart if stuck)", + "scheduler_fire_late" => "check scheduler clock / load", + "pool_preparing" => "wait for image preparation to finish", + "pool_provisioning_deficit" => "check pool capacity / provision failures", + "pool_repeated_provision_failure" => "inspect provision logs and VM host capacity", + "runner_poll_stale" => "check runner connectivity and heartbeat", + "runner_lease_stale" => "check runner lease renewal", + "vm_sampler_stale" | "vm_sample_unavailable" | "vm_unreachable" => { + "check VM host sampler and SmolVM health" + } + "vm_host_memory_pressure" => "free host memory or reduce pool size", + "vm_host_cpu_throttled" => "reduce host CPU load or raise CPU quota", + "vm_host_oom_kill" => "check host OOM kills and runner memory", + "vm_sparse_disk_pressure" => "free disk space on VM data volume", + "store_write_failure" | "store_connection_down" => "check store connectivity and disk", + "storage_capacity_pressure" => "free disk space or run GC/prune", + "limit_drop_active" => "raise that limit or reduce load", + "limit_reject_active" => "raise that limit or back off", + "github_check_update_failure" => "check GitHub App permissions and network", + "github_terminal_check_pending" => "retry check update or check GitHub status", + "github_rate_limit_low" => "back off GitHub API or wait for rate-limit reset", + "github_installation_token_expiring" => "refresh GitHub installation token", + "debug_session_stale" => "close stale debug session", + "debug_audit_evicted" => "increase audit retention or flush audits", + "telemetry_export_failure" => "check OTLP endpoint and credentials", + "state_sampler_stale" | "task_stale" | "task_exited" => "check background task health", + _ => "see runbook for this condition", + } } async fn cmd_logs(args: LogsArgs) -> anyhow::Result<()> { @@ -3085,6 +3571,20 @@ mod tests { /// pairs. static TEST_ENV_MUTEX: std::sync::Mutex<()> = std::sync::Mutex::new(()); + #[test] + fn truncate_reason_cuts_on_char_boundary() { + // 200 four-byte characters = 800 bytes; byte-slicing at 300 would + // panic. The result must be a valid, shorter string ending with the + // ellipsis. + let long = "界".repeat(200); + let truncated = truncate_reason(&long); + assert!(truncated.ends_with('…')); + assert!(truncated.len() < long.len()); + assert!(truncated.is_char_boundary(truncated.len())); + // Short input passes through untouched. + assert_eq!(truncate_reason(" ok "), "ok"); + } + #[test] fn first_serve_token_creates_private_home_and_file() { let root = tempfile::tempdir().unwrap(); @@ -3282,6 +3782,9 @@ mod tests { false, std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), std::sync::Arc::new(std::sync::RwLock::new(std::collections::BTreeMap::new())), + std::sync::Arc::new(preloop_observability::status::PoolStatus::new( + preloop_observability::status::PoolSnapshot::default(), + )), ) .unwrap(); unsafe { @@ -3357,6 +3860,9 @@ mod tests { false, std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), std::sync::Arc::new(std::sync::RwLock::new(std::collections::BTreeMap::new())), + std::sync::Arc::new(preloop_observability::status::PoolStatus::new( + preloop_observability::status::PoolSnapshot::default(), + )), ) .unwrap(); unsafe { @@ -3760,14 +4266,23 @@ mod tests { #[test] fn status_parses() { let cli = parse(&["status"]).unwrap(); - assert!(matches!(cli.command, Command::Status { run_id: None })); - let cli = parse(&["status", "550e8400-e29b-41d4-a716-446655440000"]).unwrap(); - assert!(matches!( - cli.command, - Command::Status { - run_id: Some(ref id) - } if id == "550e8400-e29b-41d4-a716-446655440000" - )); + let Command::Status(args) = cli.command else { + panic!("expected Status"); + }; + assert!(!args.json); + assert_eq!(args.limit, 20); + let cli = parse(&["status", "--json", "--limit", "5"]).unwrap(); + let Command::Status(args) = cli.command else { + panic!("expected Status"); + }; + assert!(args.json); + assert_eq!(args.limit, 5); + // Default limit is 20 and --json defaults to false + let cli = parse(&["status", "--limit", "42"]).unwrap(); + let Command::Status(args) = cli.command else { + panic!("expected Status"); + }; + assert_eq!(args.limit, 42); } #[test] diff --git a/crates/preloop-observability/src/lib.rs b/crates/preloop-observability/src/lib.rs index 1d926147..7ebd6531 100644 --- a/crates/preloop-observability/src/lib.rs +++ b/crates/preloop-observability/src/lib.rs @@ -1,4 +1,4 @@ -//! `preloop-observability` — Step 2 of Plan 002. +//! `preloop-observability` — observability handle for Preloop. //! //! Small, explicit API with no dependency on server/orchestrator internals. Both //! `preloop` and `preloop-server` construct one handle/runtime before building @@ -11,6 +11,8 @@ //! - Always retain `stderr`/`journald` even when OTLP is configured. //! - `Debug` on config never reveals headers or credential-bearing endpoint parts. +pub mod status; + use std::collections::HashMap; use std::fmt; use std::sync::Arc; @@ -91,7 +93,7 @@ impl ObservabilityConfig { // CLI defaults to `info` when unset; the standalone server historically // used `EnvFilter::from_default_env()` with no fallback (silent when - // unset). We unify on `info` per Step 2. + // unset). We unify on `info`. let rust_log = std::env::var("RUST_LOG").unwrap_or_else(|_| "info".to_string()); let service_name = @@ -473,7 +475,7 @@ pub struct ObservabilityRuntime { impl ObservabilityRuntime { fn new(handle: Observability) -> Self { - // Step 2 does not install the global subscriber here — the binaries do + // Does not install the global subscriber here — the binaries do // that via `install_fmt_subscriber`. This runtime is the place for the // future OTel provider guards and the 2s flush on `Drop`. Self { @@ -511,10 +513,10 @@ impl ObservabilityRuntime { /// Attempt to flush exporters for at most 2s. Export failure is logged, never propagated. pub async fn shutdown(self) { - // Step 2 has no exporter worker yet; this is the bounded-flush seam for + // No exporter worker yet; this is the bounded-flush seam for // the future OTel BatchSpanProcessor / metrics reader. tokio::time::timeout(Duration::from_secs(2), async { - // No-op until OTel providers are wired in Step 3+. + // No-op until OTLP providers are wired. tokio::task::yield_now().await; }) .await @@ -529,7 +531,7 @@ impl fmt::Debug for ObservabilityRuntime { } // --------------------------------------------------------------------------- -// Tests — Step 2 gates +// Tests // --------------------------------------------------------------------------- #[cfg(test)] diff --git a/crates/preloop-observability/src/status.rs b/crates/preloop-observability/src/status.rs new file mode 100644 index 00000000..06bfa2b3 --- /dev/null +++ b/crates/preloop-observability/src/status.rs @@ -0,0 +1,481 @@ +//!OperationalSnapshot and supporting types. +//! + +use std::sync::Arc; + +use chrono::{DateTime, Utc}; +use parking_lot::RwLock; +use serde::{Deserialize, Serialize}; + +// --------------------------------------------------------------------------- +// Overall +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Overall { + Ok, + Degraded, + Blocked, + ShuttingDown, +} + +impl Default for Overall { + fn default() -> Self { + Self::Ok + } +} + +// --------------------------------------------------------------------------- +// Service +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ServiceSnapshot { + pub version: String, + pub instance_id: String, + pub uptime_seconds: u64, + pub shutdown_requested: bool, +} + +// --------------------------------------------------------------------------- +// Runs / Jobs / Concurrency / Scheduler / Runners +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct RunsSnapshot { + pub queued: u32, + pub in_progress: u32, + pub completed: u32, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct JobsSnapshot { + pub ready: u32, + pub dependency_blocked: u32, + pub concurrency_blocked: u32, + pub pending_expansion: u32, + pub expanding: u32, + pub claimable: u32, + pub unclaimable: u32, + pub oldest_ready_seconds: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct ConcurrencySnapshot { + pub groups_active: u32, + pub groups_contended: u32, + pub pending_holders: u32, + pub deepest_group_pending: u32, + pub queue_max_pending: usize, + pub overflow_cancellations: u64, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct SchedulerSnapshot { + pub enabled: bool, + pub schedules: u32, + pub last_scan_at: Option>, + pub next_fire_at: Option>, + pub fired: u64, + pub skipped_overlapping: u64, + pub late_fires: u64, + pub max_fire_delay_seconds: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct RunnersSnapshot { + pub registered: u32, + pub sessions: u32, + pub idle: u32, + pub busy: u32, + pub stale: u32, + pub max_poll_age_seconds: Option, + pub max_lease_age_seconds: Option, +} + +// --------------------------------------------------------------------------- +// Pool +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum PoolMode { + Warm, + OnDemand, + External, + Disabled, +} + +impl Default for PoolMode { + fn default() -> Self { + Self::Warm + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PoolSnapshot { + pub mode: PoolMode, + pub desired: u32, + pub preparing: bool, + pub building: u32, + pub provisioning: u32, + pub idle: u32, + pub busy: u32, + pub paused: u32, + pub consecutive_provision_failures: u32, + pub last_transition_at: Option>, + /// Consolidated queue depth (server -> pool signal, now via PoolStatus). + #[serde(default)] + pub queue_depth: u32, + /// Next job labels for golden selection (server -> pool). + #[serde(default)] + pub next_job_runs_on: Vec, + /// Pending provision token count (pool -> server). + #[serde(default)] + pub pending_registrations: u32, +} + +impl Default for PoolSnapshot { + fn default() -> Self { + Self { + mode: PoolMode::Warm, + desired: 0, + preparing: false, + building: 0, + provisioning: 0, + idle: 0, + busy: 0, + paused: 0, + consecutive_provision_failures: 0, + last_transition_at: None, + queue_depth: 0, + next_job_runs_on: Vec::new(), + pending_registrations: 0, + } + } +} + +/// Shared handle that the pool updates and the sampler reads. +/// +/// Consolidates the four ad-hoc `Option>` handles. Single writer (pool) +/// + multiple readers (sampler, status) — cheap `RwLock`. +#[derive(Debug, Clone, Default)] +pub struct PoolStatus { + inner: Arc>, + /// One-time provision tokens (separate from snapshot to avoid cloning large map on every snapshot). + pending_tokens: Arc>>, +} + +impl PoolStatus { + pub fn new(snapshot: PoolSnapshot) -> Self { + Self { + inner: Arc::new(RwLock::new(snapshot)), + pending_tokens: Arc::new(RwLock::new(std::collections::BTreeMap::new())), + } + } + + pub fn snapshot(&self) -> PoolSnapshot { + let mut snap = self.inner.read().clone(); + snap.pending_registrations = self.pending_tokens.read().len() as u32; + snap + } + + pub fn set_desired(&self, desired: u32) { + self.inner.write().desired = desired; + } + + pub fn set_preparing(&self, preparing: bool) { + self.inner.write().preparing = preparing; + } + + pub fn set_counts(&self, idle: u32, busy: u32, building: u32, provisioning: u32, paused: u32) { + let mut g = self.inner.write(); + g.idle = idle; + g.busy = busy; + g.building = building; + g.provisioning = provisioning; + g.paused = paused; + } + + pub fn record_provision_failure(&self) { + self.inner.write().consecutive_provision_failures += 1; + } + + pub fn clear_provision_failures(&self) { + self.inner.write().consecutive_provision_failures = 0; + } + + pub fn set_queue_depth(&self, depth: u32) { + self.inner.write().queue_depth = depth; + } + + pub fn set_next_job_runs_on(&self, labels: Vec) { + self.inner.write().next_job_runs_on = labels; + } + + pub fn insert_pending(&self, token: String, at: std::time::SystemTime) { + self.pending_tokens.write().insert(token, at); + // `snapshot()` derives `pending_registrations` from `pending_tokens`; + // writing it here too would be dead state plus a lock-order coupling + // between the two guards. + } + + pub fn remove_pending(&self, token: &str) -> bool { + self.pending_tokens.write().remove(token).is_some() + } + + pub fn pending_tokens_snapshot( + &self, + ) -> std::collections::BTreeMap { + self.pending_tokens.read().clone() + } +} + +// --------------------------------------------------------------------------- +// VMs +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum VmSource { + CgroupV2, + Process, + Mixed, + Unavailable, +} + +impl Default for VmSource { + fn default() -> Self { + Self::Unavailable + } +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct VmFleetSnapshot { + pub source: VmSource, + pub sample_age_seconds: Option, + pub capabilities: HashMap, + pub count: VmCount, + pub configured: VmConfigured, + pub host_usage: VmHostUsage, + pub top_consumers: Vec, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct VmCount { + pub runner: u32, + pub golden: u32, + pub unavailable: u32, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct VmConfigured { + pub vcpus: u32, + pub memory_bytes: u64, + pub storage_bytes: u64, + pub overlay_bytes: u64, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct VmHostUsage { + /// `None` means "not measured yet" — absent in JSON, not a real zero. + /// A consumer of `/api/v1/status` must not read an idle fleet from an + /// unmeasured one. + #[serde(skip_serializing_if = "Option::is_none")] + pub cpu_cores: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub memory_bytes: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub sparse_disk_allocated_bytes: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct VmTopConsumer { + pub machine_name: String, + pub role: String, + pub activity: String, + pub cpu_cores: f64, + pub memory_bytes: u64, + pub sparse_disk_allocated_bytes: u64, +} + +use std::collections::HashMap; + +// --------------------------------------------------------------------------- +// Store / Storage / Github / Debug / Telemetry +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StoreBackend { + Sqlite, + Postgres, +} + +impl Default for StoreBackend { + fn default() -> Self { + Self::Sqlite + } +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct StoreSnapshot { + pub backend: StoreBackend, + pub consecutive_failures: u32, + pub last_success_at: Option>, + pub last_failure_at: Option>, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct StorageComponent { + pub store: String, + pub bytes: u64, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct StorageSnapshot { + pub state_dir: String, + pub state_fs_free_bytes: Option, + pub state_fs_free_ratio: Option, + pub components: Vec, + pub last_gc_at: Option>, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct GithubSnapshot { + pub configured: bool, + pub last_webhook_at: Option>, + pub pending_check_updates: u32, + pub last_check_success_at: Option>, + pub last_check_failure_at: Option>, + pub rate_limit: Option, + pub installation_token_expires_in_seconds: Option, + pub token_cache: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct GithubRateLimit { + pub resource: String, + pub limit: u32, + pub remaining: u32, + pub reset_at: Option>, + pub observed_at: Option>, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct TokenCacheSnapshot { + pub hits: u64, + pub misses: u64, + pub ttl_seconds: u64, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct DebugSnapshot { + pub active_sessions: u32, + pub oldest_session_seconds: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct TelemetrySnapshot { + pub otlp_enabled: bool, + pub last_export_success_at: Option>, + pub last_export_failure_at: Option>, + pub dropped_records: u64, +} + +// --------------------------------------------------------------------------- +// Limits / Tasks (from heartbeat/limit registries) +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct LimitEntry { + pub limit: String, + pub value: usize, + pub dropped: u64, + pub rejected: u64, + pub last_at: Option>, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct TaskEntry { + pub name: String, + pub critical: bool, + pub heartbeat_age_seconds: f64, + pub state: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Condition { + pub code: String, + pub severity: String, + pub message: String, + pub exemplars: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConditionExemplar { + pub run_id: Option, + pub job_id: Option, + pub runner_id: Option, + pub machine_name: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct OperationalSnapshot { + pub schema_version: u32, + pub observed_at: DateTime, + pub snapshot_age_seconds: f64, + pub overall: Overall, + pub service: ServiceSnapshot, + pub runs: RunsSnapshot, + pub jobs: JobsSnapshot, + pub concurrency: ConcurrencySnapshot, + pub scheduler: SchedulerSnapshot, + pub runners: RunnersSnapshot, + pub pool: PoolSnapshot, + pub vms: VmFleetSnapshot, + pub store: StoreSnapshot, + pub storage: StorageSnapshot, + pub limits: Vec, + pub tasks: Vec, + pub github: GithubSnapshot, + pub debug: DebugSnapshot, + pub telemetry: TelemetrySnapshot, + pub conditions: Vec, +} + +impl Default for OperationalSnapshot { + fn default() -> Self { + Self { + schema_version: 1, + observed_at: Utc::now(), + snapshot_age_seconds: 0.0, + overall: Overall::Ok, + service: ServiceSnapshot { + // The host binary owns this value; this crate's version is + // meaningless to an operator reading status, and a stale + // value is worse than an empty one. + version: String::new(), + instance_id: String::new(), + uptime_seconds: 0, + shutdown_requested: false, + }, + runs: RunsSnapshot::default(), + jobs: JobsSnapshot::default(), + concurrency: ConcurrencySnapshot::default(), + scheduler: SchedulerSnapshot::default(), + runners: RunnersSnapshot::default(), + pool: PoolSnapshot::default(), + vms: VmFleetSnapshot::default(), + store: StoreSnapshot::default(), + storage: StorageSnapshot::default(), + limits: Vec::new(), + tasks: Vec::new(), + github: GithubSnapshot::default(), + debug: DebugSnapshot::default(), + telemetry: TelemetrySnapshot::default(), + conditions: Vec::new(), + } + } +} diff --git a/crates/preloop-observability/src/vm_telemetry.rs b/crates/preloop-observability/src/vm_telemetry.rs new file mode 100644 index 00000000..87b0b521 --- /dev/null +++ b/crates/preloop-observability/src/vm_telemetry.rs @@ -0,0 +1,125 @@ +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::{Duration, SystemTime}; + +use parking_lot::RwLock; + +use crate::status::{VmConfigured, VmCount, VmFleetSnapshot, VmHostUsage, VmSource, VmTopConsumer}; + +#[derive(Debug, Clone)] +pub struct VmRuntimeInfo { + pub name: String, + pub role: String, + pub activity: String, + pub pid: Option, + pub start_time: Option, + pub cpus: u16, + pub memory_mib: u32, + pub storage_gb: u32, + pub overlay_gb: Option, + pub data_dir: Option, + pub created_at: Option, +} + +#[derive(Debug, Default)] +pub struct VmTelemetryRegistry { + inner: RwLock>, +} + +impl VmTelemetryRegistry { + pub fn register(&self, info: VmRuntimeInfo) { + self.inner.write().insert(info.name.clone(), info); + } + + pub fn deregister(&self, name: &str) { + self.inner.write().remove(name); + } + + pub fn snapshot(&self) -> Vec { + self.inner.read().values().cloned().collect() + } +} + +pub fn sample_host(_pid: Option, _data_dir: Option<&Path>) -> HostSample { + HostSample::unavailable() +} + +#[derive(Debug, Clone)] +pub struct HostSample { + pub cpu_time_secs: Option, + pub throttled_secs: Option, + pub memory_bytes: Option, + pub memory_limit_bytes: Option, + pub pids_current: Option, + pub sparse_allocated_bytes: Option, + pub pid_valid: bool, +} + +impl HostSample { + pub fn unavailable() -> Self { + Self { + cpu_time_secs: None, + throttled_secs: None, + memory_bytes: None, + memory_limit_bytes: None, + pids_current: None, + sparse_allocated_bytes: None, + pid_valid: false, + } + } +} + +pub fn build_fleet_snapshot( + registry: &VmTelemetryRegistry, + sample_age: Option, + capabilities: HashMap, +) -> VmFleetSnapshot { + let infos = registry.snapshot(); + let runner = infos.iter().filter(|i| i.role == "runner").count() as u32; + let golden = infos.iter().filter(|i| i.role == "golden").count() as u32; + let vcpus: u32 = infos.iter().map(|i| u32::from(i.cpus)).sum(); + let memory_bytes: u64 = infos + .iter() + .map(|i| u64::from(i.memory_mib) * 1024 * 1024) + .sum(); + let storage_bytes: u64 = infos + .iter() + .map(|i| u64::from(i.storage_gb) * 1024 * 1024 * 1024) + .sum(); + let overlay_bytes: u64 = infos + .iter() + .filter_map(|i| i.overlay_gb.map(|v| u64::from(v) * 1024 * 1024 * 1024)) + .sum(); + + let source = if capabilities.get("cpu").copied().unwrap_or(false) { + VmSource::CgroupV2 + } else if capabilities.get("process").copied().unwrap_or(false) { + VmSource::Process + } else { + VmSource::Unavailable + }; + + VmFleetSnapshot { + source, + sample_age_seconds: sample_age.map(|d| d.as_secs_f64()), + capabilities, + count: VmCount { + runner, + golden, + unavailable: 0, + }, + configured: VmConfigured { + vcpus, + memory_bytes, + storage_bytes, + overlay_bytes, + }, + host_usage: VmHostUsage { + cpu_cores: 0.0, + memory_bytes: 0, + sparse_disk_allocated_bytes: 0, + }, + top_consumers: Vec::new(), + } +} diff --git a/crates/preloop-orchestrator/Cargo.toml b/crates/preloop-orchestrator/Cargo.toml index 2a5c4e29..0f2763fa 100644 --- a/crates/preloop-orchestrator/Cargo.toml +++ b/crates/preloop-orchestrator/Cargo.toml @@ -10,6 +10,7 @@ description = "Job scheduling and VM lifecycle orchestration for Preloop CI" [dependencies] preloop-vm = { path = "../preloop-vm" } preloop-gha-protocol = { path = "../preloop-gha-protocol" } +preloop-observability = { path = "../preloop-observability" } preloop-runner-server = { path = "../preloop-runner-server" } preloop-gha-parser = { path = "../preloop-gha-parser" } anyhow = { workspace = true } diff --git a/crates/preloop-orchestrator/src/lib.rs b/crates/preloop-orchestrator/src/lib.rs index 3c4cfd2f..86889d35 100644 --- a/crates/preloop-orchestrator/src/lib.rs +++ b/crates/preloop-orchestrator/src/lib.rs @@ -1675,6 +1675,11 @@ pub struct RunnerPoolConfig { /// queued-job starvation clock during the warm; it is cleared before /// the pool serves its first job. pub preparing_signal: Option>, + /// Consolidated pool handle (replaces the four ad-hoc Option> fields above). + /// When `Some`, the pool updates it and the server sampler reads it. + /// Retain the legacy fields for now for backwards compatibility; new code + /// should read/write `pool_status`. + pub pool_status: Option>, } /// Cache of environment-specific golden VMs. @@ -2254,6 +2259,9 @@ impl RunnerPool

{ if let Some(signal) = &self.config.preparing_signal { signal.store(true, std::sync::atomic::Ordering::Release); } + if let Some(ps) = &self.config.pool_status { + ps.set_preparing(true); + } ensure_host_externals(&self.config)?; if self.config.use_packed_artifact || self.config.control_socket.is_none() { self.prepare_artifact(true).await?; @@ -2289,6 +2297,9 @@ impl RunnerPool

{ if let Some(signal) = &self.config.preparing_signal { signal.store(false, std::sync::atomic::Ordering::Release); } + if let Some(ps) = &self.config.pool_status { + ps.set_preparing(false); + } let mut slots = JoinSet::new(); // Runners currently registered and waiting for work. Slots consult it @@ -4919,6 +4930,7 @@ chmod +x "$destination/bin/node" next_job_runs_on: None, pending_registrations: None, preparing_signal: None, + pool_status: None, } } diff --git a/crates/preloop-orchestrator/tests/runner_pool_lifecycle.rs b/crates/preloop-orchestrator/tests/runner_pool_lifecycle.rs index 99106f88..5b97048d 100644 --- a/crates/preloop-orchestrator/tests/runner_pool_lifecycle.rs +++ b/crates/preloop-orchestrator/tests/runner_pool_lifecycle.rs @@ -368,6 +368,7 @@ impl Fixture { use_fork: false, use_packed_artifact: false, name_prefix: format!("pool-{label}-{id}"), + pool_status: None, base_image: BASE_IMAGE.to_owned(), workspace: None, artifact_stem, diff --git a/crates/preloop-runner-server/src/bootstrap.rs b/crates/preloop-runner-server/src/bootstrap.rs index 7deddf79..46ba2098 100644 --- a/crates/preloop-runner-server/src/bootstrap.rs +++ b/crates/preloop-runner-server/src/bootstrap.rs @@ -48,6 +48,12 @@ pub struct ServerConfig { /// presents the matching provisioning token. pub pending_registrations: Option>>>, + /// Consolidated pool handle (replaces the four ad-hoc Option> fields). + /// When `Some`, the pool updates it and the sampler reads it. + pub pool_status: Option>, + /// Observability handle to clone into AppState (heartbeat, limits). + /// `None` falls back to `Observability::noop()` (tests). + pub observability: Option, /// `PRELOOP_REQUIRE_JOB_ASSIGNMENTS`: refuse to dispatch any job without /// a recorded assignment, including to external runners. pub require_job_assignments: bool, @@ -77,6 +83,7 @@ impl std::fmt::Debug for ServerConfig { .field("oidc_issuer", &self.oidc_issuer) .field("enable_scheduler", &self.enable_scheduler) .field("pending_registrations", &self.pending_registrations) + .field("pool_status", &self.pool_status) .field("require_job_assignments", &self.require_job_assignments) .finish() } @@ -182,7 +189,8 @@ pub(crate) async fn reap_once(shared: &Arc) { .state .pool_preparing .as_ref() - .is_some_and(|flag| flag.load(std::sync::atomic::Ordering::Acquire)); + .is_some_and(|flag| flag.load(std::sync::atomic::Ordering::Acquire)) + || shared.state.pool_status.snapshot().preparing; let queued_jobs: Vec<_> = inner.queue.iter().cloned().collect(); let in_queue: std::collections::BTreeSet<(RunId, JobId)> = queued_jobs .iter() @@ -397,10 +405,15 @@ async fn run_background_reaper(shared: Arc) { let mut interval = tokio::time::interval(Duration::from_secs(10)); // Skip the first tick interval.tick().await; + //Heartbeat for reaper (critical) — beat each interval, no cadence change. + let heartbeat = shared.state.observability.heartbeat().clone(); + let _reaper_handle = heartbeat.register("reaper", preloop_observability::Criticality::Critical); + heartbeat.beat("reaper"); while !shared.shutdown.is_cancelled() { tokio::select! { _ = interval.tick() => { + heartbeat.beat("reaper"); reap_once(&shared).await; } _ = shared.shutdown.cancelled() => { @@ -410,6 +423,202 @@ async fn run_background_reaper(shared: Arc) { } } +fn build_operational_snapshot_sync( + queue_len: usize, + pending_jobs_len: usize, + pending_expansions_len: usize, + expanding_len: usize, + runs_queued: u32, + runs_in_progress: u32, + runs_completed: u32, + registered: u32, + sessions_len: u32, + queue_jobs: Vec, + runner_labels: Vec>, + pool_snapshot: preloop_observability::status::PoolSnapshot, + observability: &preloop_observability::Observability, + started_at: std::time::Instant, + shutdown_requested: bool, + scheduler_enabled: bool, + state_dir: &std::path::Path, + github_configured: bool, +) -> preloop_observability::status::OperationalSnapshot { + use chrono::Utc; + use preloop_observability::status::*; + let now = Utc::now(); + let uptime = started_at.elapsed().as_secs(); + // Claimability: distinguish claimable vs unclaimable using existing runner label matching. + let (claimable, unclaimable) = if queue_jobs.is_empty() { + (0, 0) + } else if runner_labels.is_empty() { + (0, queue_jobs.len() as u32) + } else if pool_snapshot.preparing { + // Temporarily unclaimable while pool prepares. + (0, queue_jobs.len() as u32) + } else { + let mut claimable = 0u32; + for job in &queue_jobs { + let matches = runner_labels + .iter() + .any(|labels| crate::runtime_scheduling::job_matches_runner(&job.runs_on, labels)); + if matches { + claimable += 1; + } + } + (claimable, queue_jobs.len() as u32 - claimable) + }; + + let oldest_ready_seconds = None; // TODO: track queued_at + + OperationalSnapshot { + schema_version: 1, + observed_at: now, + snapshot_age_seconds: 0.0, + overall: if shutdown_requested { + Overall::ShuttingDown + } else { + Overall::Ok + }, + service: ServiceSnapshot { + version: env!("CARGO_PKG_VERSION").to_string(), + instance_id: observability.instance_id().to_string(), + uptime_seconds: uptime, + shutdown_requested, + }, + runs: RunsSnapshot { + queued: runs_queued, + in_progress: runs_in_progress, + completed: runs_completed, + }, + jobs: JobsSnapshot { + ready: queue_len as u32, + dependency_blocked: pending_jobs_len as u32, + concurrency_blocked: 0, + pending_expansion: pending_expansions_len as u32, + expanding: expanding_len as u32, + claimable, + unclaimable, + oldest_ready_seconds, + }, + concurrency: ConcurrencySnapshot::default(), + scheduler: SchedulerSnapshot { + enabled: scheduler_enabled, + ..Default::default() + }, + runners: RunnersSnapshot { + registered, + sessions: sessions_len, + idle: 0, + busy: 0, + stale: 0, + max_poll_age_seconds: None, + max_lease_age_seconds: None, + }, + pool: pool_snapshot, + vms: VmFleetSnapshot { + source: VmSource::Unavailable, + sample_age_seconds: None, + ..Default::default() + }, + store: StoreSnapshot::default(), + storage: { + // Per-component bytes for the state dir. Cheap `metadata` reads on + // the 5s tick; a recursive walk would belong on the 60s cadence and + // must never run under the state lock. + let component = |name: &str, path: std::path::PathBuf| StorageComponent { + store: name.to_string(), + bytes: std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0), + }; + StorageSnapshot { + state_dir: state_dir.display().to_string(), + state_fs_free_bytes: None, + state_fs_free_ratio: None, + components: vec![ + component("database", state_dir.join("preloop.db")), + component("cache", state_dir.join("cache")), + component("artifacts", state_dir.join("artifacts")), + ], + last_gc_at: None, + } + }, + limits: Vec::new(), + tasks: Vec::new(), + github: GithubSnapshot { + configured: github_configured, + ..Default::default() + }, + debug: DebugSnapshot::default(), + telemetry: TelemetrySnapshot::default(), + conditions: Vec::new(), + } +} + +async fn run_state_sampler(shared: Arc) { + let heartbeat = shared.state.observability.heartbeat().clone(); + let _handle = heartbeat.register( + "state_sampler", + preloop_observability::Criticality::Critical, + ); + heartbeat.beat("state_sampler"); + let mut interval = tokio::time::interval(Duration::from_secs(5)); + // Immediate sample then every 5s. + interval.tick().await; + loop { + tokio::select! { + _ = interval.tick() => { + heartbeat.beat("state_sampler"); + // Clone needed state under lock, release, then build. + let (queue_len, pending_jobs_len, pending_expansions_len, expanding_len, runs_queued, runs_in_progress, runs_completed, registered, sessions_len, queue_jobs, runner_labels, scheduler_enabled) = { + let inner = shared.state.inner.lock().await; + let queue_len = inner.queue.len(); + let pending_jobs_len = inner.pending_jobs.len(); + let pending_expansions_len = inner.pending_expansions.len(); + let expanding_len = inner.expanding.len(); + let mut q = 0u32; let mut ip = 0u32; let mut c = 0u32; + for run in inner.runs.values() { + match run.status { + ExecutionStatus::Queued => q += 1, + ExecutionStatus::InProgress => ip += 1, + s if s.is_terminal() => c += 1, + _ => {} + } + } + let registered = inner.runners.len() as u32; + let sessions_len = inner.sessions.len() as u32; + let queue_jobs = inner.queue.iter().cloned().collect::>(); + let runner_labels = inner.runners.values().map(|r| r.labels.clone()).collect::>(); + let scheduler_enabled = shared.state.scheduler.is_some(); + // Clone pool snapshot outside inner lock? It's cheap, do after. + (queue_len, pending_jobs_len, pending_expansions_len, expanding_len, q, ip, c, registered, sessions_len, queue_jobs, runner_labels, scheduler_enabled) + }; + let pool_snapshot = shared.state.pool_status.snapshot(); + let snap = build_operational_snapshot_sync( + queue_len, + pending_jobs_len, + pending_expansions_len, + expanding_len, + runs_queued, + runs_in_progress, + runs_completed, + registered, + sessions_len, + queue_jobs, + runner_labels, + pool_snapshot, + &shared.state.observability, + shared.state.started_at, + shared.shutdown.is_cancelled(), + scheduler_enabled, + &shared.state.state_dir, + shared.state.github_app.is_some(), + ); + *shared.state.status_snapshot.write() = snap; + } + _ = shared.shutdown.cancelled() => break, + } + } +} + fn is_routine_unix_disconnect(error: &(dyn std::error::Error + 'static)) -> bool { error .downcast_ref::() @@ -424,6 +633,39 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> { config.store_url.as_deref(), ) .await?; + // Wire observability if supplied (CLI/server will pass its handle). + if let Some(obs) = config.observability.clone() { + state.observability = obs; + } + if let Some(ps) = config.pool_status.clone() { + state.pool_status = ps; + } + // Ensure uptime base is now (AppState::new set it, but re-arm after store load). + state.started_at = std::time::Instant::now(); + // Seed initial snapshot so /readyz and /status have data before first 5s tick. + { + let init = build_operational_snapshot_sync( + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + 0, + Vec::new(), + Vec::new(), + state.pool_status.snapshot(), + &state.observability, + state.started_at, + false, + false, + &state.state_dir, + state.github_app.is_some(), + ); + *state.status_snapshot.write() = init; + } if let Some(queue_depth) = config.queue_depth.clone() { state.queue_depth = queue_depth; // The pool shares this same atomic and only forks a runner while it @@ -435,23 +677,46 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> { state.inner.lock().await.queue.len(), std::sync::atomic::Ordering::Release, ); + state + .pool_status + .set_queue_depth(state.queue_depth.load(std::sync::atomic::Ordering::Acquire) as u32); } if let Some(next_job_runs_on) = config.next_job_runs_on.clone() { state.next_job_runs_on = next_job_runs_on; + if let Ok(v) = state.next_job_runs_on.read() { + state.pool_status.set_next_job_runs_on(v.clone()); + } } { let inner = state.inner.lock().await; crate::runtime_scheduling::sync_next_job_labels(&inner, &state.next_job_runs_on); + if state.pool_status.snapshot().next_job_runs_on.is_empty() { + if let Ok(v) = state.next_job_runs_on.read() { + state.pool_status.set_next_job_runs_on(v.clone()); + } + } } { let pool_managed = config.pending_registrations.is_some(); if let Some(pending_registrations) = config.pending_registrations.clone() { state.pending_registrations = pending_registrations; + // Mirror into consolidated handle for sampler visibility + if let Ok(map) = state.pending_registrations.read() { + for (k, v) in map.iter() { + state.pool_status.insert_pending(k.clone(), *v); + } + } } let mut inner = state.inner.lock().await; inner.pool_assignments_enabled = pool_managed; inner.require_job_assignments = config.require_job_assignments; } + if let Some(pool_preparing) = config.pool_preparing.clone() { + state.pool_preparing = Some(pool_preparing.clone()); + if pool_preparing.load(std::sync::atomic::Ordering::Acquire) { + state.pool_status.set_preparing(true); + } + } if !config.listen.ip().is_loopback() && state.system_token == DEFAULT_PRELOOP_SYSTEM_TOKEN { anyhow::bail!( "PRELOOP_SYSTEM_TOKEN must be explicitly configured when listening beyond loopback" @@ -467,6 +732,8 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> { inner.oidc_issuer = oidc_issuer; } let shutdown = CancellationToken::new(); + // Heartbeat for scheduler scan (critical) if enabled — beat periodically. + let scheduler_heartbeat = state.observability.heartbeat().clone(); if config.enable_scheduler { let scheduler = crate::scheduler::Scheduler::new(); state.scheduler = Some(scheduler.clone()); @@ -475,6 +742,24 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> { shutdown: shutdown.clone(), }); let scheduler_clone = scheduler.clone(); + // Spawn a holder task for scheduler_scan heartbeat — keep handle alive for lifetime. + let hb_for_holder = scheduler_heartbeat.clone(); + let shutdown_for_holder = shutdown.clone(); + tokio::spawn(async move { + let _handle = hb_for_holder.register( + "scheduler_scan", + preloop_observability::Criticality::Critical, + ); + hb_for_holder.beat("scheduler_scan"); + let mut int = tokio::time::interval(Duration::from_secs(10)); + int.tick().await; + while !shutdown_for_holder.is_cancelled() { + tokio::select! { + _ = int.tick() => hb_for_holder.beat("scheduler_scan"), + _ = shutdown_for_holder.cancelled() => break, + } + } + }); if let Some(workspace) = state.local_workspace.clone() { tokio::spawn(async move { scheduler_clone @@ -558,12 +843,17 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> { }; let router = build_app(state.clone(), shutdown.clone(), test_api_token); - state.pool_preparing = config.pool_preparing.clone(); let shared = Arc::new(SharedState { - state, + state: state.clone(), shutdown: shutdown.clone(), }); + // 5s sampler — clone needed state under lock, release, then publish. + let sampler_shared = shared.clone(); + tokio::spawn(async move { + run_state_sampler(sampler_shared).await; + }); + let checker_shared = shared.clone(); tokio::spawn(async move { run_background_reaper(checker_shared).await; diff --git a/crates/preloop-runner-server/src/lib_tests.rs b/crates/preloop-runner-server/src/lib_tests.rs index 91e5d4f8..1a4d802d 100644 --- a/crates/preloop-runner-server/src/lib_tests.rs +++ b/crates/preloop-runner-server/src/lib_tests.rs @@ -18536,6 +18536,8 @@ fn server_config_debug_redacts_store_url_password() { oidc_issuer: None, enable_scheduler: false, pending_registrations: None, + pool_status: None, + observability: None, require_job_assignments: false, }; let debug = format!("{config:?}"); diff --git a/crates/preloop-runner-server/src/main.rs b/crates/preloop-runner-server/src/main.rs index aed62825..44c46253 100644 --- a/crates/preloop-runner-server/src/main.rs +++ b/crates/preloop-runner-server/src/main.rs @@ -75,10 +75,10 @@ async fn main() -> anyhow::Result<()> { .install_default() .ok(); - // Unified observability init (Step 2): `RUST_LOG` now defaults to `info` + // Unified observability init: `RUST_LOG` now defaults to `info` // like the CLI, instead of falling silent when unset. `PRELOOP_LOG_FORMAT` // controls pretty/json/auto. The `Observability` handle will be cloned - // into `AppState` in Step 3; for now it is held for the life of `main`. + // into `AppState`; for now it is held for the life of `main`. let obs_config = preloop_observability::ObservabilityConfig::from_env(); let (observability, observability_runtime) = preloop_observability::Observability::from_config(obs_config); @@ -130,6 +130,8 @@ async fn main() -> anyhow::Result<()> { next_job_runs_on: None, pool_preparing: None, listen, + pool_status: None, + observability: None, systemd_socket_activation: false, unix_socket, state_dir, diff --git a/crates/preloop-runner-server/src/openapi.rs b/crates/preloop-runner-server/src/openapi.rs index 068f60e9..2d65b6f0 100644 --- a/crates/preloop-runner-server/src/openapi.rs +++ b/crates/preloop-runner-server/src/openapi.rs @@ -179,7 +179,10 @@ pub(crate) struct RunResponse { list_dispatch_runs, github_register, github_callback, - list_runners + list_runners, + readyz, + status, + metrics ), components( schemas( @@ -282,6 +285,38 @@ type JsonValue = serde_json::Value; )] fn healthz() {} +/// Server readiness check (public, reason codes on 503). +#[utoipa::path( + get, path = "/readyz", tag = "Health", + responses( + (status = 200, description = "Ready", body = JsonValue), + (status = 503, description = "Not ready", body = JsonValue) + ) +)] +fn readyz() {} + +/// Operational status snapshot (native bearer required). +#[utoipa::path( + get, path = "/api/v1/status", tag = "Health", + responses( + (status = 200, description = "Operational snapshot", body = JsonValue), + (status = 401, description = "Unauthorized", body = ApiErrorResponse) + ), + security(("native_bearer" = [])) +)] +fn status() {} + +/// Prometheus metrics (native bearer required). +#[utoipa::path( + get, path = "/metrics", tag = "Health", + responses( + (status = 200, description = "Prometheus text", content_type = "text/plain", body = String), + (status = 401, description = "Unauthorized", body = ApiErrorResponse) + ), + security(("native_bearer" = [])) +)] +fn metrics() {} + // ── Runs ──────────────────────────────────────────────────────────────────── /// Submit a workflow run. diff --git a/crates/preloop-runner-server/src/routes.rs b/crates/preloop-runner-server/src/routes.rs index d513e3a6..3757938f 100644 --- a/crates/preloop-runner-server/src/routes.rs +++ b/crates/preloop-runner-server/src/routes.rs @@ -252,8 +252,18 @@ pub(crate) fn build_app( crate::dispatch_auth::require_dispatch_auth, )); + let observability_routes = Router::new() + .route("/api/v1/status", get(status)) + .route("/metrics", get(metrics)) + .route_layer(middleware::from_fn_with_state( + shared.clone(), + require_native_bearer, + )); + let router = Router::new() .route("/healthz", get(healthz)) + .route("/readyz", get(readyz)) + .merge(observability_routes) .route("/runs/:run_id", get(get_public_run)) .route( "/openapi.json", diff --git a/crates/preloop-runner-server/src/runs.rs b/crates/preloop-runner-server/src/runs.rs index 90801dbf..b32a62c5 100644 --- a/crates/preloop-runner-server/src/runs.rs +++ b/crates/preloop-runner-server/src/runs.rs @@ -1,12 +1,137 @@ use super::*; use std::collections::BTreeSet; -pub(crate) async fn healthz(State(shared): State>) -> Json { - Json(json!({ - "ok": true, +pub(crate) async fn healthz(State(shared): State>) -> impl IntoResponse { + let shutdown = shared.shutdown.is_cancelled(); + let body = json!({ + "ok": !shutdown, "protocol_version": PROTOCOL_VERSION, - "shutdown_requested": shared.shutdown.is_cancelled(), - })) + "shutdown_requested": shutdown, + }); + if shutdown { + (StatusCode::SERVICE_UNAVAILABLE, Json(body)).into_response() + } else { + (StatusCode::OK, Json(body)).into_response() + } +} + +pub(crate) async fn readyz(State(shared): State>) -> impl IntoResponse { + if shared.shutdown.is_cancelled() { + let body = json!({ "ready": false, "reason": "shutting_down" }); + return (StatusCode::SERVICE_UNAVAILABLE, Json(body)).into_response(); + } + // Check critical heartbeats freshness (>15s stale is 3 intervals of 5s sampler) + if let Some(stale) = shared + .state + .observability + .heartbeat() + .any_critical_stale(Duration::from_secs(15)) + { + let body = json!({ "ready": false, "reason": format!("task_stale:{}", stale) }); + return (StatusCode::SERVICE_UNAVAILABLE, Json(body)).into_response(); + } + // Also check snapshot age (>15s stale sampler) + let age_secs = { + let snap = shared.state.status_snapshot.read(); + let now = chrono::Utc::now(); + (now - snap.observed_at).num_milliseconds() as f64 / 1000.0 + }; + if age_secs > 15.0 { + let body = json!({ "ready": false, "reason": "state_sampler_stale" }); + return (StatusCode::SERVICE_UNAVAILABLE, Json(body)).into_response(); + } + let body = json!({ "ready": true, "reason": serde_json::Value::Null }); + (StatusCode::OK, Json(body)).into_response() +} + +pub(crate) async fn status(State(shared): State>) -> impl IntoResponse { + // Fail-open, no InnerState lock — clone cached snapshot and update age. + let mut snap = shared.state.status_snapshot.read().clone(); + let now = chrono::Utc::now(); + let age = (now - snap.observed_at).num_milliseconds() as f64 / 1000.0; + snap.snapshot_age_seconds = if age.is_finite() && age >= 0.0 { + age + } else { + 0.0 + }; + // Also surface current heartbeat tasks without holding InnerState + // (best-effort: caller sees last sampler's tasks plus live heartbeat snapshot) + // We keep sampler's tasks but also append live task snapshot if empty. + if snap.tasks.is_empty() { + snap.tasks = shared + .state + .observability + .heartbeat() + .snapshot() + .into_iter() + .map(|t| preloop_observability::status::TaskEntry { + name: t.name.to_string(), + critical: t.critical == preloop_observability::Criticality::Critical, + heartbeat_age_seconds: t.heartbeat_age.as_secs_f64(), + state: if t.exited { + "exited".to_string() + } else if t.heartbeat_age > Duration::from_secs(15) { + "stale".to_string() + } else { + "running".to_string() + }, + }) + .collect(); + } + Json(snap).into_response() +} + +pub(crate) async fn metrics(State(shared): State>) -> impl IntoResponse { + let snap = shared.state.status_snapshot.read().clone(); + let mut out = String::new(); + out.push_str("# HELP preloop_service_uptime_seconds Service uptime in seconds.\n"); + out.push_str("# TYPE preloop_service_uptime_seconds gauge\n"); + out.push_str(&format!( + "preloop_service_uptime_seconds {}\n", + snap.service.uptime_seconds + )); + out.push_str("# HELP preloop_pool_desired Desired pool size.\n"); + out.push_str("# TYPE preloop_pool_desired gauge\n"); + out.push_str(&format!("preloop_pool_desired {}\n", snap.pool.desired)); + out.push_str("# HELP preloop_pool_preparing Pool preparing signal.\n"); + out.push_str("# TYPE preloop_pool_preparing gauge\n"); + out.push_str(&format!( + "preloop_pool_preparing {}\n", + if snap.pool.preparing { 1 } else { 0 } + )); + out.push_str("# HELP preloop_pool_idle Idle runners in pool.\n"); + out.push_str("# TYPE preloop_pool_idle gauge\n"); + out.push_str(&format!("preloop_pool_idle {}\n", snap.pool.idle)); + out.push_str("# HELP preloop_pool_busy Busy runners in pool.\n"); + out.push_str("# TYPE preloop_pool_busy gauge\n"); + out.push_str(&format!("preloop_pool_busy {}\n", snap.pool.busy)); + out.push_str("# HELP preloop_job_queue_depth Number of jobs by queue.\n"); + out.push_str("# TYPE preloop_job_queue_depth gauge\n"); + out.push_str(&format!( + "preloop_job_queue_depth{{queue=\"ready\"}} {}\n", + snap.jobs.ready + )); + out.push_str(&format!( + "preloop_job_queue_depth{{queue=\"claimable\"}} {}\n", + snap.jobs.claimable + )); + out.push_str(&format!( + "preloop_job_queue_depth{{queue=\"unclaimable\"}} {}\n", + snap.jobs.unclaimable + )); + out.push_str(&format!( + "preloop_job_queue_depth{{queue=\"dependency_blocked\"}} {}\n", + snap.jobs.dependency_blocked + )); + let body = out; + ( + [( + header::CONTENT_TYPE, + "text/plain; version=0.0.4; charset=utf-8", + )], + body, + ) + .into_response() } /// GitHub's `system.orchestrationId`: `{planId}.{jobId}.{suffix}` where the diff --git a/crates/preloop-runner-server/src/state.rs b/crates/preloop-runner-server/src/state.rs index f06e70b3..40714d52 100644 --- a/crates/preloop-runner-server/src/state.rs +++ b/crates/preloop-runner-server/src/state.rs @@ -364,6 +364,15 @@ pub struct AppState { /// lock. Monotonically increases; the inner counter is no longer the /// source of truth once this is in use. pub(crate) next_request_id: Arc, + /// Observability handle (cloneable, holds heartbeat & limit registries). + pub(crate) observability: preloop_observability::Observability, + /// Cached operational snapshot, updated every 5s by the sampler without holding `inner`. + pub status_snapshot: + Arc>, + /// Consolidated pool handle replacing the four ad-hoc Option> fields. + pub pool_status: Arc, + /// When this AppState was created (for uptime). + pub(crate) started_at: std::time::Instant, /// Jobs accepted and still waiting for a runner, refreshed whenever one /// is claimed. A supervising runner pool reads it to decide whether the /// work already queued outruns the runners it has left. @@ -786,6 +795,12 @@ impl AppState { events, message_notify: Arc::new(Notify::new()), next_request_id: Arc::new(std::sync::atomic::AtomicI64::new(next_request_id)), + observability: preloop_observability::Observability::noop(), + status_snapshot: Arc::new(parking_lot::RwLock::new( + preloop_observability::status::OperationalSnapshot::default(), + )), + pool_status: Arc::new(preloop_observability::status::PoolStatus::default()), + started_at: std::time::Instant::now(), // Mirror the recovered ready-queue size so an on-demand runner // pool spawns against the right workload after restart. queue_depth: Arc::new(std::sync::atomic::AtomicUsize::new(recovered_queue_len)), diff --git a/crates/preloop-vm/Cargo.toml b/crates/preloop-vm/Cargo.toml index 4c9deec8..86410f59 100644 --- a/crates/preloop-vm/Cargo.toml +++ b/crates/preloop-vm/Cargo.toml @@ -9,6 +9,7 @@ description = "VM provider abstraction for Preloop CI" [dependencies] async-trait = { workspace = true } +preloop-observability = { path = "../preloop-observability" } thiserror = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } diff --git a/crates/preloop-vm/src/lib.rs b/crates/preloop-vm/src/lib.rs index 75957770..6ecd96f5 100644 --- a/crates/preloop-vm/src/lib.rs +++ b/crates/preloop-vm/src/lib.rs @@ -14,6 +14,8 @@ use tokio::process::Command; use tokio::sync::mpsc; use tracing::warn; +pub mod telemetry; + const DEFAULT_CAPTURE_LIMIT: usize = 1024 * 1024; /// A validated persistent SmolVM machine name. diff --git a/crates/preloop-vm/src/telemetry.rs b/crates/preloop-vm/src/telemetry.rs new file mode 100644 index 00000000..301e7190 --- /dev/null +++ b/crates/preloop-vm/src/telemetry.rs @@ -0,0 +1,6 @@ +//! Re-export VM telemetry types from `preloop-observability` so `preloop-vm` +//! and `preloop-orchestrator` share the same registry without a circular dep. + +pub use preloop_observability::vm_telemetry::{ + build_fleet_snapshot, sample_host, HostSample, VmRuntimeInfo, VmTelemetryRegistry, +};