diff --git a/Cargo.lock b/Cargo.lock index 4855044f..2b038903 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2020,9 +2020,10 @@ dependencies = [ name = "preloop-observability" version = "0.1.0" dependencies = [ - "anyhow", "chrono", "parking_lot", + "rand 0.8.6", + "reqwest", "serde", "serde_json", "tokio", @@ -2191,6 +2192,7 @@ version = "0.21.0" dependencies = [ "async-trait", "parking_lot", + "preloop-observability", "serde", "serde_json", "tar", diff --git a/crates/preloop-runner-server/src/bootstrap.rs b/crates/preloop-runner-server/src/bootstrap.rs index 74ae0175..87fae212 100644 --- a/crates/preloop-runner-server/src/bootstrap.rs +++ b/crates/preloop-runner-server/src/bootstrap.rs @@ -515,10 +515,16 @@ fn build_operational_snapshot_sync( max_lease_age_seconds: None, }, pool: pool_snapshot, - vms: VmFleetSnapshot { - source: VmSource::Unavailable, - sample_age_seconds: None, - ..Default::default() + vms: { + // Host sampler is stubbed until the cgroup parser lands; + // the registry is the source of truth for configured counts + // and will be populated by RunnerPool on create/fork. + let caps = std::collections::HashMap::new(); + preloop_observability::vm_telemetry::build_fleet_snapshot( + observability.vm_registry(), + None, + caps, + ) }, store: StoreSnapshot::default(), storage: { @@ -533,11 +539,11 @@ fn build_operational_snapshot_sync( 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")), - ], + // Only the database is a single file. `cache` and `artifacts` + // are directories, whose `metadata().len()` is the inode + // size, not the contents size; they need the recursive walk + // on the 60s cadence. + components: vec![component("database", state_dir.join("preloop.db"))], last_gc_at: None, } }, @@ -644,15 +650,16 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> { { // One decorator around the private `Store` trait — never per-backend // duplication. The backend label is bounded to sqlite|postgres. - let backend = if config + // Mirror `open_store` precedence: the explicit URL wins, and the + // environment is consulted only when none was supplied. Otherwise a + // stale `PRELOOP_STORE_URL=postgres://…` would mislabel every + // SQLite operation. + let effective_url = config .store_url - .as_deref() - .map(|u| u.contains("postgres")) - .unwrap_or(false) - || std::env::var("PRELOOP_STORE_URL") - .map(|v| v.contains("postgres")) - .unwrap_or(false) - { + .clone() + .or_else(|| std::env::var("PRELOOP_STORE_URL").ok()) + .unwrap_or_default(); + let backend = if effective_url.contains("postgres") { "postgres" } else { "sqlite" @@ -768,33 +775,33 @@ 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, - } - } - }); + // The scheduler heartbeat must prove the startup scan progressed, not + // that an unrelated timer is awake. The scan tasks beat it per + // workflow file and deregister on completion: a scan that hangs or + // panics stops beating and `/readyz` goes 503; a completed scan is + // not a stale critical task. + let scan_hb = scheduler_heartbeat.clone(); if let Some(workspace) = state.local_workspace.clone() { + let shared_for_scan = shared_for_scan.clone(); tokio::spawn(async move { + let handle = scan_hb.register( + "scheduler_scan", + preloop_observability::Criticality::Critical, + ); scheduler_clone - .scan_workspace(&workspace, shared_for_scan) + .scan_workspace(&workspace, shared_for_scan, Some(handle)) .await; }); } else { + let shared_for_scan = shared_for_scan.clone(); tokio::spawn(async move { - scheduler_clone.scan_remote(shared_for_scan).await; + let handle = scan_hb.register( + "scheduler_scan", + preloop_observability::Criticality::Critical, + ); + scheduler_clone + .scan_remote(shared_for_scan, Some(handle)) + .await; }); } } diff --git a/crates/preloop-runner-server/src/broker.rs b/crates/preloop-runner-server/src/broker.rs index 603c2b47..b8b73fa8 100644 --- a/crates/preloop-runner-server/src/broker.rs +++ b/crates/preloop-runner-server/src/broker.rs @@ -271,6 +271,7 @@ pub(crate) async fn next_message_broker_ref( .queue_depth .store(inner.queue.len(), std::sync::atomic::Ordering::Release); runtime_scheduling::sync_next_job_labels(&inner, &shared.state.next_job_runs_on); + record_claim_queue_wait(&shared, &claimed); let Some(queued) = claimed else { drop(inner); if wait_seconds == 0 { @@ -397,6 +398,28 @@ pub(crate) async fn broker_session_root( )) } +/// Record how long a claimed job sat in the ready queue. `enqueued_at` is +/// stamped when the job enters the queue and survives requeues, so a job +/// that bounced off a purged runner still measures total queue time. Jobs +/// restored from a snapshot without the field (`0`) are not recorded. +fn record_claim_queue_wait(shared: &Arc, claimed: &Option) { + let Some(queued) = claimed else { return }; + if queued.enqueued_at_unix_nanos <= 0 { + return; + } + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos() as i64) + .unwrap_or(0); + let elapsed = std::time::Duration::from_nanos((now - queued.enqueued_at_unix_nanos) as u64); + shared + .state + .observability + .metrics() + .lifecycle + .record_queue_wait("claimed", elapsed); +} + pub(crate) async fn broker_delete_session_root( State(shared): State>, Query(params): Query>, @@ -409,13 +432,15 @@ pub(crate) async fn broker_delete_session_root( if let Some(session_id) = header_session.or_else(|| params.get("sessionId").map(String::as_str)) { remove_broker_session(&shared, session_id, runner_id).await?; + shared + .state + .observability + .metrics() + .lifecycle + .record_session_transition("delete", "ok"); } - shared - .state - .observability - .metrics() - .lifecycle - .record_session_transition("delete", "ok"); + // No session id present: nothing was deleted, so no transition is + // recorded — a 204 with no-op must not count as a successful delete. Ok(StatusCode::NO_CONTENT) } @@ -598,6 +623,7 @@ pub(crate) async fn next_message_broker_ref_root( .queue_depth .store(inner.queue.len(), std::sync::atomic::Ordering::Release); runtime_scheduling::sync_next_job_labels(&inner, &shared.state.next_job_runs_on); + record_claim_queue_wait(&shared, &claimed); if let Some(queued) = claimed { if let Some(run) = inner.runs.get_mut(&queued.run_id) { run.status = ExecutionStatus::InProgress; @@ -873,13 +899,11 @@ pub(crate) async fn broker_acquire_job( message.request_id = 0; let payload = serde_json::to_value(&message) .map_err(|error| ApiError::internal(format!("serialize broker job payload: {error}")))?; - // Queue wait and broker poll outcomes — bounded, exactly one per successful claim. - shared - .state - .observability - .metrics() - .lifecycle - .record_queue_wait("claimed", std::time::Duration::from_secs(1)); + // Broker poll outcome — bounded, exactly one per successful claim. Queue + // wait is recorded at the claim sites in `next_message_broker_ref` / + // `next_message_disttask`, where the enqueue timestamp is still on the + // job; by the time the acquire payload is built the queue position has + // been lost. shared .state .observability diff --git a/crates/preloop-runner-server/src/scheduler.rs b/crates/preloop-runner-server/src/scheduler.rs index 4423e979..3d4df827 100644 --- a/crates/preloop-runner-server/src/scheduler.rs +++ b/crates/preloop-runner-server/src/scheduler.rs @@ -301,7 +301,12 @@ impl Scheduler { /// Scan a local workspace directory and register all schedule workflows. /// /// Called once on server startup when `--enable-scheduler` is active. - pub async fn scan_workspace(self: &Arc, workspace: &PathBuf, shared: Arc) { + pub async fn scan_workspace( + self: &Arc, + workspace: &PathBuf, + shared: Arc, + heartbeat: Option, + ) { let wf_dir = workspace.join(".github").join("workflows"); let entries = match std::fs::read_dir(&wf_dir) { Ok(e) => e, @@ -309,6 +314,9 @@ impl Scheduler { }; for entry in entries.flatten() { let path = entry.path(); + if let Some(hb) = &heartbeat { + hb.beat(); + } if !matches!( path.extension().and_then(|e| e.to_str()), Some("yml") | Some("yaml") @@ -375,7 +383,14 @@ impl Scheduler { /// Fetch and install schedules for a remote-backed server at startup. /// The repository and token use `PRELOOP_GITHUB_REPOSITORY` and /// `PRELOOP_GITHUB_TOKEN`, the same explicit remote workflow configuration. - pub async fn scan_remote(self: &Arc, shared: Arc) { + pub async fn scan_remote( + self: &Arc, + shared: Arc, + heartbeat: Option, + ) { + if let Some(hb) = &heartbeat { + hb.beat(); + } let (Ok(repository), Ok(token)) = ( std::env::var("PRELOOP_GITHUB_REPOSITORY"), std::env::var("PRELOOP_GITHUB_TOKEN"), diff --git a/crates/preloop-runner-server/src/state.rs b/crates/preloop-runner-server/src/state.rs index b62e2e5d..76635cec 100644 --- a/crates/preloop-runner-server/src/state.rs +++ b/crates/preloop-runner-server/src/state.rs @@ -607,12 +607,16 @@ fn bounded_termination_reason(value: &str) -> &'static str { // - the external-host check, where the server has no runner of that // platform class at all (`no {platform} runner is registered with // this server, so `runs-on: …` cannot be scheduled`). - if value.contains("runner is registered with this server") { - return "no_platform_runner"; - } + // The starvation prose interpolates workflow-controlled `runs-on` + // labels, so the anchored prefix MUST be checked before the substring: + // a crafted label containing the platform phrase must not flip a + // starvation reason into `no_platform_runner`. if value.starts_with("no runner is registered for") { return "no_runner"; } + if value.contains("runner is registered with this server") { + return "no_platform_runner"; + } if value.starts_with("job exceeded its timeout") || value.starts_with("timed out") || value.contains("timeout-minutes") @@ -628,6 +632,25 @@ fn bounded_termination_reason(value: &str) -> &'static str { "unrecognized" } +/// Bound free-form reason prose for export as a telemetry attribute. The +/// prose interpolates workflow input (e.g. `runs-on` labels), so one job +/// must not emit an arbitrarily large attribute. Truncation cuts on a +/// character boundary — byte slicing panics on multi-byte input. +fn bounded_reason_detail(detail: &str) -> String { + const DETAIL_MAX: usize = 512; + let mut detail = detail.to_string(); + if detail.len() > DETAIL_MAX { + let cut = detail + .char_indices() + .map(|(i, _)| i) + .take_while(|&i| i <= DETAIL_MAX) + .last() + .unwrap_or(0); + detail.truncate(cut); + } + detail +} + impl AppState { pub async fn new(state_dir: PathBuf) -> anyhow::Result { let config_path = crate::config::config_path(); @@ -961,27 +984,17 @@ impl AppState { // it an `unrecognized` classification is a dead end — // you cannot tell which path produced it. if let Some(detail) = reason.as_deref() { - attributes.push(("reason.detail".to_string(), detail.to_string())); + attributes + .push(("reason.detail".to_string(), bounded_reason_detail(detail))); } attributes }, ); } - NdjsonEvent::JobCompleted { status, .. } if status.is_terminal() => { - let conclusion = execution_conclusion(*status); - self.observability - .metrics() - .lifecycle - .record_job_completed(conclusion, "completed"); - self.observability.export_log( - "INFO", - "job.completed", - vec![ - ("event.name".to_string(), "job.completed".to_string()), - ("conclusion".to_string(), conclusion.to_string()), - ], - ); - } + // `NdjsonEvent::JobCompleted` has no constructor anywhere in the + // workspace — the terminal transition is reported as a terminal + // `JobStatus`, which the arm above records. Keeping a counter + // call here would make the record look double-sourced. _ => {} } // Capture the projection under the lock, then persist after releasing @@ -1479,6 +1492,29 @@ mod termination_reason_tests { } } + #[test] + fn reason_detail_is_bounded_on_a_char_boundary() { + use super::bounded_reason_detail; + // 4-byte characters: 300 of them is 1200 bytes, way over the cap. + let long = "w".repeat(300); + let bounded = bounded_reason_detail(&long); + assert!(bounded.len() <= 512); + assert!(bounded.is_char_boundary(bounded.len())); + // Short prose passes through untouched. + assert_eq!(bounded_reason_detail("short"), "short"); + } + + #[test] + fn crafted_runs_on_label_cannot_flip_the_classification() { + // The starvation prose interpolates `runs-on` labels verbatim. A + // label containing the platform phrase must still classify as + // `no_runner` — the anchored prefix is checked first. + let starved = "no runner is registered for `runs-on: self-hosted, \ + runner is registered with this server` and none \ + appeared within 120s, so the job cannot be scheduled"; + assert_eq!(bounded_termination_reason(starved), "no_runner"); + } + #[test] fn platform_and_starvation_do_not_collide() { let starved = "no runner is registered for `runs-on: self-hosted, Linux, ARM64` and none \ diff --git a/crates/preloop-runner-server/src/store.rs b/crates/preloop-runner-server/src/store.rs index 8ec61c5d..d850c892 100644 --- a/crates/preloop-runner-server/src/store.rs +++ b/crates/preloop-runner-server/src/store.rs @@ -71,7 +71,7 @@ pub(crate) struct InstrumentedStore { } impl InstrumentedStore { - pub(crate) fn new( + fn new( inner: Arc, observability: preloop_observability::Observability, backend: &str, @@ -90,60 +90,55 @@ impl InstrumentedStore { ) -> Arc { Arc::new(Self::new(inner, observability, backend)) } -} -#[async_trait] -impl Store for InstrumentedStore { - async fn load_into(&self, inner: &mut InnerState) -> anyhow::Result<()> { - let start = Instant::now(); - let res = self.inner.load_into(inner).await; - let outcome = if res.is_ok() { "ok" } else { "error" }; + /// Time one delegated call, record duration and outcome, return the + /// original result unchanged. Centralizing this means a newly added + /// `Store` method cannot silently skip instrumentation. + fn record( + &self, + operation: &'static str, + start: Instant, + result: anyhow::Result, + ) -> anyhow::Result { + let outcome = if result.is_ok() { "ok" } else { "error" }; self.observability.metrics().store.observe( &self.backend, - "load_into", + operation, outcome, start.elapsed(), ); - res + result + } +} + +#[async_trait] +impl Store for InstrumentedStore { + async fn load_into(&self, inner: &mut InnerState) -> anyhow::Result<()> { + let start = Instant::now(); + self.record("load_into", start, self.inner.load_into(inner).await) } async fn store_inner(&self, snapshot: &StoreSnapshot) -> anyhow::Result<()> { let start = Instant::now(); - let res = self.inner.store_inner(snapshot).await; - let outcome = if res.is_ok() { "ok" } else { "error" }; - self.observability.metrics().store.observe( - &self.backend, - "store_inner", - outcome, - start.elapsed(), - ); - res + self.record("store_inner", start, self.inner.store_inner(snapshot).await) } async fn store_meta_only(&self, meta: &MetaSnapshot) -> anyhow::Result<()> { let start = Instant::now(); - let res = self.inner.store_meta_only(meta).await; - let outcome = if res.is_ok() { "ok" } else { "error" }; - self.observability.metrics().store.observe( - &self.backend, + self.record( "store_meta_only", - outcome, - start.elapsed(), - ); - res + start, + self.inner.store_meta_only(meta).await, + ) } async fn store_run_event(&self, projection: RunProjection) -> anyhow::Result<()> { let start = Instant::now(); - let res = self.inner.store_run_event(projection).await; - let outcome = if res.is_ok() { "ok" } else { "error" }; - self.observability.metrics().store.observe( - &self.backend, + self.record( "store_run_event", - outcome, - start.elapsed(), - ); - res + start, + self.inner.store_run_event(projection).await, + ) } async fn store_workflow_run_counter( @@ -152,18 +147,13 @@ impl Store for InstrumentedStore { next_run_number: u64, ) -> anyhow::Result<()> { let start = Instant::now(); - let res = self - .inner - .store_workflow_run_counter(workflow_path, next_run_number) - .await; - let outcome = if res.is_ok() { "ok" } else { "error" }; - self.observability.metrics().store.observe( - &self.backend, + self.record( "store_workflow_run_counter", - outcome, - start.elapsed(), - ); - res + start, + self.inner + .store_workflow_run_counter(workflow_path, next_run_number) + .await, + ) } async fn store_log_chunk( @@ -175,31 +165,18 @@ impl Store for InstrumentedStore { line_count: i64, ) -> anyhow::Result<()> { let start = Instant::now(); - let res = self - .inner - .store_log_chunk(key, chunk_index, payload, byte_count, line_count) - .await; - let outcome = if res.is_ok() { "ok" } else { "error" }; - self.observability.metrics().store.observe( - &self.backend, + self.record( "store_log_chunk", - outcome, - start.elapsed(), - ); - res + start, + self.inner + .store_log_chunk(key, chunk_index, payload, byte_count, line_count) + .await, + ) } async fn append_event(&self, event: &NdjsonEvent) -> anyhow::Result<()> { let start = Instant::now(); - let res = self.inner.append_event(event).await; - let outcome = if res.is_ok() { "ok" } else { "error" }; - self.observability.metrics().store.observe( - &self.backend, - "append_event", - outcome, - start.elapsed(), - ); - res + self.record("append_event", start, self.inner.append_event(event).await) } } 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, +};