diff --git a/.repository-projection.json b/.repository-projection.json index 245c3f4..36e2b12 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "borg", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "296fa1359ec1145ea6d30631ec2ee02e38d9aa2e", + "sourceSha": "54af7346321ae5f3a1aeb1ae703bd34e80c204ec", "destinationRepository": "dx-corp/borg", - "priorProjectedBase": "1c7af61c966a6ab9a7b2d566b3c1c2a65c12de4c", + "priorProjectedBase": "ec3cb38cabf0371e8e514cab74a1d0b494c02395", "definitionDigest": "78ba8d1b5794c261ff0afba6d99abc10b9297ea0dd94be4397fb1cbf63cf0c6e", "toolDigest": "61f64178999b764105e6e5c5bc7c5971c164909e", - "contentDigest": "d1f9aa12cfbdad0dbefe3fd520a413d63e09fb36cc7ac977092a62698df564e0", + "contentDigest": "cd2134012915159bfbe895ad883939b40f10ccea2675fd6869366f7c13de4979", "publicationEligible": true } diff --git a/docs/operations.md b/docs/operations.md index 2fbca43..c3e72c6 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -10,6 +10,7 @@ Configuration, scheduling, drone administration, and protocol details for - [Config files](#config-files) - [Queen](#queen) - [Job telemetry](#job-telemetry) + - [Per-job host diagnostics](#per-job-host-diagnostics) - [Drone bids and demand pressure](#drone-bids-and-demand-pressure) - [Wake-on-LAN](#wake-on-lan) - [Build environment contract](#build-environment-contract) @@ -236,6 +237,66 @@ their meaning. Keep it consistent across drones: `duration_secs`. A retried result may be sent again; the queen keeps the latest. +### Per-job host diagnostics + +A job that fails on an overloaded drone looks like a code failure unless +someone compares runners by hand. The drone samples its host just before the +runner starts and again when it exits, and sends both samples with the lease +result (`host: {start, end}` next to `exit_code`). Each sample has: + +| Field | Source | +| --- | --- | +| `at`, `cpus` | Unix seconds; CPUs the drone may use. | +| `load` | `/proc/loadavg` 1, 5 and 15 minute averages. Inside an LXC, lxcfs computes this and it over-reports, so use the pressure and busy fields instead. | +| `cpu_psi`, `memory_psi`, `io_psi` | `/proc/pressure/*`, `some`/`full` `avg10`/`avg60` in percent. | +| `cgroup_cpu_psi`, `cgroup_memory_psi` | `/sys/fs/cgroup/*.pressure`: the cgroup-namespace root, which is the whole container in an LXC. | +| `host_cpu_usage_usec`, `host_cpu_busy_cores` | `usage_usec` of `/sys/fs/cgroup/cpu.stat`, and the cores busy since the drone's previous reading (heartbeats feed it, so at most one heartbeat old). | +| `drone_cpu_usage_usec`, `drone_cpu_throttled_usec` | `cpu.stat` of the drone's own cgroup (`/proc/self/cgroup`). Jobs run inside it. | +| `mem_available_mib`, `work_free_mib`, `cache_free_mib` | `MemAvailable`; `df` of `work_root` and `cache_root`. | +| `active_leases` | Leases the drone held at that moment, this one included. | + +Every field is optional, and a field the drone cannot read is left out. A drone +that predates this sends no `host`, and the row still records which drone ran +the job. + +The queen stores one row per lease in `lease_hosts`. It adds the GitHub run id, +job id and job name when its queue poll sees the job on a runner it minted +(the same match as [GitHub job feedback](#github-job-feedback)). Rows are kept +for 7 days and at most 50,000 rows, oldest first. Two read endpoints are +unauthenticated like `/status`: + +- `GET /jobs/{job_id}` returns the lease that ran the job, or 404. +- `GET /runs/{run_id}` returns `{run_id, jobs: [...]}`, oldest lease first. + +```json +{ + "lease_id": "a1b2c3", "drone": "shared-box", "class": "heavy", "scope": "main", + "repo": "example-org/app", "run_id": 123456789, "job_id": 987654321, "job_name": "ui-e2e (shard 3)", + "leased_at": 1760650000, "finished_at": 1760651800, "duration_secs": 1790, "exit_code": 1, + "start": { "at": 1760650010, "cpus": 16, "load": [77.1, 60.2, 50.3], + "cgroup_cpu_psi": { "some_avg10": 54.9, "some_avg60": 52.4 }, + "host_cpu_busy_cores": 15.2, "active_leases": 2, "…": "…" }, + "end": { "…": "…" }, + "during": { "secs": 1790, "host_cpu_busy_cores": 15.5, "drone_cpu_busy_cores": 3.0, + "drone_cpu_throttled_secs": 0.0 }, + "slow_host": true, + "slow_host_reasons": ["cpu pressure 54.9% at start (threshold 30%)"] +} +``` + +`slow_host` is true when any of these held: CPU pressure (`some avg10`, the +higher of the host and cgroup files) at start at or above the admission soft +threshold (`psi_soft_pm / 10`, or 30% without an `admission` block); memory +pressure at start at or above 10%; load average per CPU at start at or above +1.5, used only when the host has no CPU pressure reading; 90% of the CPUs busy +at start or on average during the job; less than 10 GiB free on the work or +cache volume at start. `during` is computed from the two samples. + +The job id is known only if a queue poll saw the job while it ran or a direct +lookup found it afterwards (see [GitHub job feedback](#github-job-feedback)). +A job that starts and finishes between polls has a row without `job_id`, and +`/jobs/{job_id}` returns 404 for it. + ### Drone bids and demand pressure A drone may add `bids` to its heartbeat: at most one per `(class, scope)` it is @@ -349,6 +410,8 @@ old as unknown, so probe at least that often): | `toolchains` | For each toolchain `rustup toolchain list` shows, `{channel, rustc_version}`: the channel as `rust-toolchain.toml` would name it (`1.95.0`, `stable`, `nightly-2026-01-01`; strip the host triple) and the `release:` line of `rustc + -vV`. `null` if rustup could not be queried. | | `mem_available_mib` | Linux `MemAvailable` from `/proc/meminfo`; macOS (free + inactive + speculative pages) times the page size, in MiB. | | `cpu_psi_some_avg10_pm` | Linux `/proc/pressure/cpu` line `some`, `avg10` times ten, rounded; absent where PSI does not exist. | +| `load_per_cpu_pm` | `/proc/loadavg` 1-minute average divided by the usable CPUs, times 1000. | +| `cpu_busy_pm` | Share of the CPUs busy since the previous heartbeat, from `usage_usec` in `/sys/fs/cgroup/cpu.stat`, in permille. | | `tools` | For each of `git`, `gh`, `node`, `npm`, `pnpm`, `python3`, `docker`, `cargo`, `rustc`, `bazel`, `go`, `uv`, `mise`, `jq`, `cosign`: resolve the executable on the job user's `PATH` (as the job user, with the job environment) and record the first line of ` --version` (`go version` for `go`), trimmed, at most 80 characters. `"missing"` when it is not on `PATH`, `"unknown"` when it is but prints nothing. The map is absent when the probe did not run. | A unit's toolchain (`rust_toolchain`) matches an installed `channel` or @@ -873,23 +936,55 @@ one saturated host can all show free slots. Optional and off unless the queen config has an `admission` block: ```json -"admission": { "psi_soft_pm": 300, "psi_hard_pm": 600, "facts_max_age_secs": 180 } +"admission": { "psi_soft_pm": 300, "psi_light_only_pm": 500, "psi_hard_pm": 600, "facts_max_age_secs": 180 } ``` Each drone reports Linux PSI `cpu some avg10` in permille on every heartbeat (the higher of `/proc/pressure/cpu` and the cgroup file -`/sys/fs/cgroup/cpu.pressure`). At or above `psi_soft_pm` the drone takes no -new pull-request grant; main work is still placed. At or above `psi_hard_pm` it -takes no new grant of any scope. Leases already granted are not touched. A -reading older than `facts_max_age_secs`, or no reading, puts no restriction on -the drone. The starvation guard obeys the same rule: a starving pull-request -lease is not placed on a soft-blocked drone. Requires `0 < psi_soft_pm <= -psi_hard_pm <= 1000` and a positive age. - -`/status` shows `admission: {state, cpu_psi_pm}` per drone (`state` is `open`, -`pr_blocked` or `closed`). -Metrics: `borg_drone_cpu_psi_pm{drone}` and `borg_drone_admission{drone,state}`. -The build environment contract's `resources` check is separate and still works. +`/sys/fs/cgroup/cpu.pressure`). The queen compares it with three thresholds: + +| Pressure | State | New grants | +| --- | --- | --- | +| below `psi_soft_pm` | `open` | all | +| at or above `psi_soft_pm` | `pr_blocked` | no pull-request work | +| at or above `psi_light_only_pm` | `light_only` | no pull-request work and no main medium or heavy work; main light and deploy work still placed | +| at or above `psi_hard_pm` | `closed` | none | + +`psi_light_only_pm` is optional and must lie between the other two. Without +it a drone goes from `pr_blocked` straight to `closed`. Leases already granted +are not touched. A reading older than `facts_max_age_secs`, or no reading at +all, puts no restriction on the drone. The starvation guard obeys the same +rule: a starving pull-request lease is not placed on a soft-blocked drone, and +a starving main heavy lease is not placed on a `light_only` drone. Requires +`0 < psi_soft_pm <= psi_hard_pm <= 1000` and a positive age. + +**Hosts without PSI.** A kernel without PSI (or booted with `psi=0`) reports +no pressure. Such a drone would never be blocked, however loaded. The drone +therefore also reports `load_per_cpu_pm`, the 1-minute load average divided by +its usable CPUs in permille, and `cpu_busy_pm`, the share of CPUs busy since the +previous heartbeat from the root cgroup's `cpu.stat`. When PSI is missing the +queen estimates pressure as `(load - cpus) / load`, the share of runnable tasks +that wait for a CPU (load 93 on 16 CPUs gives 828 permille). The estimate is 0 +when the load is at or below one task per CPU, or when `cpu_busy_pm` is under +900. In that case the CPUs have idle time, so the run queue is I/O wait or an +lxcfs-computed load average, not CPU contention. PSI always wins when present. + +**A null `cpu_psi_pm` usually means an old drone binary.** Drones built before +CPU-pressure admission send no pressure at all, so `/status` shows +`cpu_psi_pm: null` and the drone is never blocked. Before trusting admission, +check that every drone runs a current binary (see +[Updating a drone](#updating-a-drone)). Inside an LXC, `/proc/loadavg` is +lxcfs's estimate and over-reports badly, while the container's +`/sys/fs/cgroup/cpu.pressure` is kernel-computed. An LXC drone with a current +binary therefore admits on real pressure. + +`/status` shows `admission: {state, cpu_psi_pm, cpu_load_pm}` per drone +(`state` is `open`, `pr_blocked`, `light_only` or `closed`; `cpu_load_pm` is the +load estimate, used only when `cpu_psi_pm` is null). Placement diagnostics name +the rejection `pr_pressure`, `light_only_pressure` or `admission_closed`. +Metrics: `borg_drone_cpu_psi_pm{drone}`, `borg_drone_cpu_load_pm{drone}` and +`borg_drone_admission{drone,state}`. The build environment contract's +`resources` check is separate and still works. ### Standby runners and deploy fail-open diff --git a/src/allocator.rs b/src/allocator.rs index 964bf7e..5900680 100644 --- a/src/allocator.rs +++ b/src/allocator.rs @@ -85,12 +85,34 @@ pub struct DroneState { /// or stale. Only read when [`AllocatorInput::admission`] is set. #[serde(default)] pub cpu_psi_pm: Option, + /// CPU pressure estimated from the load average, for a drone without PSI + /// (see [`load_pressure_pm`]). Admission uses it only when `cpu_psi_pm` + /// is `None`. + #[serde(default)] + pub cpu_load_pm: Option, +} + +/// CPU pressure (permille) estimated from the 1-minute load average per CPU +/// (`load_per_cpu_pm`, 1000 = one runnable task per CPU) for hosts without +/// PSI. With `L` runnable tasks on `N` CPUs, `(L - N) / L` of them wait, which +/// approximates PSI `cpu some`. When the root cgroup shows the CPUs under 90% +/// busy (`cpu_busy_pm`), the run queue is not CPU contention (I/O wait, or an +/// lxcfs load average) and the estimate is 0. +pub fn load_pressure_pm(load_per_cpu_pm: Option, cpu_busy_pm: Option) -> Option { + let load = u64::from(load_per_cpu_pm?); + if load <= 1000 || cpu_busy_pm.is_some_and(|b| b < 900) { + return Some(0); + } + u16::try_from((load - 1000) * 1000 / load).ok() } /// CPU-pressure thresholds (permille) for admission; see [`AdmissionState`]. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct AdmissionThresholds { pub psi_soft_pm: u16, + /// From this pressure only light main and deploy grants; at or above + /// `psi_hard_pm` it has no effect. + pub psi_light_only_pm: u16, pub psi_hard_pm: u16, } @@ -102,44 +124,66 @@ pub enum AdmissionState { Open, /// At or above soft: pull-request grants are shed, everything else placed. PrBlocked, + /// At or above `psi_light_only_pm`: pull-request grants are shed, and main + /// work only of the light class; deploy work is still placed. + LightOnly, /// At or above hard: no new grant of any scope. Closed, } impl AdmissionState { - pub const ALL: [AdmissionState; 3] = [Self::Open, Self::PrBlocked, Self::Closed]; + pub const ALL: [AdmissionState; 4] = + [Self::Open, Self::PrBlocked, Self::LightOnly, Self::Closed]; pub fn as_str(self) -> &'static str { match self { Self::Open => "open", Self::PrBlocked => "pr_blocked", + Self::LightOnly => "light_only", Self::Closed => "closed", } } - pub fn of(psi: Option, t: Option) -> Self { - match (psi, t) { + /// The state for a drone's pressure: its PSI reading, or the load + /// estimate when it has no PSI. + pub fn of(pressure: Option, t: Option) -> Self { + match (pressure, t) { (Some(p), Some(t)) if p >= t.psi_hard_pm => Self::Closed, + (Some(p), Some(t)) if p >= t.psi_light_only_pm => Self::LightOnly, (Some(p), Some(t)) if p >= t.psi_soft_pm => Self::PrBlocked, _ => Self::Open, } } - fn admits(self, scope: Scope) -> bool { + pub fn admits(self, class: Class, scope: Scope) -> bool { match self { Self::Open => true, Self::PrBlocked => scope != Scope::Pr, + Self::LightOnly => { + scope == Scope::Deploy || (scope == Scope::Main && class == Class::Light) + } Self::Closed => false, } } } +impl DroneState { + /// The pressure admission uses: PSI when measured, else the load estimate. + pub fn admission_pressure_pm(&self) -> Option { + self.cpu_psi_pm.or(self.cpu_load_pm) + } +} + /// One drone's admission state this pass. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct DroneAdmission { pub drone_id: String, pub state: AdmissionState, pub cpu_psi_pm: Option, + /// The load-average estimate; drives `state` only when `cpu_psi_pm` is + /// `None`. + #[serde(default)] + pub cpu_load_pm: Option, } impl DroneState { @@ -159,6 +203,7 @@ impl DroneState { profiles: Vec::new(), blocked_classes: Vec::new(), cpu_psi_pm: None, + cpu_load_pm: None, trust: Trust::Shared, } } @@ -299,6 +344,8 @@ pub enum Rejection { NotAccepting, AdmissionClosed, PrPressure, + /// CPU pressure at or above `psi_light_only_pm`: no main medium or heavy. + LightOnlyPressure, Trust, Environment, Docker, @@ -511,7 +558,12 @@ impl Free<'_> { } match self.admission { AdmissionState::Closed => reasons.push(Rejection::AdmissionClosed), - AdmissionState::PrBlocked if scope == Scope::Pr => reasons.push(Rejection::PrPressure), + AdmissionState::PrBlocked | AdmissionState::LightOnly if scope == Scope::Pr => { + reasons.push(Rejection::PrPressure); + } + AdmissionState::LightOnly if !self.admission.admits(class, scope) => { + reasons.push(Rejection::LightOnlyPressure); + } _ => {} } if !self.drone.trust.admits(scope) { @@ -732,7 +784,7 @@ fn allocate_inner( vcpu: drone.vcpu, leases_left: drone.max_leases, in_flight: 0, - admission: AdmissionState::of(drone.cpu_psi_pm, input.admission), + admission: AdmissionState::of(drone.admission_pressure_pm(), input.admission), granted: BTreeMap::new(), }); } @@ -865,6 +917,7 @@ fn allocate_inner( drone_id: (*id).to_owned(), state: f.admission, cpu_psi_pm: f.drone.cpu_psi_pm, + cpu_load_pm: f.drone.cpu_load_pm, }) .collect(), promote, @@ -905,7 +958,7 @@ fn allocate_inner( !reserved.contains_key(**id) && f.drone.trust.admits(scope) && f.could_ever_fit(spec) - && f.admission.admits(scope) + && f.admission.admits(class, scope) && !f.drone.blocked_classes.contains(&class) && f.offer( class, @@ -988,6 +1041,7 @@ mod tests { profiles: vec![], blocked_classes: vec![], cpu_psi_pm: None, + cpu_load_pm: None, trust: Trust::Shared, } } @@ -1070,6 +1124,94 @@ mod tests { } } + #[test] + fn load_estimate_approximates_waiting_share_and_ignores_idle_cpus() { + assert_eq!(load_pressure_pm(None, Some(1000)), None); + assert_eq!(load_pressure_pm(Some(800), None), Some(0)); + assert_eq!(load_pressure_pm(Some(1000), Some(1000)), Some(0)); + // Load 93 on 16 CPUs, CPUs saturated: ~83% of runnable tasks wait. + assert_eq!(load_pressure_pm(Some(5838), Some(990)), Some(828)); + assert_eq!(load_pressure_pm(Some(2000), None), Some(500)); + // Same run queue with idle CPU time is not CPU contention. + assert_eq!(load_pressure_pm(Some(5838), Some(550)), Some(0)); + } + + #[test] + fn light_only_sheds_pr_and_main_medium_heavy_but_keeps_light_main_and_deploy() { + let t = AdmissionThresholds { + psi_soft_pm: 300, + psi_light_only_pm: 500, + psi_hard_pm: 600, + }; + let of = |p| AdmissionState::of(Some(p), Some(t)); + assert_eq!(of(299), AdmissionState::Open); + assert_eq!(of(300), AdmissionState::PrBlocked); + assert_eq!(of(540), AdmissionState::LightOnly); + assert_eq!(of(600), AdmissionState::Closed); + let s = AdmissionState::LightOnly; + assert!(s.admits(Class::Light, Scope::Main)); + assert!(s.admits(Class::Heavy, Scope::Deploy)); + assert!(!s.admits(Class::Medium, Scope::Main)); + assert!(!s.admits(Class::Heavy, Scope::Main)); + assert!(!s.admits(Class::Light, Scope::Pr)); + // Unset light-only tier (== hard): PR-blocked straight to closed. + let old = AdmissionThresholds { + psi_light_only_pm: 600, + ..t + }; + assert_eq!( + AdmissionState::of(Some(540), Some(old)), + AdmissionState::PrBlocked + ); + + // A loaded drone without PSI: its load estimate drives admission. + let loaded = DroneState { + cpu_load_pm: Some(828), + ..host("loaded") + }; + let mut i = input( + &[ + (Class::Heavy, Scope::Main, 1), + (Class::Medium, Scope::Main, 1), + (Class::Light, Scope::Main, 1), + ], + vec![loaded.clone()], + ); + i.admission = Some(t); + let a = allocate(&i); + assert_eq!(a.admission[0].state, AdmissionState::Closed); + assert!(a.grants.is_empty()); + + let busy = DroneState { + cpu_psi_pm: Some(540), + ..loaded + }; + i.drones = vec![busy]; + let a = allocate(&i); + assert_eq!( + a.admission[0].state, + AdmissionState::LightOnly, + "PSI wins over the load estimate" + ); + assert_eq!(count(&a, Class::Light, Scope::Main), 1); + assert_eq!(count(&a, Class::Medium, Scope::Main), 0); + assert_eq!(count(&a, Class::Heavy, Scope::Main), 0); + let rejected: Vec<_> = a + .decisions + .iter() + .filter(|d| d.class == Class::Heavy) + .flat_map(|d| { + d.candidates + .iter() + .flat_map(|c| c.rejection_reasons.clone()) + }) + .collect(); + assert!( + rejected.contains(&Rejection::LightOnlyPressure), + "{rejected:?}" + ); + } + #[test] fn free_room_and_slots_subtract_what_the_drone_already_holds() { let classes = Classes::default(); @@ -1610,6 +1752,7 @@ mod tests { profiles: vec![], blocked_classes: vec![], cpu_psi_pm: None, + cpu_load_pm: None, trust: Trust::Shared, }, DroneState { @@ -1623,6 +1766,7 @@ mod tests { profiles: vec![], blocked_classes: vec![], cpu_psi_pm: None, + cpu_load_pm: None, trust: Trust::Shared, }, ]; diff --git a/src/api.rs b/src/api.rs index 35c170d..701d667 100644 --- a/src/api.rs +++ b/src/api.rs @@ -1,7 +1,8 @@ //! HTTP surface of the queen. //! -//! * `GET /status` and `GET /metrics` are read-only and unauthenticated; bind -//! them to a private interface. +//! * `GET /status`, `GET /metrics`, `GET /report`, `GET /jobs/{job_id}` and +//! `GET /runs/{run_id}` are read-only and unauthenticated; bind them to a +//! private interface. //! * `POST /v1/heartbeat` and `POST /v1/leases/{id}/result` are the drone API, //! authenticated with `Authorization: Bearer `. //! @@ -29,6 +30,7 @@ use crate::enroll::hash_token; use crate::envcontract::{CheckState, EnvFacts, Need, Verdict, blocked_classes, evaluate}; use crate::health::{DroneHealth, HealthState}; use crate::homeostasis::KeyReport; +use crate::hostdiag::{JobHostView, LeaseHostReport}; use crate::probe::Capacity; use crate::queen::RolloutView; use crate::queen::{Queen, QueenError, QueueEntry}; @@ -97,6 +99,7 @@ fn sanitize_facts(f: &EnvFacts, now: i64) -> EnvFacts { }; let mut out = f.clone(); out.observed_at = out.observed_at.min(now); + out.cpu_busy_pm = out.cpu_busy_pm.map(|b| b.min(1000)); for fact in out.capabilities.values_mut() { fact.observed_at = fact.observed_at.min(now); } @@ -272,6 +275,10 @@ pub struct ResultRequest { /// The drone stopped the job because the queen preempted the lease. #[serde(default)] pub preempted: bool, + /// Host load when the runner started and when it exited; see + /// [`crate::hostdiag`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub host: Option, } #[derive(Debug)] @@ -327,6 +334,8 @@ pub fn router(queen: Arc) -> Router { .route("/metrics", get(metrics)) .route("/report", get(report)) .route("/explain", get(explain)) + .route("/jobs/{job_id}", get(job_host)) + .route("/runs/{run_id}", get(run_hosts)) .route("/v1/enroll", post(tailnet_enroll)) .route("/v1/heartbeat", post(heartbeat)) .route("/v1/leases/{id}/result", post(lease_result)) @@ -627,6 +636,15 @@ async fn lease_result( .finish_lease(&id, &drone, &result, now) .await? { + if let Some(lease) = queen.store().lease(&id).await? + && let Err(e) = queen + .store() + .record_lease_host(&lease, &result, req.host.as_ref(), now) + .await + { + // Diagnostics never fail a result. + tracing::warn!("lease {id}: host diagnostics not stored: {e}"); + } if let Some(t) = req.telemetry { match t.sanitized() { Some(t) => { @@ -791,6 +809,9 @@ pub struct CapacityStatus { pub struct DroneAdmissionView { pub state: AdmissionState, pub cpu_psi_pm: Option, + /// Pressure estimated from the load average; `state` uses it only when + /// `cpu_psi_pm` is null. + pub cpu_load_pm: Option, } /// One drone's build environment contract. @@ -1107,6 +1128,7 @@ pub async fn build_status(queen: &Queen, now: i64) -> Result .map(|a| DroneAdmissionView { state: a.state, cpu_psi_pm: a.cpu_psi_pm, + cpu_load_pm: a.cpu_load_pm, }), contract: cfg.env_contract.enabled.then(|| { let verdict = |class| { @@ -1311,6 +1333,54 @@ async fn report( ))) } +/// The CPU pressure (percent) above which a job's host counts as slow: the +/// admission soft threshold when admission is on. +fn slow_cpu_psi_pct(queen: &Queen) -> f64 { + queen + .config() + .admission + .as_ref() + .map_or(crate::hostdiag::SLOW_CPU_PSI_PCT, |a| { + f64::from(a.psi_soft_pm) / 10.0 + }) +} + +/// `GET /jobs/{job_id}`: the drone that ran GitHub job `job_id` and its host +/// load at start and end. 404 when the queen never saw that job on a runner it +/// minted, or the row aged out. Unauthenticated like `/status`. +async fn job_host( + State(queen): State>, + Path(job_id): Path, +) -> Result, ApiError> { + queen + .store() + .job_host(job_id, slow_cpu_psi_pct(&queen)) + .await? + .map(Json) + .ok_or(ApiError::NotFound) +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct RunHosts { + pub run_id: u64, + pub jobs: Vec, +} + +/// `GET /runs/{run_id}`: every job of a workflow run that ran on a borg +/// runner, oldest lease first. An unknown run has an empty `jobs` list. +async fn run_hosts( + State(queen): State>, + Path(run_id): Path, +) -> Result, ApiError> { + Ok(Json(RunHosts { + run_id, + jobs: queen + .store() + .run_hosts(run_id, slow_cpu_psi_pct(&queen)) + .await?, + })) +} + async fn status(State(queen): State>) -> Result, ApiError> { Ok(Json( build_status(&queen, OffsetDateTime::now_utc().unix_timestamp()).await?, @@ -1568,9 +1638,24 @@ pub fn render_metrics(status: &Status) -> String { }) .collect(), ); + gauge( + "borg_drone_cpu_load_pm", + "CPU pressure (permille) estimated from the drone's load average; admission uses it when PSI is absent.", + status + .drones + .iter() + .filter_map(|d| { + let load = d.admission.as_ref()?.cpu_load_pm?; + Some(( + format!("drone=\"{}\"", label_value(&d.name)), + f64::from(load), + )) + }) + .collect(), + ); gauge( "borg_drone_admission", - "1 for the drone's CPU-pressure admission state (open, pr_blocked, closed).", + "1 for the drone's CPU-pressure admission state (open, pr_blocked, light_only, closed).", status .drones .iter() diff --git a/src/config.rs b/src/config.rs index 5033ab2..dae08b5 100644 --- a/src/config.rs +++ b/src/config.rs @@ -202,6 +202,12 @@ pub struct AdmissionSettings { /// From this `cpu some avg10` (permille) a drone takes no new /// pull-request grants. pub psi_soft_pm: u16, + /// From this pressure a drone takes, besides deploy work, only light main + /// grants: no medium or heavy main work. Optional; between `psi_soft_pm` + /// and `psi_hard_pm`. Unset, the drone goes from pull-request blocked + /// straight to closed. + #[serde(default)] + pub psi_light_only_pm: Option, /// From this pressure a drone takes no new grants at all. pub psi_hard_pm: u16, /// A drone's reported pressure older than this counts as unknown. @@ -212,6 +218,7 @@ impl AdmissionSettings { pub fn thresholds(&self) -> AdmissionThresholds { AdmissionThresholds { psi_soft_pm: self.psi_soft_pm, + psi_light_only_pm: self.psi_light_only_pm.unwrap_or(self.psi_hard_pm), psi_hard_pm: self.psi_hard_pm, } } @@ -220,6 +227,13 @@ impl AdmissionSettings { if self.psi_soft_pm == 0 || self.psi_soft_pm > self.psi_hard_pm || self.psi_hard_pm > 1000 { return Err("admission needs 0 < psi_soft_pm <= psi_hard_pm <= 1000".to_owned()); } + if let Some(l) = self.psi_light_only_pm + && !(self.psi_soft_pm..=self.psi_hard_pm).contains(&l) + { + return Err( + "admission needs psi_soft_pm <= psi_light_only_pm <= psi_hard_pm".to_owned(), + ); + } if self.facts_max_age_secs <= 0 { return Err("admission.facts_max_age_secs must be positive".to_owned()); } @@ -665,6 +679,12 @@ mod tests { let ok = with(r#"{"psi_soft_pm":300,"psi_hard_pm":600,"facts_max_age_secs":180}"#).unwrap(); let t = ok.admission.unwrap().thresholds(); assert_eq!((t.psi_soft_pm, t.psi_hard_pm), (300, 600)); + assert_eq!(t.psi_light_only_pm, 600, "unset means no light-only tier"); + let t = with(r#"{"psi_soft_pm":300,"psi_light_only_pm":500,"psi_hard_pm":600,"facts_max_age_secs":180}"#) + .unwrap() + .admission + .unwrap(); + assert_eq!(t.thresholds().psi_light_only_pm, 500); for bad in [ r#"{"psi_soft_pm":700,"psi_hard_pm":600,"facts_max_age_secs":180}"#, r#"{"psi_soft_pm":0,"psi_hard_pm":600,"facts_max_age_secs":180}"#, @@ -672,6 +692,8 @@ mod tests { r#"{"psi_soft_pm":300,"psi_hard_pm":600,"facts_max_age_secs":0}"#, r#"{"psi_soft_pm":300,"psi_hard_pm":600,"facts_max_age_secs":180,"nope":1}"#, r#"{"psi_soft_pm":300,"psi_hard_pm":600}"#, + r#"{"psi_soft_pm":300,"psi_light_only_pm":200,"psi_hard_pm":600,"facts_max_age_secs":180}"#, + r#"{"psi_soft_pm":300,"psi_light_only_pm":700,"psi_hard_pm":600,"facts_max_age_secs":180}"#, ] { assert!(with(bad).is_err(), "{bad}"); } diff --git a/src/drone/mod.rs b/src/drone/mod.rs index 897f581..20f4c1a 100644 --- a/src/drone/mod.rs +++ b/src/drone/mod.rs @@ -96,6 +96,8 @@ pub struct Drone { /// Last Queen floor, retained while the Queen is unreachable. disk_floor: AtomicU64, capability_cache: capabilities::CapabilityCache, + /// Busy-cores rate of the host, fed by heartbeats and lease samples. + cpu_meter: Arc, } impl Drone { @@ -137,6 +139,7 @@ impl Drone { tools_cache: probe::ToolsCache::default(), disk_floor: AtomicU64::new(crate::envcontract::DEFAULT_MIN_DISK_FREE_BYTES), capability_cache: capabilities::CapabilityCache::default(), + cpu_meter: Arc::default(), client: Arc::new(client), host, exec, @@ -218,14 +221,26 @@ impl Drone { status: 0, })?; let host = Arc::clone(&self.host); + let meter = Arc::clone(&self.cpu_meter); let now = time::OffsetDateTime::now_utc().unix_timestamp(); - let mut env_facts = - tokio::task::spawn_blocking(move || probe::env_facts(host.as_ref(), now)) - .await - .map_err(|e| ClientError::Status { - path: format!("probe: {e}"), - status: 0, - })?; + let mut env_facts = tokio::task::spawn_blocking(move || { + let now_ms = + i64::try_from(time::OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000) + .unwrap_or(i64::MAX); + let busy = crate::hostdiag::observe_cpu(host.as_ref(), &meter, now_ms); + let mut facts = probe::env_facts(host.as_ref(), now); + facts.cpu_busy_pm = busy.zip(host.available_parallelism()).map(|(b, cpus)| { + (b / f64::from(cpus.max(1)) * 1000.0) + .round() + .clamp(0.0, 1000.0) as u16 + }); + facts + }) + .await + .map_err(|e| ClientError::Status { + path: format!("probe: {e}"), + status: 0, + })?; env_facts.cache_dir_free_bytes = cache_free; let accepting = accepting && !crate::envcontract::disk_admission( @@ -386,6 +401,8 @@ impl Drone { }; let ctx = LeaseContext { cfg: Arc::clone(&self.cfg), + host: Arc::clone(&self.host), + cpu_meter: Arc::clone(&self.cpu_meter), client: Arc::clone(&self.client), exec: Arc::clone(&self.exec), job_exec: Arc::clone(&self.job_exec), @@ -395,7 +412,7 @@ impl Drone { proc_root: Arc::clone(&self.proc_root), caches: Arc::clone(&self.caches), slot, - _held: held, + held, }; tasks.spawn(handle_lease(ctx, grant)); true @@ -557,6 +574,8 @@ impl Drone { struct LeaseContext { cfg: Arc, + host: Arc, + cpu_meter: Arc, client: Arc, exec: Arc, job_exec: Arc, @@ -565,9 +584,39 @@ struct LeaseContext { preempt: tokio::sync::watch::Receiver, proc_root: Arc, caches: Arc, - /// Dropped before `_held`, so the slot is free whenever the id is. + /// Dropped before `held`, so the slot is free whenever the id is. slot: SlotGuard, - _held: HeldGuard, + held: HeldGuard, +} + +/// Sample the host for this lease's diagnostics. Never fails the lease. +async fn sample_host(ctx: &LeaseContext) -> Option { + let (host, cfg, meter) = ( + Arc::clone(&ctx.host), + Arc::clone(&ctx.cfg), + Arc::clone(&ctx.cpu_meter), + ); + let active = ctx + .held + .held + .lock() + .map(|h| u32::try_from(h.len()).unwrap_or(u32::MAX)) + .unwrap_or(0); + tokio::task::spawn_blocking(move || { + let now_ms = + i64::try_from(time::OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000) + .unwrap_or(i64::MAX); + crate::hostdiag::sample( + host.as_ref(), + &cfg.work_root, + &cfg.cache_root, + active, + &meter, + now_ms, + ) + }) + .await + .ok() } /// Run one lease, clean up unconditionally, report the result. @@ -593,6 +642,7 @@ async fn handle_lease(ctx: LeaseContext, grant: LeaseGrant) { job_user: ctx.cfg.job_user.as_ref(), proc_root: &ctx.proc_root, }; + let host_start = sample_host(&ctx).await; let outcome = run_lease( &spec, ctx.job_exec.as_ref(), @@ -601,6 +651,11 @@ async fn handle_lease(ctx: LeaseContext, grant: LeaseGrant) { ctx.preempt.clone(), ) .await; + let host_end = sample_host(&ctx).await; + let host = Some(crate::hostdiag::LeaseHostReport { + start: host_start, + end: host_end, + }); // Runs whether the job passed, failed, timed out, was aborted or never started. let (work_root, id, slot_name, docker, exec, fs_ops) = ( @@ -645,6 +700,7 @@ async fn handle_lease(ctx: LeaseContext, grant: LeaseGrant) { peak_rss_mib: o.peak_rss_mib, telemetry: None, preempted: o.preempted, + host, }, Err(e) => { tracing::warn!("lease {} could not run: {e}", grant.lease_id); @@ -654,6 +710,7 @@ async fn handle_lease(ctx: LeaseContext, grant: LeaseGrant) { peak_rss_mib: None, telemetry: None, preempted: false, + host, } } }; diff --git a/src/envcontract.rs b/src/envcontract.rs index a258c5c..06442f5 100644 --- a/src/envcontract.rs +++ b/src/envcontract.rs @@ -200,6 +200,14 @@ pub struct EnvFacts { /// where PSI does not exist. #[serde(default)] pub cpu_psi_some_avg10_pm: Option, + /// `/proc/loadavg` 1-minute average per CPU, in permille (1000 is one + /// runnable task per CPU). Admission falls back to it without PSI. + #[serde(default)] + pub load_per_cpu_pm: Option, + /// Share of the host's CPUs busy since the previous heartbeat, from the + /// root cgroup's `cpu.stat`, in permille. + #[serde(default)] + pub cpu_busy_pm: Option, /// For each tool in [`PROBE_TOOLS`], the first line of its `--version` /// output, or `"missing"`; `None` when the drone did not measure it. #[serde(default)] @@ -958,6 +966,8 @@ mod tests { }]), mem_available_mib: Some(20_000), cpu_psi_some_avg10_pm: Some(100), + load_per_cpu_pm: None, + cpu_busy_pm: None, tools: Some(tool_map(&[ ("git", "git version 2.45.0"), ("gh", "gh version 2.50.0"), diff --git a/src/hostdiag.rs b/src/hostdiag.rs new file mode 100644 index 0000000..56b9a64 --- /dev/null +++ b/src/hostdiag.rs @@ -0,0 +1,978 @@ +//! Per-lease host diagnostics: how loaded the drone was when a job started +//! and when it ended. +//! +//! The drone samples its host just before it starts the runner and again when +//! the runner exits, and sends both samples with the lease result +//! (`POST /v1/leases/{id}/result`, field `host`). The queen stores one row per +//! lease, adds the GitHub run id and job id when it sees the job on a runner it +//! minted, and serves the rows on `GET /jobs/{job_id}` and `GET /runs/{run_id}`. +//! +//! A slow job on an overloaded drone looks like a code failure unless someone +//! compares hosts by hand. These rows let a triage tool show which drone ran a +//! job and how loaded that drone was. + +use std::path::Path; +use std::sync::Mutex; + +use serde::{Deserialize, Serialize}; +use sqlx::Row; + +use crate::classes::{Class, Scope}; +use crate::probe::{self, Host}; +use crate::store::{LeaseResult, LeaseRow, Store, StoreError}; + +/// Rows are kept this long after the lease was granted. +pub const KEEP_SECS: i64 = 7 * 86_400; +/// At most this many rows are kept; the oldest go first. +pub const MAX_ROWS: i64 = 50_000; +/// CPU pressure (`some avg10`, percent) at job start that marks a slow host +/// when the queen has no `admission` block. +pub const SLOW_CPU_PSI_PCT: f64 = 30.0; +/// Memory pressure (`some avg10`, percent) at job start that marks a slow host. +pub const SLOW_MEM_PSI_PCT: f64 = 10.0; +/// Load average per CPU that marks a slow host. Only used when the host has no +/// CPU pressure reading, because an LXC's load average is computed by lxcfs and +/// over-reports. +pub const SLOW_LOAD_PER_CPU: f64 = 1.5; +/// Share of the host's CPUs busy that marks a slow host. +pub const SLOW_BUSY_SHARE: f64 = 0.9; +/// Free disk on the work or cache volume that marks a slow host. +pub const SLOW_FREE_DISK_MIB: u64 = 10 * 1024; + +const MAX_NAME_CHARS: usize = 120; + +/// One Linux PSI file (`/proc/pressure/*` or a cgroup `*.pressure`), in percent. +#[derive(Debug, Clone, Copy, Default, PartialEq, Serialize, Deserialize)] +pub struct Psi { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub some_avg10: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub some_avg60: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub full_avg10: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub full_avg60: Option, +} + +impl Psi { + /// Parse `some avg10=1.00 avg60=...` / `full ...` lines. `None` when no + /// field could be read. + pub fn parse(text: &str) -> Option { + let field = |kind: &str, key: &str| -> Option { + text.lines() + .find(|l| l.starts_with(kind))? + .split_whitespace() + .find_map(|f| f.strip_prefix(key))? + .parse::() + .ok() + .filter(|v| v.is_finite()) + .map(|v| v.clamp(0.0, 100.0)) + }; + let psi = Self { + some_avg10: field("some ", "avg10="), + some_avg60: field("some ", "avg60="), + full_avg10: field("full ", "avg10="), + full_avg60: field("full ", "avg60="), + }; + (psi != Self::default()).then_some(psi) + } + + fn sanitized(self) -> Self { + let pct = |v: Option| v.filter(|v| v.is_finite()).map(|v| v.clamp(0.0, 100.0)); + Self { + some_avg10: pct(self.some_avg10), + some_avg60: pct(self.some_avg60), + full_avg10: pct(self.full_avg10), + full_avg60: pct(self.full_avg60), + } + } +} + +/// The drone's host at one moment. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct HostSample { + /// Unix seconds. + pub at: i64, + /// CPUs the drone process may use. + #[serde(default)] + pub cpus: Option, + /// `/proc/loadavg` 1, 5 and 15 minute averages. Inside an LXC this is + /// lxcfs's estimate and over-reports; prefer the pressure and busy fields. + #[serde(default)] + pub load: Option<[f64; 3]>, + /// `/proc/pressure/cpu`. + #[serde(default)] + pub cpu_psi: Option, + /// `/proc/pressure/memory`. + #[serde(default)] + pub memory_psi: Option, + /// `/proc/pressure/io`. + #[serde(default)] + pub io_psi: Option, + /// `/sys/fs/cgroup/cpu.pressure`: the cgroup-namespace root, which is the + /// whole container inside an LXC. + #[serde(default)] + pub cgroup_cpu_psi: Option, + /// `/sys/fs/cgroup/memory.pressure`. + #[serde(default)] + pub cgroup_memory_psi: Option, + /// `usage_usec` of `/sys/fs/cgroup/cpu.stat` (whole host or container). + #[serde(default)] + pub host_cpu_usage_usec: Option, + /// Cores busy in that cgroup, averaged since the drone's previous sample + /// (at most one heartbeat ago). + #[serde(default)] + pub host_cpu_busy_cores: Option, + /// `usage_usec` of the drone's own cgroup; jobs run inside it. + #[serde(default)] + pub drone_cpu_usage_usec: Option, + /// `throttled_usec` of the drone's own cgroup. + #[serde(default)] + pub drone_cpu_throttled_usec: Option, + #[serde(default)] + pub mem_available_mib: Option, + /// Free space on the filesystem holding the drone's `work_root`. + #[serde(default)] + pub work_free_mib: Option, + /// Free space on the filesystem holding the drone's `cache_root`. + #[serde(default)] + pub cache_free_mib: Option, + /// Leases the drone held at this moment, this one included. + #[serde(default)] + pub active_leases: u32, +} + +impl HostSample { + /// Clamp what a drone sent: finite non-negative numbers, no future time. + pub fn sanitized(mut self, now: i64) -> Self { + let num = |v: f64| v.is_finite().then(|| v.max(0.0)); + self.at = self.at.min(now); + self.load = self + .load + .and_then(|l| Some([num(l[0])?, num(l[1])?, num(l[2])?])); + for psi in [ + &mut self.cpu_psi, + &mut self.memory_psi, + &mut self.io_psi, + &mut self.cgroup_cpu_psi, + &mut self.cgroup_memory_psi, + ] { + *psi = psi.map(Psi::sanitized); + } + self.host_cpu_busy_cores = self.host_cpu_busy_cores.and_then(num); + self + } + + /// The higher CPU `some avg10` of the host file and the cgroup file. + pub fn cpu_some_avg10(&self) -> Option { + [self.cpu_psi, self.cgroup_cpu_psi] + .into_iter() + .flatten() + .filter_map(|p| p.some_avg10) + .reduce(f64::max) + } + + fn mem_some_avg10(&self) -> Option { + [self.memory_psi, self.cgroup_memory_psi] + .into_iter() + .flatten() + .filter_map(|p| p.some_avg10) + .reduce(f64::max) + } +} + +/// What a drone sends in its lease result. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct LeaseHostReport { + #[serde(default)] + pub start: Option, + #[serde(default)] + pub end: Option, +} + +/// Turns successive `cpu.stat` readings into a busy-cores rate. +#[derive(Debug, Default)] +pub struct CpuMeter(Mutex>); + +#[derive(Debug, Clone, Copy)] +struct MeterState { + at_ms: i64, + usage_usec: u64, + busy_cores: Option, +} + +/// Readings closer together than this keep the previous rate. +const METER_MIN_GAP_MS: i64 = 2_000; + +impl CpuMeter { + /// Record a reading and return the latest rate. A reading that arrives + /// within two seconds of the stored one returns the stored rate, so a lease + /// that starts right after a heartbeat still gets a measured value. + pub fn observe(&self, at_ms: i64, usage_usec: u64) -> Option { + let mut slot = self.0.lock().ok()?; + let next = match *slot { + Some(prev) if at_ms - prev.at_ms < METER_MIN_GAP_MS && at_ms >= prev.at_ms => { + return prev.busy_cores; + } + Some(prev) if at_ms > prev.at_ms && usage_usec >= prev.usage_usec => MeterState { + at_ms, + usage_usec, + busy_cores: Some( + (usage_usec - prev.usage_usec) as f64 / ((at_ms - prev.at_ms) as f64 * 1000.0), + ), + }, + _ => MeterState { + at_ms, + usage_usec, + busy_cores: None, + }, + }; + *slot = Some(next); + next.busy_cores + } +} + +/// `usage_usec` and `throttled_usec` from a cgroup v2 `cpu.stat`. +fn cpu_stat(host: &dyn Host, dir: &Path) -> (Option, Option) { + let Some(text) = host.read_to_string(&dir.join("cpu.stat")) else { + return (None, None); + }; + let field = |key: &str| { + text.lines() + .find_map(|l| l.strip_prefix(key)?.trim().parse::().ok()) + }; + (field("usage_usec "), field("throttled_usec ")) +} + +/// The drone's own cgroup directory under `/sys/fs/cgroup`, from the cgroup v2 +/// line of `/proc/self/cgroup`. +fn own_cgroup(host: &dyn Host) -> Option { + let text = host.read_to_string(Path::new("/proc/self/cgroup"))?; + let rel = text.lines().find_map(|l| l.strip_prefix("0::"))?.trim(); + let rel = rel.trim_start_matches('/'); + if rel.contains("..") { + return None; + } + Some(Path::new("/sys/fs/cgroup").join(rel)) +} + +fn loadavg(host: &dyn Host) -> Option<[f64; 3]> { + let text = host.read_to_string(Path::new("/proc/loadavg"))?; + let mut it = text + .split_whitespace() + .map(|f| f.parse::().ok().filter(|v| v.is_finite())); + Some([it.next()??, it.next()??, it.next()??]) +} + +/// Feed the meter from the root cgroup and return the busy-cores rate. Called +/// on every heartbeat so a lease start has a recent rate. +pub fn observe_cpu(host: &dyn Host, meter: &CpuMeter, at_ms: i64) -> Option { + let (usage, _) = cpu_stat(host, Path::new("/sys/fs/cgroup")); + meter.observe(at_ms, usage?) +} + +/// Sample the host for a lease start or end. Every field degrades to `None` +/// on its own; this never fails. +pub fn sample( + host: &dyn Host, + work_root: &Path, + cache_root: &Path, + active_leases: u32, + meter: &CpuMeter, + at_ms: i64, +) -> HostSample { + let psi = |path: &str| Psi::parse(&host.read_to_string(Path::new(path))?); + let (host_usage, _) = cpu_stat(host, Path::new("/sys/fs/cgroup")); + let (drone_usage, drone_throttled) = own_cgroup(host) + .map(|dir| cpu_stat(host, &dir)) + .unwrap_or_default(); + HostSample { + at: at_ms.div_euclid(1000), + cpus: host.available_parallelism(), + load: loadavg(host), + cpu_psi: psi("/proc/pressure/cpu"), + memory_psi: psi("/proc/pressure/memory"), + io_psi: psi("/proc/pressure/io"), + cgroup_cpu_psi: psi("/sys/fs/cgroup/cpu.pressure"), + cgroup_memory_psi: psi("/sys/fs/cgroup/memory.pressure"), + host_cpu_usage_usec: host_usage, + host_cpu_busy_cores: host_usage.and_then(|u| meter.observe(at_ms, u)), + drone_cpu_usage_usec: drone_usage, + drone_cpu_throttled_usec: drone_throttled, + mem_available_mib: probe::mem_available_mib(host), + work_free_mib: probe::free_disk_mib_at(host, work_root), + cache_free_mib: probe::free_disk_mib_at(host, cache_root), + active_leases, + } +} + +// ---------------------------------------------------------------- queen side + +/// Averages over the job, from the start and end samples. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct During { + pub secs: i64, + /// Cores busy on the whole host (or container). + #[serde(skip_serializing_if = "Option::is_none")] + pub host_cpu_busy_cores: Option, + /// Cores used by the drone's cgroup: this job and any other lease. + #[serde(skip_serializing_if = "Option::is_none")] + pub drone_cpu_busy_cores: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub drone_cpu_throttled_secs: Option, +} + +fn during(start: &HostSample, end: &HostSample) -> Option { + let secs = end.at - start.at; + if secs <= 0 { + return None; + } + let rate = + |a: Option, b: Option| Some(b?.checked_sub(a?)? as f64 / 1e6 / secs as f64); + Some(During { + secs, + host_cpu_busy_cores: rate(start.host_cpu_usage_usec, end.host_cpu_usage_usec), + drone_cpu_busy_cores: rate(start.drone_cpu_usage_usec, end.drone_cpu_usage_usec), + drone_cpu_throttled_secs: end + .drone_cpu_throttled_usec + .zip(start.drone_cpu_throttled_usec) + .and_then(|(b, a)| b.checked_sub(a)) + .map(|d| d as f64 / 1e6), + }) +} + +/// Why a job's host was overloaded, or nothing. `cpu_psi_pct` is the CPU +/// pressure threshold (the queen's admission soft threshold when set). +pub fn slow_host_reasons( + start: Option<&HostSample>, + during: Option<&During>, + cpu_psi_pct: f64, +) -> Vec { + let mut out = Vec::new(); + if let Some(s) = start { + let cpus = s.cpus.filter(|&c| c > 0).map(f64::from); + match s.cpu_some_avg10() { + Some(p) if p >= cpu_psi_pct => { + out.push(format!( + "cpu pressure {p:.1}% at start (threshold {cpu_psi_pct:.0}%)" + )); + } + Some(_) => {} + None => { + if let (Some(l), Some(c)) = (s.load, cpus) + && l[0] / c >= SLOW_LOAD_PER_CPU + { + out.push(format!( + "load {:.1} on {c:.0} cpus at start ({:.1} per cpu, threshold {SLOW_LOAD_PER_CPU})", + l[0], + l[0] / c + )); + } + } + } + if let Some(p) = s.mem_some_avg10() + && p >= SLOW_MEM_PSI_PCT + { + out.push(format!( + "memory pressure {p:.1}% at start (threshold {SLOW_MEM_PSI_PCT:.0}%)" + )); + } + if let (Some(b), Some(c)) = (s.host_cpu_busy_cores, cpus) + && b / c >= SLOW_BUSY_SHARE + { + out.push(format!("{b:.1} of {c:.0} cpus busy at start")); + } + for (name, free) in [("work", s.work_free_mib), ("cache", s.cache_free_mib)] { + if let Some(f) = free + && f < SLOW_FREE_DISK_MIB + { + out.push(format!("{name} volume {f} MiB free at start")); + } + } + if let (Some(d), Some(c)) = (during, cpus) + && let Some(b) = d.host_cpu_busy_cores + && b / c >= SLOW_BUSY_SHARE + { + out.push(format!( + "{b:.1} of {c:.0} cpus busy on average during the job" + )); + } + } + out +} + +/// One lease as `GET /jobs/{id}` and `GET /runs/{id}` show it. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct JobHostView { + pub lease_id: String, + pub drone: String, + pub class: Class, + pub scope: Scope, + pub repo: Option, + pub run_id: Option, + pub job_id: Option, + pub job_name: Option, + /// When the queen granted the lease. + pub leased_at: i64, + /// When the drone reported the result. + pub finished_at: Option, + /// Runner wall time the drone reported. + pub duration_secs: Option, + pub exit_code: Option, + pub start: Option, + pub end: Option, + pub during: Option, + /// The drone was overloaded when the job started or while it ran. + pub slow_host: bool, + pub slow_host_reasons: Vec, +} + +fn to_i64(v: u64) -> i64 { + i64::try_from(v).unwrap_or(i64::MAX) +} + +fn json(v: Option<&T>) -> Result, StoreError> { + v.map(|v| serde_json::to_string(v).map_err(|e| StoreError::Corrupt(e.to_string()))) + .transpose() +} + +fn enum_text(v: &T) -> Result { + match serde_json::to_value(v) { + Ok(serde_json::Value::String(s)) => Ok(s), + _ => Err(StoreError::Corrupt("enum is not a string".to_owned())), + } +} + +fn enum_from Deserialize<'de>>(s: &str) -> Result { + serde_json::from_value(serde_json::Value::String(s.to_owned())) + .map_err(|e| StoreError::Corrupt(e.to_string())) +} + +fn clean(s: &str) -> String { + s.chars() + .filter(|c| !c.is_control()) + .take(MAX_NAME_CHARS) + .collect::() + .trim() + .to_owned() +} + +const SELECT: &str = "SELECT lease_id, drone, class, scope, repo, run_id, job_id, job_name, leased_at, finished_at, exit_code, duration_secs, start_sample, end_sample FROM lease_hosts"; + +impl Store { + /// Record a drone's result and host samples for its lease. + pub async fn record_lease_host( + &self, + lease: &LeaseRow, + result: &LeaseResult, + report: Option<&LeaseHostReport>, + now: i64, + ) -> Result<(), StoreError> { + let start = report + .and_then(|r| r.start.clone()) + .map(|s| s.sanitized(now)); + let end = report.and_then(|r| r.end.clone()).map(|s| s.sanitized(now)); + // sqlx-runtime-query: standalone-buildable SQLite store; the query! macros need a build-time database + sqlx::query( + "INSERT INTO lease_hosts (lease_id, drone, class, scope, leased_at, finished_at, exit_code, duration_secs, start_sample, end_sample) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(lease_id) DO UPDATE SET + finished_at = excluded.finished_at, exit_code = excluded.exit_code, + duration_secs = excluded.duration_secs, + start_sample = COALESCE(excluded.start_sample, start_sample), + end_sample = COALESCE(excluded.end_sample, end_sample)", + ) + .bind(&lease.id) + .bind(&lease.drone) + .bind(enum_text(&lease.class)?) + .bind(enum_text(&lease.scope)?) + .bind(lease.created_at) + .bind(now) + .bind(i64::from(result.exit_code)) + .bind(to_i64(result.duration_secs)) + .bind(json(start.as_ref())?) + .bind(json(end.as_ref())?) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// Attach the GitHub job a lease's runner picked up. + pub async fn bind_lease_host_job( + &self, + lease: &LeaseRow, + repo: &str, + run_id: u64, + job_id: u64, + job_name: &str, + ) -> Result<(), StoreError> { + let name = clean(job_name); + let run_id = (run_id != 0).then(|| to_i64(run_id)); + // sqlx-runtime-query: standalone-buildable SQLite store; the query! macros need a build-time database + sqlx::query( + "INSERT INTO lease_hosts (lease_id, drone, class, scope, leased_at, repo, run_id, job_id, job_name) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(lease_id) DO UPDATE SET + repo = excluded.repo, run_id = COALESCE(excluded.run_id, run_id), + job_id = excluded.job_id, job_name = COALESCE(excluded.job_name, job_name) + WHERE job_id IS NOT excluded.job_id OR run_id IS NOT COALESCE(excluded.run_id, run_id) + OR job_name IS NOT COALESCE(excluded.job_name, job_name)", + ) + .bind(&lease.id) + .bind(&lease.drone) + .bind(enum_text(&lease.class)?) + .bind(enum_text(&lease.scope)?) + .bind(lease.created_at) + .bind(repo) + .bind(run_id) + .bind(to_i64(job_id)) + .bind((!name.is_empty()).then_some(name)) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// The lease that ran GitHub job `job_id` (the newest, if several did). + pub async fn job_host( + &self, + job_id: u64, + cpu_psi_pct: f64, + ) -> Result, StoreError> { + // sqlx-runtime-query: standalone-buildable SQLite store; the query! macros need a build-time database + let row = sqlx::query(&format!( + "{SELECT} WHERE job_id = ? ORDER BY leased_at DESC LIMIT 1" + )) + .bind(to_i64(job_id)) + .fetch_optional(&self.pool) + .await?; + row.as_ref().map(|r| view(r, cpu_psi_pct)).transpose() + } + + /// One lease's row, by lease id. + pub async fn lease_host( + &self, + lease_id: &str, + cpu_psi_pct: f64, + ) -> Result, StoreError> { + // sqlx-runtime-query: standalone-buildable SQLite store; the query! macros need a build-time database + let row = sqlx::query(&format!("{SELECT} WHERE lease_id = ?")) + .bind(lease_id) + .fetch_optional(&self.pool) + .await?; + row.as_ref().map(|r| view(r, cpu_psi_pct)).transpose() + } + + /// Every lease that ran a job of workflow run `run_id`, oldest first. + pub async fn run_hosts( + &self, + run_id: u64, + cpu_psi_pct: f64, + ) -> Result, StoreError> { + // sqlx-runtime-query: standalone-buildable SQLite store; the query! macros need a build-time database + let rows = sqlx::query(&format!( + "{SELECT} WHERE run_id = ? ORDER BY leased_at, lease_id LIMIT 1000" + )) + .bind(to_i64(run_id)) + .fetch_all(&self.pool) + .await?; + rows.iter().map(|r| view(r, cpu_psi_pct)).collect() + } + + /// Drop rows granted before `before`, then all but the newest `max_rows`. + pub async fn prune_lease_hosts(&self, before: i64, max_rows: i64) -> Result { + // sqlx-runtime-query: standalone-buildable SQLite store; the query! macros need a build-time database + let old = sqlx::query("DELETE FROM lease_hosts WHERE leased_at < ?") + .bind(before) + .execute(&self.pool) + .await? + .rows_affected(); + // sqlx-runtime-query: standalone-buildable SQLite store; the query! macros need a build-time database + let over = sqlx::query( + "DELETE FROM lease_hosts WHERE lease_id IN + (SELECT lease_id FROM lease_hosts ORDER BY leased_at DESC, lease_id DESC LIMIT -1 OFFSET ?)", + ) + .bind(max_rows) + .execute(&self.pool) + .await? + .rows_affected(); + Ok(old + over) + } +} + +fn view(r: &sqlx::sqlite::SqliteRow, cpu_psi_pct: f64) -> Result { + let sample = |col: &str| -> Result, StoreError> { + r.get::, _>(col) + .map(|s| serde_json::from_str(&s).map_err(|e| StoreError::Corrupt(e.to_string()))) + .transpose() + }; + let start = sample("start_sample")?; + let end = sample("end_sample")?; + let during = start + .as_ref() + .zip(end.as_ref()) + .and_then(|(s, e)| during(s, e)); + let reasons = slow_host_reasons(start.as_ref(), during.as_ref(), cpu_psi_pct); + let u = |v: Option| v.and_then(|v| u64::try_from(v).ok()); + Ok(JobHostView { + lease_id: r.get("lease_id"), + drone: r.get("drone"), + class: enum_from(&r.get::("class"))?, + scope: enum_from(&r.get::("scope"))?, + repo: r.get("repo"), + run_id: u(r.get("run_id")), + job_id: u(r.get("job_id")), + job_name: r.get("job_name"), + leased_at: r.get("leased_at"), + finished_at: r.get("finished_at"), + duration_secs: u(r.get("duration_secs")), + exit_code: r + .get::, _>("exit_code") + .and_then(|v| i32::try_from(v).ok()), + start, + end, + during, + slow_host: !reasons.is_empty(), + slow_host_reasons: reasons, + }) +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + + use super::*; + use crate::store::LeaseStatus; + + #[derive(Default)] + struct FakeHost { + files: HashMap, + df: HashMap, + } + + impl Host for FakeHost { + fn read_to_string(&self, path: &Path) -> Option { + self.files.get(path.to_str()?).cloned() + } + fn exists(&self, path: &Path) -> bool { + path.to_str().is_some_and(|p| self.files.contains_key(p)) + } + fn run(&self, program: &str, args: &[&str]) -> Option { + (program == "df") + .then(|| self.df.get(*args.last()?).cloned()) + .flatten() + } + fn os(&self) -> &'static str { + "linux" + } + fn arch(&self) -> &'static str { + "x86_64" + } + fn available_parallelism(&self) -> Option { + Some(16) + } + } + + fn df(avail_kib: u64) -> String { + format!("fs blocks used avail cap mount\nfs 999999999 0 {avail_kib} 0% /x") + } + + /// An LXC build box shared with interactive builds, overloaded. + fn lxc_host(root_usage: u64) -> FakeHost { + let files = [ + ("/proc/loadavg", "93.40 78.56 69.31 90/3277 3570264\n".to_owned()), + ( + "/proc/pressure/cpu", + "some avg10=53.99 avg60=52.14 avg300=47.57 total=1\nfull avg10=23.75 avg60=25.01 avg300=21.45 total=1\n".to_owned(), + ), + ( + "/proc/pressure/memory", + "some avg10=1.84 avg60=0.55 avg300=0.77 total=1\nfull avg10=0.00 avg60=0.00 avg300=0.21 total=1\n".to_owned(), + ), + ( + "/sys/fs/cgroup/cpu.pressure", + "some avg10=54.89 avg60=52.37 avg300=47.65 total=1\nfull avg10=23.61 avg60=24.95 avg300=21.46 total=1\n".to_owned(), + ), + ( + "/sys/fs/cgroup/cpu.stat", + format!("usage_usec {root_usage}\nuser_usec 1\nnr_throttled 0\nthrottled_usec 0\n"), + ), + ("/proc/self/cgroup", "0::/system.slice/borg-drone.service\n".to_owned()), + ( + "/sys/fs/cgroup/system.slice/borg-drone.service/cpu.stat", + "usage_usec 27600656603\nthrottled_usec 5\n".to_owned(), + ), + ("/proc/meminfo", "MemTotal: 1 kB\nMemAvailable: 4194304 kB\n".to_owned()), + ]; + FakeHost { + files: files.into_iter().map(|(k, v)| (k.to_owned(), v)).collect(), + df: HashMap::from([ + ("/work".to_owned(), df(30 * 1024 * 1024)), + ("/cache".to_owned(), df(5 * 1024 * 1024)), + ]), + } + } + + #[test] + fn psi_parses_some_and_full_and_rejects_empty() { + let p = Psi::parse("some avg10=53.99 avg60=52.14 avg300=1 total=2\nfull avg10=23.75 avg60=25.01 avg300=21.45 total").unwrap(); + assert_eq!(p.some_avg10, Some(53.99)); + assert_eq!(p.some_avg60, Some(52.14)); + assert_eq!(p.full_avg10, Some(23.75)); + assert_eq!(p.full_avg60, Some(25.01)); + assert_eq!(Psi::parse("nothing here"), None); + assert_eq!( + Psi::parse("some avg10=500 avg60=nan").unwrap().some_avg10, + Some(100.0) + ); + } + + #[test] + fn sample_reads_load_pressure_cgroups_disk_and_meter() { + let meter = CpuMeter::default(); + // A heartbeat ten seconds earlier primes the meter. + assert_eq!(observe_cpu(&lxc_host(1_000_000_000), &meter, 0), None); + // 120 core-seconds in 10 s: 12 cores busy. + let s = sample( + &lxc_host(1_120_000_000), + Path::new("/work"), + Path::new("/cache"), + 2, + &meter, + 10_000, + ); + assert_eq!(s.at, 10); + assert_eq!(s.cpus, Some(16)); + assert_eq!(s.load, Some([93.40, 78.56, 69.31])); + assert_eq!(s.cpu_psi.unwrap().some_avg10, Some(53.99)); + assert_eq!(s.cgroup_cpu_psi.unwrap().some_avg10, Some(54.89)); + assert_eq!(s.cpu_some_avg10(), Some(54.89)); + assert_eq!(s.memory_psi.unwrap().some_avg10, Some(1.84)); + assert_eq!(s.io_psi, None); + assert_eq!(s.host_cpu_usage_usec, Some(1_120_000_000)); + assert_eq!(s.host_cpu_busy_cores, Some(12.0)); + assert_eq!(s.drone_cpu_usage_usec, Some(27_600_656_603)); + assert_eq!(s.drone_cpu_throttled_usec, Some(5)); + assert_eq!(s.mem_available_mib, Some(4096)); + assert_eq!(s.work_free_mib, Some(30 * 1024)); + assert_eq!(s.cache_free_mib, Some(5 * 1024)); + assert_eq!(s.active_leases, 2); + // An empty host degrades every field to None. + let bare = sample( + &FakeHost::default(), + Path::new("/work"), + Path::new("/cache"), + 1, + &CpuMeter::default(), + 5_000, + ); + assert_eq!(bare.load, None); + assert_eq!(bare.cpu_psi, None); + assert_eq!(bare.host_cpu_busy_cores, None); + assert_eq!(bare.drone_cpu_usage_usec, None); + } + + #[test] + fn meter_keeps_the_last_rate_for_close_readings_and_resets_on_counter_reset() { + let m = CpuMeter::default(); + assert_eq!(m.observe(0, 0), None); + assert_eq!(m.observe(10_000, 40_000_000), Some(4.0)); + // A lease starting 500 ms after the heartbeat reuses its rate. + assert_eq!(m.observe(10_500, 40_100_000), Some(4.0)); + assert_eq!(m.observe(20_000, 120_000_000), Some(8.0)); + // Counter went backwards (cgroup recreated): no rate until next reading. + assert_eq!(m.observe(30_000, 5), None); + assert_eq!(m.observe(40_000, 10_000_005), Some(1.0)); + } + + #[test] + fn own_cgroup_rejects_escapes() { + let mut h = FakeHost::default(); + h.files + .insert("/proc/self/cgroup".into(), "0::/../etc\n".into()); + assert_eq!(own_cgroup(&h), None); + h.files.insert("/proc/self/cgroup".into(), "0::/\n".into()); + assert_eq!(own_cgroup(&h), Some(Path::new("/sys/fs/cgroup").into())); + } + + #[test] + fn slow_host_uses_pressure_and_ignores_lxc_load_when_pressure_exists() { + let start = HostSample { + at: 0, + cpus: Some(16), + load: Some([93.4, 78.0, 69.0]), + cgroup_cpu_psi: Some(Psi { + some_avg10: Some(10.0), + ..Psi::default() + }), + ..HostSample::default() + }; + // Pressure below threshold: lxcfs load average alone is not evidence. + assert!(slow_host_reasons(Some(&start), None, 30.0).is_empty()); + let hot = HostSample { + cgroup_cpu_psi: Some(Psi { + some_avg10: Some(54.9), + ..Psi::default() + }), + ..start.clone() + }; + let r = slow_host_reasons(Some(&hot), None, 30.0); + assert_eq!(r, vec!["cpu pressure 54.9% at start (threshold 30%)"]); + // No pressure at all: fall back to load per cpu. + let no_psi = HostSample { + cgroup_cpu_psi: None, + ..start.clone() + }; + let r = slow_host_reasons(Some(&no_psi), None, 30.0); + assert_eq!(r.len(), 1); + assert!(r[0].starts_with("load 93.4 on 16 cpus"), "{r:?}"); + // A quiet burst drone is not slow. + let quiet = HostSample { + at: 0, + cpus: Some(56), + load: Some([20.0, 20.0, 20.0]), + cpu_psi: Some(Psi { + some_avg10: Some(0.5), + ..Psi::default() + }), + host_cpu_busy_cores: Some(20.0), + work_free_mib: Some(500_000), + cache_free_mib: Some(500_000), + ..HostSample::default() + }; + assert!(slow_host_reasons(Some(&quiet), None, 30.0).is_empty()); + let busy = During { + secs: 100, + host_cpu_busy_cores: Some(55.0), + drone_cpu_busy_cores: None, + drone_cpu_throttled_secs: None, + }; + assert_eq!( + slow_host_reasons(Some(&quiet), Some(&busy), 30.0), + vec!["55.0 of 56 cpus busy on average during the job"] + ); + assert!(slow_host_reasons(None, None, 30.0).is_empty()); + } + + #[test] + fn sanitize_drops_non_finite_and_future_times() { + let s = HostSample { + at: 999, + load: Some([f64::NAN, 1.0, 1.0]), + host_cpu_busy_cores: Some(-3.0), + cpu_psi: Some(Psi { + some_avg10: Some(f64::INFINITY), + some_avg60: Some(250.0), + ..Psi::default() + }), + ..HostSample::default() + } + .sanitized(100); + assert_eq!(s.at, 100); + assert_eq!(s.load, None); + assert_eq!(s.host_cpu_busy_cores, Some(0.0)); + assert_eq!(s.cpu_psi.unwrap().some_avg10, None); + assert_eq!(s.cpu_psi.unwrap().some_avg60, Some(100.0)); + } + + fn lease(id: &str, created_at: i64) -> LeaseRow { + LeaseRow { + id: id.into(), + drone: "shared-box".into(), + class: Class::Heavy, + scope: Scope::Main, + labels: vec![], + runner_name: format!("borg-{id}"), + runner_id: 1, + jit_config: None, + state: LeaseStatus::Running, + created_at, + expires_at: created_at + 600, + } + } + + #[tokio::test] + async fn rows_bind_jobs_and_results_in_either_order_and_prune() { + let s = Store::open_in_memory().await.unwrap(); + let (a, b, c) = (lease("a", 100), lease("b", 200), lease("c", 300)); + let result = LeaseResult { + exit_code: 1, + duration_secs: 90, + peak_rss_mib: None, + }; + let start = HostSample { + at: 110, + cpus: Some(16), + cpu_psi: Some(Psi { + some_avg10: Some(54.0), + ..Psi::default() + }), + host_cpu_usage_usec: Some(0), + drone_cpu_usage_usec: Some(0), + ..HostSample::default() + }; + let end = HostSample { + at: 210, + host_cpu_usage_usec: Some(1_500_000_000), + drone_cpu_usage_usec: Some(400_000_000), + ..start.clone() + }; + let report = LeaseHostReport { + start: Some(start), + end: Some(end), + }; + // a: job seen first, result later. + s.bind_lease_host_job(&a, "o/r", 7, 70, "e2e (shard 3)") + .await + .unwrap(); + s.record_lease_host(&a, &result, Some(&report), 220) + .await + .unwrap(); + // b: result first (old drone, no samples), job later. + s.record_lease_host(&b, &result, None, 230).await.unwrap(); + s.bind_lease_host_job(&b, "o/r", 7, 71, "lint") + .await + .unwrap(); + // c: another run. + s.bind_lease_host_job(&c, "o/r", 8, 80, "x").await.unwrap(); + + let v = s.job_host(70, 30.0).await.unwrap().unwrap(); + assert_eq!(v.drone, "shared-box"); + assert_eq!((v.run_id, v.job_id), (Some(7), Some(70))); + assert_eq!(v.job_name.as_deref(), Some("e2e (shard 3)")); + assert_eq!( + (v.exit_code, v.duration_secs, v.finished_at), + (Some(1), Some(90), Some(220)) + ); + let d = v.during.unwrap(); + assert_eq!(d.secs, 100); + assert_eq!(d.host_cpu_busy_cores, Some(15.0)); + assert_eq!(d.drone_cpu_busy_cores, Some(4.0)); + assert!(v.slow_host); + assert_eq!(v.slow_host_reasons.len(), 2, "{:?}", v.slow_host_reasons); + + let old = s.job_host(71, 30.0).await.unwrap().unwrap(); + assert_eq!(old.start, None); + assert!(!old.slow_host); + assert_eq!(old.exit_code, Some(1)); + + let run = s.run_hosts(7, 30.0).await.unwrap(); + assert_eq!( + run.iter().map(|v| v.lease_id.as_str()).collect::>(), + ["a", "b"] + ); + assert!(s.job_host(999, 30.0).await.unwrap().is_none()); + + // Re-binding the same job is a no-op; a retried result keeps samples. + s.bind_lease_host_job(&a, "o/r", 7, 70, "e2e (shard 3)") + .await + .unwrap(); + s.record_lease_host(&a, &result, None, 240).await.unwrap(); + assert!(s.job_host(70, 30.0).await.unwrap().unwrap().start.is_some()); + + // Age prunes a; the row cap then keeps only the newest one (c). + assert_eq!(s.prune_lease_hosts(150, 1).await.unwrap(), 2); + assert!(s.job_host(70, 30.0).await.unwrap().is_none()); + assert!(s.job_host(71, 30.0).await.unwrap().is_none()); + assert!(s.job_host(80, 30.0).await.unwrap().is_some()); + } +} diff --git a/src/lib.rs b/src/lib.rs index f54d334..68dcafe 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,7 @@ pub mod explain; pub mod github; pub mod health; pub mod homeostasis; +pub mod hostdiag; pub mod janitor; pub mod lead; pub mod mirror_auth; diff --git a/src/preempt.rs b/src/preempt.rs index ddb3628..e3a9dd9 100644 --- a/src/preempt.rs +++ b/src/preempt.rs @@ -210,6 +210,7 @@ mod tests { blocked_classes: Vec::new(), trust: crate::classes::Trust::Shared, cpu_psi_pm: None, + cpu_load_pm: None, } } diff --git a/src/probe/mod.rs b/src/probe/mod.rs index 76b9904..586ee98 100644 --- a/src/probe/mod.rs +++ b/src/probe/mod.rs @@ -89,10 +89,24 @@ pub fn env_facts(host: &dyn Host, now: i64) -> EnvFacts { observed_at: now, mem_available_mib: mem_available_mib(host), cpu_psi_some_avg10_pm: cpu_psi_pm(host), + load_per_cpu_pm: load_per_cpu_pm(host), ..EnvFacts::default() } } +/// The 1-minute load average per usable CPU, in permille. +fn load_per_cpu_pm(host: &dyn Host) -> Option { + let load = host + .read_to_string(Path::new("/proc/loadavg"))? + .split_whitespace() + .next()? + .parse::() + .ok() + .filter(|l| l.is_finite() && *l >= 0.0)?; + let cpus = f64::from(host.available_parallelism()?.max(1)); + Some((load / cpus * 1000.0).round().min(f64::from(u32::MAX)) as u32) +} + /// How long a tools probe is reused before the tools are looked up again. const TOOLS_TTL_SECS: i64 = 60; /// Longest version string reported for a tool. @@ -162,7 +176,7 @@ impl ToolsCache { } } -fn mem_available_mib(host: &dyn Host) -> Option { +pub(crate) fn mem_available_mib(host: &dyn Host) -> Option { let meminfo = host.read_to_string(Path::new("/proc/meminfo"))?; let kib = meminfo .lines() @@ -510,6 +524,18 @@ mod tests { assert!(exec.calls.lock().unwrap().len() > n); } + #[test] + fn load_per_cpu_is_permille_of_usable_cpus() { + let mut h = linux(); + assert_eq!(load_per_cpu_pm(&h), None); + h.files + .insert("/proc/loadavg", "93.40 78.56 69.31 90/3277 3570264\n"); + assert_eq!(load_per_cpu_pm(&h), Some(5838)); + assert_eq!(env_facts(&h, 1).load_per_cpu_pm, Some(5838)); + h.files.insert("/proc/loadavg", "garbage"); + assert_eq!(load_per_cpu_pm(&h), None); + } + #[test] fn psi_is_the_higher_of_host_and_cgroup_pressure() { let mut h = linux(); diff --git a/src/queen.rs b/src/queen.rs index b983335..15ff14d 100644 --- a/src/queen.rs +++ b/src/queen.rs @@ -595,12 +595,15 @@ impl Queen { None, now_ts, ); - if let Some(a) = &self.cfg.admission { - state.cpu_psi_pm = d + if let Some(a) = &self.cfg.admission + && let Some(f) = d .env_facts .as_ref() .filter(|f| now_ts - f.observed_at <= a.facts_max_age_secs) - .and_then(|f| f.cpu_psi_some_avg10_pm); + { + state.cpu_psi_pm = f.cpu_psi_some_avg10_pm; + state.cpu_load_pm = + crate::allocator::load_pressure_pm(f.load_per_cpu_pm, f.cpu_busy_pm); } state.profiles = profiles .iter() @@ -963,6 +966,16 @@ impl Queen { if let Err(e) = self.store.prune_telemetry(now_ts - keep).await { tracing::warn!("prune telemetry failed: {e}"); } + if let Err(e) = self + .store + .prune_lease_hosts( + now_ts - crate::hostdiag::KEEP_SECS, + crate::hostdiag::MAX_ROWS, + ) + .await + { + tracing::warn!("prune lease host diagnostics failed: {e}"); + } // Grants and the snapshot are published before optional feedback. Bound // this pass so unavailable job history cannot stall subsequent ticks. match tokio::time::timeout( @@ -1366,11 +1379,10 @@ impl Queen { groups[&(u.platform.as_str(), u.rust_toolchain.as_deref())] .iter() .filter(|d| { - let state = AdmissionState::of(d.cpu_psi_pm, admission); d.accepting && d.trust.admits(u.scope) - && state != AdmissionState::Closed - && (state != AdmissionState::PrBlocked || u.scope != Scope::Pr) + && AdmissionState::of(d.admission_pressure_pm(), admission) + .admits(u.class, u.scope) && !d.blocked_classes.contains(&u.class) && max_slots(d, self.cfg.classes.spec(u.class)) > 0 && (d.bids.is_empty() @@ -1641,6 +1653,9 @@ impl Queen { && let Some(lease) = self.store.lease_for_runner(id, name).await? { self.store.remember_job(&lease, repo, job.id, now).await?; + self.store + .bind_lease_host_job(&lease, repo, job.run_id, job.id, &job.name) + .await?; } } let mut pending = self.store.pending_jobs().await?; @@ -1687,6 +1702,9 @@ impl Queen { { continue; } + self.store + .bind_lease_host_job(&lease, &repo, job.run_id, job.id, &job.name) + .await?; if let Some((telemetry, duration, finished)) = job.telemetry(&repo) { self.store .record_github_telemetry(&lease, &telemetry, duration, finished) diff --git a/src/store.rs b/src/store.rs index fff1d44..6e6e150 100644 --- a/src/store.rs +++ b/src/store.rs @@ -178,6 +178,27 @@ CREATE TABLE IF NOT EXISTS job_telemetry ( finished_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS job_telemetry_finished ON job_telemetry(finished_at); +-- Host load at lease start and end, with the GitHub job the lease ran; see +-- hostdiag.rs. Pruned after seven days and to a row cap. +CREATE TABLE IF NOT EXISTS lease_hosts ( + lease_id TEXT PRIMARY KEY, + drone TEXT NOT NULL, + class TEXT NOT NULL, + scope TEXT NOT NULL, + leased_at INTEGER NOT NULL, + repo TEXT, + run_id INTEGER, + job_id INTEGER, + job_name TEXT, + finished_at INTEGER, + exit_code INTEGER, + duration_secs INTEGER, + start_sample TEXT, + end_sample TEXT +); +CREATE INDEX IF NOT EXISTS lease_hosts_job ON lease_hosts(job_id); +CREATE INDEX IF NOT EXISTS lease_hosts_run ON lease_hosts(run_id); +CREATE INDEX IF NOT EXISTS lease_hosts_leased ON lease_hosts(leased_at); CREATE TABLE IF NOT EXISTS preemptions ( lease_id TEXT PRIMARY KEY, drone TEXT NOT NULL, diff --git a/tests/drone.rs b/tests/drone.rs index 75da712..2629fa7 100644 --- a/tests/drone.rs +++ b/tests/drone.rs @@ -198,6 +198,19 @@ async fn runs_a_granted_lease_cleans_up_and_reports_the_result() { .is_empty(), "result reported, lease done" ); + let diag = rig + .h + .queen + .store() + .lease_host(&lease_id, 30.0) + .await + .unwrap() + .expect("host samples stored with the result"); + assert_eq!(diag.drone, "d1"); + assert_eq!(diag.exit_code, Some(0)); + let (start, end) = (diag.start.unwrap(), diag.end.unwrap()); + assert_eq!((start.cpus, start.active_leases), (Some(16), 1)); + assert!(end.at >= start.at); let calls = rig.exec.0.lock().unwrap().clone(); assert!( calls diff --git a/tests/queen.rs b/tests/queen.rs index a3dfe63..f838da1 100644 --- a/tests/queen.rs +++ b/tests/queen.rs @@ -785,6 +785,168 @@ async fn observed_running_job_is_collected_after_workflow_leaves_the_queue() { assert!(h.queen.store().pending_jobs().await.unwrap().is_empty()); } +#[tokio::test] +async fn job_and_run_endpoints_show_the_drone_and_its_host_load() { + let h = Harness::start( + json!({"admission": {"psi_soft_pm": 300, "psi_hard_pm": 600, "facts_max_age_secs": 180}}), + ) + .await; + let token = h.enroll("shared-box").await; + h.heartbeat(&token, &[]).await; + let lease_id = lease_for(&h, &token, "borg-ci-heavy", 4242).await; + let lease = h.queen.store().lease(&lease_id).await.unwrap().unwrap(); + let start = OffsetDateTime::now_utc() - time::Duration::seconds(45); + let ts = start + .format(&time::format_description::well_known::Rfc3339) + .unwrap(); + h.state().jobs = vec![json!({ + "id": 4242, "run_id": 1, "name": "ui-e2e (shard 3)", "status": "in_progress", + "conclusion": null, "labels": ["borg-ci-heavy"], + "runner_id": lease.runner_id, "runner_name": lease.runner_name, + "created_at": ts, "started_at": ts, "completed_at": null, "steps": [] + })]; + h.tick().await; + + // Seen running, no result yet: the drone is known, samples are not. + let job: Value = h + .http + .get(format!("{}/jobs/4242", h.url)) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(job["drone"], "shared-box"); + // The mock serves this job under queued run 1. + assert_eq!(job["run_id"], 1); + assert_eq!(job["job_name"], "ui-e2e (shard 3)"); + assert_eq!(job["start"], Value::Null); + assert_eq!(job["slow_host"], false); + + let status = post_result( + &h, + &token, + &lease_id, + json!({"exit_code": 1, "duration_secs": 100, "host": { + "start": {"at": 1000, "cpus": 16, "load": [77.0, 60.0, 50.0], + "cgroup_cpu_psi": {"some_avg10": 54.9, "some_avg60": 52.4}, + "host_cpu_usage_usec": 0, "drone_cpu_usage_usec": 0, + "work_free_mib": 30000, "cache_free_mib": 30000, "active_leases": 2}, + "end": {"at": 1100, "cpus": 16, "host_cpu_usage_usec": 1550000000u64, + "drone_cpu_usage_usec": 300000000, "active_leases": 1}}}), + ) + .await; + assert_eq!(status, 204); + + let job: Value = h + .http + .get(format!("{}/jobs/4242", h.url)) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(job["exit_code"], 1); + assert_eq!(job["duration_secs"], 100); + assert_eq!(job["start"]["active_leases"], 2); + assert_eq!(job["during"]["host_cpu_busy_cores"], 15.5); + assert_eq!(job["during"]["drone_cpu_busy_cores"], 3.0); + assert_eq!(job["slow_host"], true); + let reasons = job["slow_host_reasons"].as_array().unwrap(); + assert!( + reasons[0] + .as_str() + .unwrap() + .starts_with("cpu pressure 54.9% at start"), + "{reasons:?}" + ); + + let run: Value = h + .http + .get(format!("{}/runs/1", h.url)) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(run["jobs"].as_array().unwrap().len(), 1); + assert_eq!(run["jobs"][0]["job_id"], 4242); + + let missing = h + .http + .get(format!("{}/jobs/1", h.url)) + .send() + .await + .unwrap(); + assert_eq!(missing.status(), 404); + let empty: Value = h + .http + .get(format!("{}/runs/2", h.url)) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(empty["jobs"], json!([])); +} + +#[tokio::test] +async fn admission_falls_back_to_load_without_psi_and_light_only_sheds_main_heavy() { + let h = Harness::start(json!({"admission": { + "psi_soft_pm": 300, "psi_light_only_pm": 500, "psi_hard_pm": 600, "facts_max_age_secs": 180}})) + .await; + let token = h.enroll("no-psi").await; + let now = OffsetDateTime::now_utc().unix_timestamp(); + // No PSI, load 93 on 16 saturated CPUs: estimated 828 permille, closed. + h.heartbeat_with( + &token, + json!({"env_facts": {"observed_at": now, "load_per_cpu_pm": 5838, "cpu_busy_pm": 990}}), + ) + .await; + h.queue_job(1, "borg-ci-heavy", 5); + h.tick().await; + let s = h.status().await; + let a = &s["drones"][0]["admission"]; + assert_eq!(a["state"], "closed", "{a}"); + assert_eq!(a["cpu_psi_pm"], Value::Null); + assert_eq!(a["cpu_load_pm"], 828); + assert!( + h.state().mints.is_empty(), + "an overloaded drone without PSI gets no main heavy" + ); + assert!( + h.metrics() + .await + .contains("borg_drone_cpu_load_pm{drone=\"no-psi\"} 828") + ); + + // With PSI 540 the drone is light-only: main light placed, main heavy not. + h.heartbeat_with( + &token, + json!({"env_facts": {"observed_at": now, "cpu_psi_some_avg10_pm": 540, + "load_per_cpu_pm": 5838, "cpu_busy_pm": 990}}), + ) + .await; + h.queue_job(2, "borg-ci-light", 5); + h.tick().await; + assert_eq!( + h.status().await["drones"][0]["admission"]["state"], + "light_only" + ); + let minted: Vec = h + .state() + .mints + .iter() + .map(|m| m["labels"].to_string()) + .collect(); + assert_eq!(minted.len(), 1, "{minted:?}"); + assert!(minted[0].contains("borg-ci-light"), "{minted:?}"); +} + #[tokio::test] async fn github_telemetry_requires_exact_runner_identity_and_preserves_drone_evidence() { let h = Harness::start(json!({})).await;