diff --git a/docs/CAPACITY-TELEMETRY.md b/docs/CAPACITY-TELEMETRY.md new file mode 100644 index 0000000..89b274b --- /dev/null +++ b/docs/CAPACITY-TELEMETRY.md @@ -0,0 +1,23 @@ +# Local capacity telemetry + +The dedicated capacity timer checks every 30 seconds and records a sample only while at least one managed runner is active. This interval captures normal jobs shorter than the five-minute health period without increasing heartbeat or status-report traffic. Samples remain on the controller host at `/var/lib/ci-fleet/capacity/samples.jsonl`; they are never added to status-reporting or heartbeat payloads. + +Each sample contains only: + +- timestamp and logical controller/pool ID; +- interval host CPU utilization plus memory, swap, disk-byte, and inode counters; +- anonymous per-runner CPU percentage and memory use/limit. + +Container IDs, names, repositories, jobs, logs, environment variables, source, network counters, and credentials are not stored. The history directory must be a root-owned, non-symlink mode-`0700` directory and the file must be mode `0600`. Malformed records are ignored atomically, records older than eight days are discarded even while the pool is idle, and history is compacted from at most 26,000 records back to 24,000 (more than eight days at the scheduled 30-second interval). Normal samples append one record; compaction rewrites only when retention, malformed input, or the rotation threshold requires it. Uninstall removes this fleet-owned local history. Runner-stat failures fail the capacity service and appear in the normal health report instead of being recorded as zero observations. + +## Weekly report + +Run on the controller host; reads local state only and changes nothing: + +```bash +sudo /opt/ci-fleet/current/scripts/health.py capacity-report +``` + +The JSON report groups the preceding seven days by logical pool and gives nearest-rank p50/p95 values for every observed host and runner metric, plus sample and anonymous runner-observation counts. An empty report means no managed runner was observed during the period. + +Review the report before changing runner count, per-runner resources, or pool budget. A capacity change still requires its own reviewed private-configuration change; this public repository does not create or modify that external change. diff --git a/docs/DESIRED-STATE.md b/docs/DESIRED-STATE.md index dca40f5..b702f8a 100644 --- a/docs/DESIRED-STATE.md +++ b/docs/DESIRED-STATE.md @@ -88,7 +88,7 @@ The installer: 10. starts the controller only when its desired state is active and verifies runtime health; 11. atomically records redacted installation state, then enables the maintenance timers. -A successful second `--install` run reports `NO_CHANGE` and performs no unnecessary replacement. A successful engine upgrade advances both the runtime release and the maintenance installer-manager to the same pinned commit; rollback restores both. +A successful second `--install` run reports `NO_CHANGE` and performs no unnecessary replacement. Engine upgrades advance the runtime release and maintenance installer-manager together. A deliberate runtime downgrade retains a newer compatible manager so a later reviewed upgrade can still be rendered and activated; maintenance units always come from the selected runtime. Rollback restores both. ## Adopt an existing controller diff --git a/docs/README.md b/docs/README.md index 90b0499..8107a48 100644 --- a/docs/README.md +++ b/docs/README.md @@ -22,7 +22,7 @@ New operator? Use [Operator workflows](OPERATOR-WORKFLOWS.md) to select a suppor | Make a project compliant | [Project CI standard](PROJECT-STANDARD.md) and [compliance checklist](COMPLIANCE-CHECKLIST.md) | | Split tests across parallel workers | [Project CI standard](PROJECT-STANDARD.md) and the [parallel workflow example](../examples/workflows/parallel-ci.yml.example) | | Configure automatic updates and cleanup | [Host maintenance](HOST-MAINTENANCE.md) | -| Monitor hosts and detect missed reports | [Fleet health monitoring](HEALTH-MONITORING.md) and [authenticated status reporting](STATUS-REPORTING.md) | +| Monitor hosts and detect missed reports | [Fleet health monitoring](HEALTH-MONITORING.md), [local capacity telemetry](CAPACITY-TELEMETRY.md), and [authenticated status reporting](STATUS-REPORTING.md) | | Handle GitHub App, workflow, or deployment secrets | [Secrets model](SECRETS.md) and [security policy](../SECURITY.md) | | Review accepted implementation scope | [Design decisions](DESIGN-DECISIONS.md) | | Run private CI or deployment for a public project | [Public projects, private delivery, and private configuration](PUBLIC-PRIVATE-CONFIGURATION.md) | diff --git a/engine-capabilities.json b/engine-capabilities.json index 12c4d6d..bf07344 100644 --- a/engine-capabilities.json +++ b/engine-capabilities.json @@ -1,6 +1,7 @@ { "schema_version": 1, "capabilities": { + "capacity_telemetry": true, "status_reporting_config": true, "required_status_reporting": true } diff --git a/host/systemd/ci-fleet-capacity.service b/host/systemd/ci-fleet-capacity.service new file mode 100644 index 0000000..7904231 --- /dev/null +++ b/host/systemd/ci-fleet-capacity.service @@ -0,0 +1,21 @@ +[Unit] +Description=Sample local ci-fleet capacity +After=docker.service +Requires=docker.service +ConditionPathExists=/etc/ci-fleet/ci-fleet.env + +[Service] +Type=oneshot +ExecStart=/opt/ci-fleet/current/scripts/capacity-sample.sh +User=root +Group=root +NoNewPrivileges=yes +PrivateTmp=yes +ProtectHome=yes +ProtectSystem=strict +ReadWritePaths=/var/lib/ci-fleet +LockPersonality=yes +MemoryDenyWriteExecute=yes +RestrictAddressFamilies=AF_UNIX +SystemCallArchitectures=native +UMask=0077 diff --git a/host/systemd/ci-fleet-capacity.timer b/host/systemd/ci-fleet-capacity.timer new file mode 100644 index 0000000..4af5d60 --- /dev/null +++ b/host/systemd/ci-fleet-capacity.timer @@ -0,0 +1,11 @@ +[Unit] +Description=Sample active ci-fleet capacity every 30 seconds + +[Timer] +OnActiveSec=15s +OnUnitActiveSec=30s +AccuracySec=1s +Unit=ci-fleet-capacity.service + +[Install] +WantedBy=timers.target diff --git a/scripts/capacity-sample.sh b/scripts/capacity-sample.sh new file mode 100755 index 0000000..a8e1461 --- /dev/null +++ b/scripts/capacity-sample.sh @@ -0,0 +1,16 @@ +#!/usr/bin/env bash +set -Eeuo pipefail + +script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +environment=/etc/ci-fleet/ci-fleet.env +args=(capacity-sample) +if [[ ${CI_FLEET_TESTING:-0} == 1 && -n ${CI_FLEET_ROOT_PREFIX:-} ]]; then + environment="$CI_FLEET_ROOT_PREFIX/etc/ci-fleet/ci-fleet.env" + args+=(--root "$CI_FLEET_ROOT_PREFIX" --history "$CI_FLEET_ROOT_PREFIX/var/lib/ci-fleet/capacity/samples.jsonl") +fi +[[ -r $environment ]] || { printf 'CRITICAL capacity_configuration_missing\n' >&2; exit 2; } +set -a +# shellcheck disable=SC1090 +. "$environment" +set +a +exec python3 "$script_dir/health.py" "${args[@]}" diff --git a/scripts/desired_state.py b/scripts/desired_state.py index f19af22..aaaa41f 100755 --- a/scripts/desired_state.py +++ b/scripts/desired_state.py @@ -30,6 +30,7 @@ HOST_OPTIONAL = {"CI_FLEET_RUNNER_TTL"} REQUIRED_STATUS_CAPABILITY = "required_status_reporting" STATUS_REPORTING_CONFIG_CAPABILITY = "status_reporting_config" +CAPACITY_TELEMETRY_CAPABILITY = "capacity_telemetry" class DesiredStateError(ValueError): @@ -194,6 +195,8 @@ def build_rendered_env( "CI_FLEET_VERSION": short_commit, **validate_host_values(host_values), } + if CAPACITY_TELEMETRY_CAPABILITY in (engine_capabilities or set()): + rendered["CI_FLEET_POOL"] = controller["pool"] reporting_configured = "status_reporting" in controller reporting_required = (controller.get("status_reporting") or {}).get("enabled") is True if reporting_required and REQUIRED_STATUS_CAPABILITY not in (engine_capabilities or set()): @@ -292,6 +295,8 @@ def command_validate_engine_capabilities(args: argparse.Namespace) -> None: raise DesiredStateError("selected engine does not support status reporting configuration") if args.require_status_reporting and REQUIRED_STATUS_CAPABILITY not in capabilities: raise DesiredStateError("selected engine does not advertise required status reporting") + if args.require_capacity_telemetry and CAPACITY_TELEMETRY_CAPABILITY not in capabilities: + raise DesiredStateError("selected engine does not advertise capacity telemetry") print("ENGINE_CAPABILITIES_OK") @@ -329,6 +334,7 @@ def parse_args() -> argparse.Namespace: capabilities.add_argument("--manifest", type=Path, required=True) capabilities.add_argument("--require-status-reporting-config", action="store_true") capabilities.add_argument("--require-status-reporting", action="store_true") + capabilities.add_argument("--require-capacity-telemetry", action="store_true") capabilities.set_defaults(function=command_validate_engine_capabilities) return parser.parse_args() diff --git a/scripts/health.py b/scripts/health.py index f47b9e1..0b0b4a8 100644 --- a/scripts/health.py +++ b/scripts/health.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse +import math import http.client import json import os @@ -17,6 +18,7 @@ from pathlib import Path from typing import Any, Callable +sys.dont_write_bytecode = True from status_auth import sign_headers @@ -242,7 +244,7 @@ def build_status_report(snapshot: dict[str, Any], health_report: dict[str, Any], }, "runners": snapshot.get("runners", {"current": 0, "busy": 0, "maximum": 0}), "metrics": { - "cpu": snapshot.get("cpu", {"logical": 1, "used_percent": 0}), + "cpu": {name: snapshot.get("cpu", {}).get(name, default) for name, default in (("logical", 1), ("used_percent", 0))}, "memory": snapshot.get("memory", {"total_bytes": 0, "available_bytes": 0}), "swap": snapshot.get("swap", {"total_bytes": 0, "used_bytes": 0}), "disk": {name: {"total_bytes": value.get("total_bytes", 0), "used_bytes": value.get("used_bytes", 0)} for name, value in disks.items()}, @@ -383,16 +385,15 @@ def _memory(root: Path) -> tuple[int, int]: def _cpu(root: Path) -> dict[str, float | int]: - # ponytail: cumulative boot-average CPU; persist the prior sample if interval utilization becomes necessary. try: fields = next(line for line in (root / "proc/stat").read_text().splitlines() if line.startswith("cpu ")).split()[1:] counters = [int(value) for value in fields] total = sum(counters[:8]) - used = total - counters[3] - percent = round(100 * used / max(total, 1), 1) + idle = counters[3] + percent = round(100 * (total - idle) / max(total, 1), 1) except (OSError, ValueError, IndexError, StopIteration): - percent = 0.0 - return {"logical": max(os.cpu_count() or 1, 1), "used_percent": percent} + total, idle, percent = 0, 0, 0.0 + return {"logical": max(os.cpu_count() or 1, 1), "used_percent": percent, "total_ticks": total, "idle_ticks": idle} def _boot_time(root: Path) -> int: @@ -498,6 +499,9 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run managed["restarting"] += int(state == "restarting" or "restarting" in status) oom = run(["journalctl", "--dmesg", "--since=-24h", "--grep=Out of memory|Killed process", "--quiet"]) timer_ages = {"health": 900, "cleanup": 172800, "drift": 3600} + capacity_installed = (root / "etc/systemd/system/ci-fleet-capacity.timer").is_file() + if capacity_installed: + timer_ages["capacity"] = 120 remote_config = bool(re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", values.get("CI_FLEET_CONFIG_REPOSITORY", ""))) reconciliation = _reconcile_state(root / "var/lib/ci-fleet/reconcile/state.json") if remote_config else None if reconciliation and values.get("CI_FLEET_HEALTH_BOOTSTRAP") == "1": @@ -505,10 +509,9 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run if remote_config: timer_ages["reconcile"] = 900 timers = {name: _unit_state(run, f"ci-fleet-{name}.timer", timer=True, max_age_seconds=age) for name, age in timer_ages.items()} - service_units = { - "cleanup": "ci-fleet-cleanup.service", - "drift": "ci-fleet-drift.service", - } + service_units = {"cleanup": "ci-fleet-cleanup.service", "drift": "ci-fleet-drift.service"} + if capacity_installed: + service_units["capacity"] = "ci-fleet-capacity.service" if remote_config: service_units["reconcile"] = "ci-fleet-reconcile.service" services = {name: _unit_state(run, unit) for name, unit in service_units.items()} @@ -525,6 +528,7 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run stale["build_cache"] = _count(run, ["docker", "buildx", "du", "--filter", "until=168h", "--format", "json"]) if docker_ok else 0 return { "controller_id": instance, + "pool_id": values.get("CI_FLEET_POOL", instance), "desired_state": values.get("CI_FLEET_CONTROLLER_STATE", "active"), "disks": {"root": _disk(str(root)), "docker": _disk(str(root / docker_root.lstrip("/")))}, "cpu": _cpu(root), @@ -584,6 +588,254 @@ def _write_report(path: Path, report: dict[str, Any]) -> None: temporary.replace(path) +CAPACITY_RETENTION_SECONDS = 8 * 24 * 60 * 60 +CAPACITY_MAX_SAMPLES = 24000 +CAPACITY_COMPACT_SAMPLES = 26000 + + +def _quantity_bytes(value: str) -> int: + match = re.fullmatch(r"([0-9]+(?:\.[0-9]+)?)\s*([KMGT]?i?B)", value.strip(), re.IGNORECASE) + if not match: + raise ValueError("invalid Docker memory quantity") + units = {"b": 1, "kb": 1000, "mb": 1000**2, "gb": 1000**3, "tb": 1000**4, + "kib": 1024, "mib": 1024**2, "gib": 1024**3, "tib": 1024**4} + return round(float(match.group(1)) * units[match.group(2).lower()]) + + +def _runner_capacity(run: Runner, instance: str) -> list[dict[str, int | float]]: + result = run(["docker", "ps", "-q", "--filter", "label=io.randomdevelopment.ci-fleet.managed=true", + "--filter", "label=io.randomdevelopment.ci-fleet.kind=runner", + "--filter", f"label=io.randomdevelopment.ci-fleet.instance={instance}"]) + if result.returncode != 0: + raise ValueError("managed runner discovery failed") + ids = [value for value in result.stdout.splitlines() if re.fullmatch(r"[0-9a-f]{12,64}", value)] + if not ids: + raise ValueError("active managed runners have no stat targets") + result = run(["docker", "stats", "--no-stream", "--format", "{{.CPUPerc}}\t{{.MemUsage}}", *ids]) + if result.returncode != 0: + raise ValueError("managed runner stat collection failed") + metrics = [] + for line in result.stdout.splitlines(): + try: + cpu, memory = line.split("\t", 1) + used, limit = memory.split("/", 1) + cpu_percent = float(cpu.removesuffix("%")) + if not math.isfinite(cpu_percent) or cpu_percent < 0: + raise ValueError + metrics.append({"cpu_percent": cpu_percent, "memory_used_bytes": _quantity_bytes(used), "memory_limit_bytes": _quantity_bytes(limit)}) + except ValueError as error: + raise ValueError("managed runner stat output is invalid") from error + if len(metrics) != len(ids): + raise ValueError("managed runner stat output is incomplete") + return metrics + + +def _capacity_state(path: Path, timestamp: int, *, create: bool = False) -> tuple[list[dict[str, Any]], bool]: + expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0 + try: + parent = path.parent.lstat() + except FileNotFoundError: + if not create: + return [], False + path.parent.mkdir(parents=True, mode=0o700) + parent = path.parent.lstat() + if not stat.S_ISDIR(parent.st_mode) or parent.st_uid != expected_owner or stat.S_IMODE(parent.st_mode) != 0o700: + raise ValueError("capacity history directory must be root-owned mode 0700") + retained = [] + dirty = False + try: + info = path.lstat() + except FileNotFoundError: + info = None + if info is not None: + if not stat.S_ISREG(info.st_mode) or info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) != 0o600: + raise ValueError("capacity history must be root-owned mode 0600") + for line in path.read_text().splitlines(): + try: + previous = json.loads(line) + if timestamp - CAPACITY_RETENTION_SECONDS <= previous["timestamp"] <= timestamp: + retained.append(previous) + else: + dirty = True + except (KeyError, TypeError, ValueError, json.JSONDecodeError): + dirty = True + continue + if len(retained) >= CAPACITY_COMPACT_SAMPLES: + retained = retained[-CAPACITY_COMPACT_SAMPLES:] + dirty = True + return retained, dirty + + +def _write_capacity(path: Path, samples: list[dict[str, Any]]) -> None: + temporary = path.with_suffix(".tmp") + temporary.write_text("".join(json.dumps(value, separators=(",", ":"), sort_keys=True) + "\n" for value in samples)) + os.chmod(temporary, 0o600) + temporary.replace(path) + + +def _append_capacity(path: Path, sample: dict[str, Any]) -> None: + with path.open("a", encoding="utf-8") as stream: + stream.write(json.dumps(sample, separators=(",", ":"), sort_keys=True) + "\n") + os.chmod(path, 0o600) + + +def _update_cpu_baseline(path: Path, timestamp: int, cpu: dict[str, Any]) -> dict[str, int] | None: + baseline = path.with_suffix(path.suffix + ".cpu") + expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0 + previous = None + if baseline.exists(): + info = baseline.lstat() + if not stat.S_ISREG(info.st_mode) or info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) != 0o600: + raise ValueError("capacity CPU baseline must be root-owned mode 0600") + try: + value = json.loads(baseline.read_text()) + if set(value) != {"timestamp", "total_ticks", "idle_ticks"} or any(type(value[name]) is not int or value[name] < 0 for name in value): + raise ValueError + if value["idle_ticks"] > value["total_ticks"]: + raise ValueError + previous = value + except (TypeError, ValueError, json.JSONDecodeError) as error: + raise ValueError("capacity CPU baseline is invalid") from error + current = {"timestamp": timestamp, "total_ticks": cpu.get("total_ticks"), "idle_ticks": cpu.get("idle_ticks")} + if any(type(current[name]) is not int or current[name] < 0 for name in current) or current["idle_ticks"] > current["total_ticks"]: + raise ValueError("current CPU counters are invalid") + temporary = baseline.with_suffix(baseline.suffix + ".tmp") + temporary.write_text(json.dumps(current, separators=(",", ":"), sort_keys=True) + "\n") + os.chmod(temporary, 0o600) + temporary.replace(baseline) + if previous and 0 < timestamp - previous["timestamp"] <= 120 and current["total_ticks"] >= previous["total_ticks"] and current["idle_ticks"] >= previous["idle_ticks"]: + return previous + return None + + +def record_capacity(path: Path, snapshot: dict[str, Any], *, run: Runner = _run, now: int | None = None) -> bool: + timestamp = int(time.time() if now is None else now) + retained, dirty = _capacity_state(path, timestamp, create=True) + cpu = snapshot["cpu"] + previous_host = _update_cpu_baseline(path, timestamp, cpu) + if snapshot.get("runners", {}).get("current", 0) < 1: + if dirty: + _write_capacity(path, retained[-CAPACITY_MAX_SAMPLES:]) + return False + pool_id = snapshot.get("pool_id", snapshot["controller_id"]) + memory = snapshot["memory"] + host = { + "memory_used_bytes": memory["total_bytes"] - memory["available_bytes"], + "swap_used_bytes": snapshot["swap"]["used_bytes"], + "disk_used_bytes": {name: value["used_bytes"] for name, value in snapshot["disks"].items()}, + "inode_used": {name: value["inode_used"] for name, value in snapshot["disks"].items()}, + "cpu_total_ticks": cpu["total_ticks"], + "cpu_idle_ticks": cpu["idle_ticks"], + } + if previous_host: + total_delta = cpu["total_ticks"] - previous_host["total_ticks"] + idle_delta = cpu["idle_ticks"] - previous_host["idle_ticks"] + if total_delta > 0 and 0 <= idle_delta <= total_delta: + host["cpu_percent"] = round(100 * (total_delta - idle_delta) / total_delta, 1) + sample = {"timestamp": timestamp, "pool": pool_id, "host": host, "runners": _runner_capacity(run, snapshot["controller_id"])} + if dirty or len(retained) >= CAPACITY_COMPACT_SAMPLES: + _write_capacity(path, (retained + [sample])[-CAPACITY_MAX_SAMPLES:]) + else: + _append_capacity(path, sample) + return True + + +def _percentiles(values: list[int | float]) -> dict[str, int | float] | None: + if not values: + return None + ordered = sorted(values) + return {name: ordered[max(0, math.ceil(len(ordered) * percentile) - 1)] for name, percentile in (("p50", 0.50), ("p95", 0.95))} + + +def _sample_values(sample: Any, timestamp: int) -> tuple[str, dict[str, list[int | float]], int] | None: + if not isinstance(sample, dict) or set(sample) != {"timestamp", "pool", "host", "runners"}: + return None + if type(sample["timestamp"]) is not int or not timestamp - 7 * 24 * 60 * 60 <= sample["timestamp"] <= timestamp: + return None + if not isinstance(sample["pool"], str) or not re.fullmatch(r"[A-Za-z0-9._-]+", sample["pool"]): + return None + host, runners = sample["host"], sample["runners"] + required_host = {"memory_used_bytes", "swap_used_bytes", "disk_used_bytes", "inode_used", "cpu_total_ticks", "cpu_idle_ticks"} + if not isinstance(host, dict) or not required_host <= set(host) <= required_host | {"cpu_percent"} or not isinstance(runners, list): + return None + total_ticks, idle_ticks = host["cpu_total_ticks"], host["cpu_idle_ticks"] + if any(type(value) is not int or value < 0 for value in (total_ticks, idle_ticks)) or idle_ticks > total_ticks: + return None + values: dict[str, list[int | float]] = {} + for name in ("memory_used_bytes", "swap_used_bytes"): + if type(host[name]) is not int or host[name] < 0: + return None + values[f"host_{name}"] = [host[name]] + if "cpu_percent" in host: + if type(host["cpu_percent"]) not in (int, float) or not math.isfinite(host["cpu_percent"]) or not 0 <= host["cpu_percent"] <= 100: + return None + values["host_cpu_percent"] = [host["cpu_percent"]] + for kind in ("disk_used_bytes", "inode_used"): + if not isinstance(host[kind], dict): + return None + for name, value in host[kind].items(): + if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z0-9._-]+", name) or type(value) is not int or value < 0: + return None + values[f"host_{kind}_{name}"] = [value] + runner_count = 0 + for runner in runners: + if not isinstance(runner, dict) or set(runner) != {"cpu_percent", "memory_used_bytes", "memory_limit_bytes"}: + return None + if type(runner["cpu_percent"]) not in (int, float) or not math.isfinite(runner["cpu_percent"]) or runner["cpu_percent"] < 0: + return None + if any(type(runner[name]) is not int or runner[name] < 0 for name in ("memory_used_bytes", "memory_limit_bytes")): + return None + runner_count += 1 + for name, value in runner.items(): + values.setdefault(f"runner_{name}", []).append(value) + return sample["pool"], values, runner_count + + +def capacity_report(path: Path, *, now: int | None = None) -> dict[str, Any]: + timestamp = int(time.time() if now is None else now) + pools: dict[str, dict[str, Any]] = {} + samples, _ = _capacity_state(path, timestamp) + for sample in samples: + parsed = _sample_values(sample, timestamp) + if parsed is None: + continue + pool_name, sample_values, runner_count = parsed + pool = pools.setdefault(pool_name, {"samples": 0, "runner_observations": 0, "values": {}}) + pool["samples"] += 1 + pool["runner_observations"] += runner_count + for name, observed in sample_values.items(): + pool["values"].setdefault(name, []).extend(observed) + for pool in pools.values(): + pool["metrics"] = {name: _percentiles(values) for name, values in sorted(pool.pop("values").items())} + return {"schema_version": 1, "generated_at": timestamp, "period_seconds": 7 * 24 * 60 * 60, "pools": pools} + + +def collect_capacity_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Runner = _run) -> dict[str, Any]: + instance = values.get("CI_FLEET_INSTANCE", "") + if not re.fullmatch(r"[A-Za-z0-9._-]+", instance): + raise ValueError("controller identity is invalid") + docker_root_result = run(["docker", "info", "--format", "{{.DockerRootDir}}"]) + docker_root = docker_root_result.stdout.strip() + if docker_root_result.returncode != 0 or not docker_root.startswith("/"): + raise ValueError("Docker root directory is unavailable") + runners = run(["docker", "ps", "-q", "--filter", "label=io.randomdevelopment.ci-fleet.managed=true", + "--filter", "label=io.randomdevelopment.ci-fleet.kind=runner", + "--filter", f"label=io.randomdevelopment.ci-fleet.instance={instance}"]) + if runners.returncode != 0: + raise ValueError("managed runner discovery failed") + current = len([value for value in runners.stdout.splitlines() if re.fullmatch(r"[0-9a-f]{12,64}", value)]) + memory, swap = _memory_details(root) + return { + "controller_id": instance, + "pool_id": values.get("CI_FLEET_POOL", instance), + "cpu": _cpu(root), + "memory": memory, + "swap": swap, + "disks": {"root": _disk(str(root)), "docker": _disk(str(root / docker_root.lstrip("/")))}, + "runners": {"current": current}, + } + + def _send_status( values: dict[str, str], report: dict[str, Any], @@ -721,6 +973,18 @@ def _heartbeats(args: argparse.Namespace) -> int: return int(report["exit_code"]) +def _capacity(args: argparse.Namespace) -> int: + print(json.dumps(capacity_report(args.history), indent=2, sort_keys=True)) + return 0 + + +def _capacity_sample(args: argparse.Namespace) -> int: + snapshot = collect_capacity_snapshot(dict(os.environ), root=args.root) + recorded = record_capacity(args.history, snapshot) + print(f"CAPACITY_SAMPLE recorded={str(recorded).lower()}") + return 0 + + def main() -> int: parser = argparse.ArgumentParser(description="Redacted ci-fleet host health") commands = parser.add_subparsers(dest="command", required=True) @@ -735,6 +999,13 @@ def main() -> int: heartbeats.add_argument("--grace-seconds", type=int, default=900) heartbeats.add_argument("--json", action="store_true") heartbeats.set_defaults(handler=_heartbeats) + capacity = commands.add_parser("capacity-report") + capacity.add_argument("--history", type=Path, default=Path("/var/lib/ci-fleet/capacity/samples.jsonl")) + capacity.set_defaults(handler=_capacity) + sample = commands.add_parser("capacity-sample") + sample.add_argument("--history", type=Path, default=Path("/var/lib/ci-fleet/capacity/samples.jsonl")) + sample.add_argument("--root", type=Path, default=Path("/")) + sample.set_defaults(handler=_capacity_sample) args = parser.parse_args() try: return args.handler(args) diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh index ad15656..dda8853 100755 --- a/scripts/install-worker-controller.sh +++ b/scripts/install-worker-controller.sh @@ -120,6 +120,7 @@ unit_names=( ci-fleet-drift.service ci-fleet-drift.timer ) timer_names=(ci-fleet-health.timer ci-fleet-cleanup.timer ci-fleet-drift.timer) +capacity_unit_names=(ci-fleet-capacity.service ci-fleet-capacity.timer) optional_unit_names=( ci-fleet-reconcile.service ci-fleet-reconcile.timer ) @@ -507,6 +508,12 @@ runtime_release_complete() { [[ "$actual_digest" == "$stored_digest" ]] } +engine_has_capacity_telemetry() { + local path=$1 + [[ -f "$path/engine-capabilities.json" ]] && python3 "$repo_root/scripts/desired_state.py" validate-engine-capabilities \ + --manifest "$path/engine-capabilities.json" --require-capacity-telemetry >/dev/null 2>&1 +} + manager_release_complete() { local path=$1 expected=$2 require_status=${3:-0} require_schema=${4:-0} marker required unit runtime_release_complete "$path" "$expected" "$require_status" "$require_schema" || return 1 @@ -516,6 +523,9 @@ manager_release_complete() { done [[ -x "$path/templates/config-repository/scripts/validate.sh" ]] || return 1 for unit in "${unit_names[@]}"; do [[ -f "$path/host/systemd/$unit" ]] || return 1; done + if engine_has_capacity_telemetry "$path" || [[ -e "$path/scripts/capacity-sample.sh" || -e "$path/host/systemd/ci-fleet-capacity.service" || -e "$path/host/systemd/ci-fleet-capacity.timer" ]]; then + [[ -x "$path/scripts/capacity-sample.sh" && -f "$path/host/systemd/ci-fleet-capacity.service" && -f "$path/host/systemd/ci-fleet-capacity.timer" ]] || return 1 + fi marker=$(<"$path/.ci-fleet-engine-ref") [[ "$marker" == "$expected" ]] } @@ -538,19 +548,32 @@ managed_images_match() { } systemd_matches() { - local expected_manager unit + local expected_manager manager_commit unit expected_manager=$manager_releases/$engine_ref - manager_release_complete "$expected_manager" "$engine_ref" "$status_reporting_required" "$status_reporting_configured" || return 1 [[ -L "$manager_current" ]] || return 1 + if ! engine_has_capacity_telemetry "$release_dir" && engine_has_capacity_telemetry "$(readlink -f "$manager_current")"; then + expected_manager=$(readlink -f "$manager_current") + fi + [[ -f "$expected_manager/.ci-fleet-engine-ref" ]] || return 1 + manager_commit=$(<"$expected_manager/.ci-fleet-engine-ref") + manager_release_complete "$expected_manager" "$manager_commit" "$status_reporting_required" "$status_reporting_configured" || return 1 [[ $(readlink -f "$manager_current") == $(readlink -f "$expected_manager") ]] || return 1 for unit in "${unit_names[@]}"; do [[ -f "$systemd_dir/$unit" ]] || return 1 - cmp -s "$expected_manager/host/systemd/$unit" "$systemd_dir/$unit" || return 1 + cmp -s "$release_dir/host/systemd/$unit" "$systemd_dir/$unit" || return 1 done for unit in "${timer_names[@]}"; do systemctl is-enabled --quiet "$unit" || return 1 systemctl is-active --quiet "$unit" || return 1 done + if [[ -x "$release_dir/scripts/capacity-sample.sh" ]]; then + for unit in "${capacity_unit_names[@]}"; do + [[ -f "$systemd_dir/$unit" ]] && cmp -s "$release_dir/host/systemd/$unit" "$systemd_dir/$unit" || return 1 + done + systemctl is-enabled --quiet ci-fleet-capacity.timer && systemctl is-active --quiet ci-fleet-capacity.timer || return 1 + else + for unit in "${capacity_unit_names[@]}"; do [[ ! -e "$systemd_dir/$unit" ]] || return 1; done + fi } drift_count() { @@ -635,8 +658,16 @@ install_release() { } install_manager() { - local manager_commit manager_release archive staged_manager + local manager_commit manager_release archive staged_manager current_manager current_manager_commit manager_commit=$engine_ref + current_manager=$(readlink -f "$manager_current" 2>/dev/null || true) + current_manager_commit= + [[ -z "$current_manager" || ! -f "$current_manager/.ci-fleet-engine-ref" ]] || current_manager_commit=$(<"$current_manager/.ci-fleet-engine-ref") + if ! engine_has_capacity_telemetry "$release_dir" && [[ -n "$current_manager_commit" ]] \ + && engine_has_capacity_telemetry "$current_manager" \ + && manager_release_complete "$current_manager" "$current_manager_commit" "$status_reporting_required" "$status_reporting_configured"; then + return + fi [[ "$manager_commit" =~ ^[0-9a-f]{40}$ ]] || die 'installer manager commit is invalid' runtime_release_complete "$release_dir" "$manager_commit" "$status_reporting_required" "$status_reporting_configured" || die 'desired engine release is unavailable for installer manager activation' manager_release=$manager_releases/$manager_commit @@ -695,7 +726,7 @@ make_checkpoint() { printf '%s\n' "$target" >"$checkpoint_dir/manager-target" chmod 0600 "$checkpoint_dir/manager-target" fi - for unit in "${unit_names[@]}" "${optional_unit_names[@]}"; do + for unit in "${unit_names[@]}" "${capacity_unit_names[@]}" "${optional_unit_names[@]}"; do [[ ! -f "$systemd_dir/$unit" ]] || install -m 0644 "$systemd_dir/$unit" "$checkpoint_dir/systemd/$unit" done : >"$checkpoint_dir/enabled-timers" @@ -704,6 +735,8 @@ make_checkpoint() { if systemctl is-enabled --quiet "$timer" 2>/dev/null; then printf '%s\n' "$timer" >>"$checkpoint_dir/enabled-timers"; fi if systemctl is-active --quiet "$timer" 2>/dev/null; then printf '%s\n' "$timer" >>"$checkpoint_dir/active-timers"; fi done + if systemctl is-enabled --quiet ci-fleet-capacity.timer 2>/dev/null; then printf '%s\n' ci-fleet-capacity.timer >>"$checkpoint_dir/enabled-timers"; fi + if systemctl is-active --quiet ci-fleet-capacity.timer 2>/dev/null; then printf '%s\n' ci-fleet-capacity.timer >>"$checkpoint_dir/active-timers"; fi local opt_name for opt_name in "${optional_unit_names[@]}"; do case "$opt_name" in *.timer) @@ -790,6 +823,16 @@ install_systemd_units() { install -d -m 0755 "$systemd_dir" install -m 0644 "$source/host/systemd/ci-fleet-health.service" "$systemd_dir/" install -m 0644 "$source/host/systemd/ci-fleet-health.timer" "$systemd_dir/" + if [[ -x "$source/scripts/capacity-sample.sh" ]]; then + install -m 0644 "$source/host/systemd/ci-fleet-capacity.service" "$systemd_dir/" + install -m 0644 "$source/host/systemd/ci-fleet-capacity.timer" "$systemd_dir/" + else + systemctl disable --now ci-fleet-capacity.timer >/dev/null 2>&1 || true + if [[ -f "$systemd_dir/ci-fleet-capacity.service" ]] && ! systemctl stop ci-fleet-capacity.service; then + return 1 + fi + rm -f "$systemd_dir/ci-fleet-capacity.service" "$systemd_dir/ci-fleet-capacity.timer" + fi install -m 0644 "$source/host/systemd/ci-fleet-cleanup.service" "$systemd_dir/" install -m 0644 "$source/host/systemd/ci-fleet-cleanup.timer" "$systemd_dir/" install -m 0644 "$source/host/systemd/ci-fleet-drift.service" "$systemd_dir/" @@ -803,11 +846,15 @@ install_systemd_units() { remove_systemd_units() { systemctl disable --now "${timer_names[@]}" >/dev/null 2>&1 || true + systemctl disable --now ci-fleet-capacity.timer >/dev/null 2>&1 || true + if [[ -f "$systemd_dir/ci-fleet-capacity.service" ]] && ! systemctl stop ci-fleet-capacity.service; then + return 1 + fi local unit for unit in "${optional_unit_names[@]}"; do case "$unit" in *.timer) systemctl disable --now "$unit" >/dev/null 2>&1 || true ;; esac done - for unit in "${unit_names[@]}" "${optional_unit_names[@]}"; do rm -f "$systemd_dir/$unit"; done + for unit in "${unit_names[@]}" "${capacity_unit_names[@]}" "${optional_unit_names[@]}"; do rm -f "$systemd_dir/$unit"; done systemctl daemon-reload } @@ -858,7 +905,7 @@ activate_candidate() { ln -sfn "$release_dir" "$temporary/current" mv -Tf "$temporary/current" "$current_link" install_manager - install_systemd_units "$(readlink -f "$manager_current")" + install_systemd_units "$release_dir" if [[ "$target_state" == active ]]; then compose "$release_dir" "$rendered_env" up -d --no-deps controller sleep "${CI_FLEET_STARTUP_WAIT_SECONDS:-2}" @@ -887,6 +934,7 @@ PY chmod 0600 "$staged_state" mv -f "$staged_state" "$state_file" systemctl enable --now "${timer_names[@]}" >/dev/null + if [[ -f "$systemd_dir/ci-fleet-capacity.timer" ]]; then systemctl enable --now ci-fleet-capacity.timer >/dev/null; fi local opt_timer for opt_timer in "${optional_unit_names[@]}"; do case "$opt_timer" in *.timer) @@ -908,11 +956,14 @@ restore_systemd_snapshot() { for unit in "${unit_names[@]}"; do [[ ! -f "$checkpoint_dir/systemd/$unit" ]] || install -m 0644 "$checkpoint_dir/systemd/$unit" "$systemd_dir/$unit" || failed=1 done + for unit in "${capacity_unit_names[@]}"; do + [[ ! -f "$checkpoint_dir/systemd/$unit" ]] || install -m 0644 "$checkpoint_dir/systemd/$unit" "$systemd_dir/$unit" || failed=1 + done for unit in "${optional_unit_names[@]}"; do [[ ! -f "$checkpoint_dir/systemd/$unit" ]] || install -m 0644 "$checkpoint_dir/systemd/$unit" "$systemd_dir/$unit" || failed=1 done systemctl daemon-reload || failed=1 - for timer in "${timer_names[@]}"; do + for timer in "${timer_names[@]}" ci-fleet-capacity.timer; do if grep -Fxq "$timer" "$checkpoint_dir/enabled-timers"; then systemctl enable "$timer" >/dev/null || failed=1; else systemctl disable "$timer" >/dev/null 2>&1 || true; fi if grep -Fxq "$timer" "$checkpoint_dir/active-timers"; then systemctl start "$timer" || failed=1; else systemctl stop "$timer" >/dev/null 2>&1 || true; fi done @@ -1090,7 +1141,7 @@ perform_uninstall() { fi remove_systemd_units rm -f "$current_link" "$rendered_env" "$state_file" - rm -rf -- "$state_root/health" + rm -rf -- "$state_root/health" "$state_root/capacity" rm -f "$manager_current" transaction_active=false note "UNINSTALL_OK host_config_preserved=$host_config secrets_preserved=$etc_dir/secrets" diff --git a/scripts/test-install-worker-controller.sh b/scripts/test-install-worker-controller.sh index 445a77d..d093c12 100755 --- a/scripts/test-install-worker-controller.sh +++ b/scripts/test-install-worker-controller.sh @@ -140,6 +140,11 @@ EOF cat >"$fake_bin/systemctl" <<'EOF' #!/usr/bin/env bash +[[ -z "${FAKE_SYSTEMCTL_LOG:-}" ]] || printf '%s\n' "$*" >>"$FAKE_SYSTEMCTL_LOG" +if [[ -n "${FAKE_FAIL_CAPACITY_STOP_ONCE:-}" && -f "$FAKE_FAIL_CAPACITY_STOP_ONCE" && "$*" == 'stop ci-fleet-capacity.service' ]]; then + rm -f "$FAKE_FAIL_CAPACITY_STOP_ONCE" + exit 96 +fi if [[ "${1:-}" == enable && "${2:-}" == --now && ! -f "${CI_FLEET_ROOT_PREFIX:-}/var/lib/ci-fleet/install-state.json" ]]; then exit 98 fi @@ -231,6 +236,7 @@ for dockerfile in "$repo_root/controller/Dockerfile" "$repo_root/runner/Dockerfi done grep -Fq ' user: "0:0"' "$repo_root/deploy/compose.yaml" || fail 'controller cannot read the required root-owned mode-0600 GitHub App PEM' grep -Fq 'export PYTHONDONTWRITEBYTECODE=1' "$repo_root/scripts/install-worker-controller.sh" || fail 'managed validation may write Python bytecode into the immutable manager release' +grep -Fq 'sys.dont_write_bytecode = True' "$repo_root/scripts/health.py" || fail 'manual capacity reports may write Python bytecode into the immutable runtime release' grep -Fq ' trap - ERR' "$repo_root/scripts/install-worker-controller.sh" || fail 'warning health subprocess inherits the transactional rollback trap' grep -Fq "CI_FLEET_COMMIT: \${CI_FLEET_COMMIT:-unknown}" "$repo_root/deploy/compose.yaml" || fail 'runner build lacks engine provenance argument' config_repo=$tmp/config-repo @@ -577,7 +583,10 @@ grep -Fq 'CI_FLEET_MAX_RUNNERS=2' "$root/etc/ci-fleet/ci-fleet.env" || fail 'upg mkdir -p "$root/var/lib/ci-fleet/checkpoints/99999999-incomplete" printf 'restarting\n' >"$FAKE_CONTROLLER_STATUS_FILE" rm -f "$root/var/lib/ci-fleet/install-state.json" "$root/etc/ci-fleet/ci-fleet.env" +export FAKE_SYSTEMCTL_LOG=$tmp/rollback-systemctl.log expect_success "$installer" --rollback >/dev/null +grep -Fxq 'start ci-fleet-capacity.timer' "$FAKE_SYSTEMCTL_LOG" || fail 'rollback did not restore the active capacity timer' +unset FAKE_SYSTEMCTL_LOG [[ ! -f "$FAKE_CONTROLLER_STATUS_FILE" ]] || fail 'explicit rollback did not recover a restarting controller' grep -Fq 'CI_FLEET_MAX_RUNNERS=1' "$root/etc/ci-fleet/ci-fleet.env" || fail 'rollback did not restore capacity one' @@ -690,8 +699,31 @@ legacy_ref=$(write_config active 1 1 "$legacy_engine_ref" omit) export FAKE_ENGINE_REF=$legacy_engine_ref export FAKE_RUNNER_IMAGE=ci-fleet-runner:${legacy_engine_ref:0:12} export FAKE_CONTROLLER_IMAGE=ci-fleet-controller:${legacy_engine_ref:0:12} +export FAKE_SYSTEMCTL_LOG=$tmp/legacy-systemctl.log +export FAKE_FAIL_CAPACITY_STOP_ONCE=$tmp/fail-capacity-stop-once +: >"$FAKE_FAIL_CAPACITY_STOP_ONCE" +expect_failure 'ROLLBACK_RESTORED' "$installer" --upgrade "${base_args[@]}" --ref "$legacy_ref" +unset FAKE_FAIL_CAPACITY_STOP_ONCE expect_success "$installer" --upgrade "${base_args[@]}" --ref "$legacy_ref" >/dev/null +grep -Fxq 'stop ci-fleet-capacity.service' "$FAKE_SYSTEMCTL_LOG" || fail 'legacy upgrade did not stop the in-flight capacity service' +unset FAKE_SYSTEMCTL_LOG [[ $(readlink -f "$adopt_root/opt/ci-fleet/current") == "$adopt_root/opt/ci-fleet/releases/$legacy_engine_ref" ]] || fail 'upgrade could not restore a pre-health-contract engine' +[[ $(readlink -f "$adopt_root/opt/ci-fleet/manager/current") == "$adopt_root/opt/ci-fleet/manager/releases/$engine_ref" ]] || fail 'legacy downgrade replaced the forward-compatible manager' +[[ ! -e "$adopt_root/etc/systemd/system/ci-fleet-capacity.timer" ]] || fail 'legacy runtime retained capacity units' +retained_installer=$adopt_root/opt/ci-fleet/manager/current/scripts/install-worker-controller.sh +export FAKE_ENGINE_REF=$engine_ref +export FAKE_RUNNER_IMAGE=ci-fleet-runner:${engine_ref:0:12} +export FAKE_CONTROLLER_IMAGE=ci-fleet-controller:${engine_ref:0:12} +expect_success "$retained_installer" --upgrade "${base_args[@]}" --ref "$ref_one" >/dev/null +[[ -f "$adopt_root/etc/systemd/system/ci-fleet-capacity.timer" ]] || fail 'upgrade back from legacy did not restore capacity units' +grep -Fq 'CI_FLEET_POOL=trusted-ci' "$adopt_root/etc/ci-fleet/ci-fleet.env" || fail 'upgrade back from legacy did not restore the telemetry pool' +rollback_stop_ref=$(write_config active 2 2) +export FAKE_FAIL_UP_ONCE=$tmp/rollback-stop-fail-up +export FAKE_FAIL_CAPACITY_STOP_ONCE=$tmp/rollback-stop-failure +: >"$FAKE_FAIL_UP_ONCE" +: >"$FAKE_FAIL_CAPACITY_STOP_ONCE" +expect_failure 'ROLLBACK_FAILED' "$retained_installer" --upgrade "${base_args[@]}" --ref "$rollback_stop_ref" +unset FAKE_FAIL_UP_ONCE FAKE_FAIL_CAPACITY_STOP_ONCE grep -Fq 'Issue #7' "$repo_root/docs/DESIGN-DECISIONS.md" || fail 'isolated proof approval is not recorded' if grep -Fq '/etc/ci-fleet/ci-fleet.env.before-max2' "$repo_root/docs/CAPACITY-PROMOTION.md"; then fail 'capacity runbook still edits rendered host state'; fi diff --git a/scripts/test_desired_state.py b/scripts/test_desired_state.py index c37a441..75acb36 100755 --- a/scripts/test_desired_state.py +++ b/scripts/test_desired_state.py @@ -45,7 +45,7 @@ def render(self, value: dict | None = None, capabilities: set[str] | None = None config_repository="example-org/example-fleet-config", config_ref=CONFIG_COMMIT, docker_gid=998, - engine_capabilities={"status_reporting_config"} if capabilities is None else capabilities, + engine_capabilities={"status_reporting_config", "capacity_telemetry"} if capabilities is None else capabilities, ) def test_active_controller_renders_configured_capacity(self) -> None: @@ -53,9 +53,14 @@ def test_active_controller_renders_configured_capacity(self) -> None: self.assertEqual(environment["CI_FLEET_MAX_RUNNERS"], "1") self.assertEqual(environment["CI_FLEET_CONFIGURED_MAX_RUNNERS"], "1") self.assertEqual(environment["CI_FLEET_LABELS"], "docker-ci") + self.assertEqual(environment["CI_FLEET_POOL"], "trusted-ci") self.assertEqual(environment["CI_FLEET_COMMIT"], environment["CI_FLEET_ENGINE_REF"]) self.assertEqual(metadata["controller_state"], "active") + def test_legacy_engine_does_not_receive_capacity_pool_variable(self) -> None: + environment, _ = self.render(capabilities={"status_reporting_config"}) + self.assertNotIn("CI_FLEET_POOL", environment) + def test_status_reporting_requires_fixed_host_local_configuration(self) -> None: value = config() value["controllers"]["example-ci-01"]["status_reporting"] = { diff --git a/scripts/test_health.py b/scripts/test_health.py index ed4cf56..601056e 100644 --- a/scripts/test_health.py +++ b/scripts/test_health.py @@ -123,9 +123,137 @@ def test_malformed_reconciliation_state_is_observable(self) -> None: snapshot = healthy_snapshot() snapshot["controller_id"] = "example-ci-01" snapshot["reconciliation"] = reconciliation + snapshot["cpu"] = {"logical": 8, "used_percent": 10, "total_ticks": 123, "idle_ticks": 45} report = health.build_status_report(snapshot, health.evaluate(snapshot, health.Thresholds()), generated_at=1_000) self.assertEqual(report["configuration"], {"desired_commit": "", "applied_commit": ""}) self.assertEqual(report["error"]["code"], "reconciliation_invalid") + self.assertEqual(set(report["metrics"]["cpu"]), {"logical", "used_percent"}) + + def test_capacity_history_is_resource_only_bounded_and_aggregated(self) -> None: + with tempfile.TemporaryDirectory() as directory: + history = Path(directory) / "capacity" / "samples.jsonl" + snapshot = healthy_snapshot() + snapshot.update({ + "cpu": {"logical": 8, "used_percent": 10, "total_ticks": 100, "idle_ticks": 40}, + "memory": {"total_bytes": 1000, "available_bytes": 600}, + "swap": {"total_bytes": 100, "used_bytes": 5}, + "runners": {"current": 1, "busy": 1, "maximum": 2}, + "secret": "must-not-be-recorded", + }) + for value in snapshot["disks"].values(): + value.update(used_bytes=200, inode_used=20) + runner_cpu = [12.5, 90.0] + + def run(args): + output = "a" * 12 + "\n" if args[:3] == ["docker", "ps", "-q"] else f"{runner_cpu.pop(0)}%\t256MiB / 2GiB\n" + return health.subprocess.CompletedProcess(args, 0, output, "") + + previous_testing = os.environ.get("CI_FLEET_TESTING") + os.environ["CI_FLEET_TESTING"] = "1" + try: + history.parent.mkdir(mode=0o700) + history.write_text('{"timestamp":999999}\n' * health.CAPACITY_MAX_SAMPLES) + history.chmod(0o600) + self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_000)) + history_inode = history.stat().st_ino + snapshot["cpu"].update(used_percent=20, total_ticks=200, idle_ticks=50) + self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_030)) + self.assertLessEqual(len(history.read_text().splitlines()), health.CAPACITY_COMPACT_SAMPLES) + self.assertEqual(history.stat().st_ino, history_inode) + self.assertEqual(history.stat().st_mode & 0o777, 0o600) + self.assertNotIn("must-not-be-recorded", history.read_text()) + report = health.capacity_report(history, now=1_000_030) + pool = report["pools"]["example-ci-01"] + self.assertEqual((pool["samples"], pool["runner_observations"]), (2, 2)) + self.assertEqual(pool["metrics"]["host_cpu_percent"], {"p50": 90.0, "p95": 90.0}) + self.assertEqual(pool["metrics"]["runner_cpu_percent"], {"p50": 12.5, "p95": 90.0}) + snapshot["runners"]["current"] = 0 + history_before_idle = history.read_bytes() + self.assertFalse(health.record_capacity(history, snapshot, run=run, now=1_000_900)) + self.assertEqual(history.read_bytes(), history_before_idle) + snapshot["runners"]["current"] = 1 + snapshot["cpu"].update(total_ticks=300, idle_ticks=70) + runner_cpu.append(25.0) + self.assertTrue(health.record_capacity(history, snapshot, run=run, now=1_000_930)) + self.assertEqual(json.loads(history.read_text().splitlines()[-1])["host"]["cpu_percent"], 80.0) + finally: + if previous_testing is None: + os.environ.pop("CI_FLEET_TESTING", None) + else: + os.environ["CI_FLEET_TESTING"] = previous_testing + + def test_capacity_prunes_inactive_history_and_ignores_incomplete_samples(self) -> None: + with tempfile.TemporaryDirectory() as directory: + history = Path(directory) / "capacity" / "samples.jsonl" + missing = Path(directory) / "missing" / "samples.jsonl" + self.assertEqual(health.capacity_report(missing, now=1_000_000)["pools"], {}) + self.assertFalse(missing.parent.exists()) + history.parent.mkdir(mode=0o700) + valid = { + "timestamp": 999_999, "pool": "example-ci-01", + "host": {"memory_used_bytes": 1, "swap_used_bytes": 0, "disk_used_bytes": {"root": 1}, + "inode_used": {"root": 1}, "cpu_total_ticks": 10, "cpu_idle_ticks": 5, "cpu_percent": 50.0}, + "runners": [{"cpu_percent": 1.0, "memory_used_bytes": 1, "memory_limit_bytes": 2}], + } + history.write_text(json.dumps({**valid, "timestamp": 1}) + "\n" + json.dumps(valid) + "\n") + history.chmod(0o600) + previous_testing = os.environ.get("CI_FLEET_TESTING") + os.environ["CI_FLEET_TESTING"] = "1" + try: + self.assertFalse(health.record_capacity(history, {"runners": {"current": 0}, "cpu": {"total_ticks": 20, "idle_ticks": 10}}, now=1_000_000)) + self.assertEqual(len(history.read_text().splitlines()), 1) + incomplete = {**valid, "timestamp": 999_998, "host": dict(valid["host"])} + incomplete.pop("runners") + malformed_counters = {**valid, "timestamp": 999_997, "host": {**valid["host"], "cpu_total_ticks": "10"}} + history.write_text(history.read_text() + json.dumps(incomplete) + "\n" + json.dumps(malformed_counters) + "\n") + report = health.capacity_report(history, now=1_000_000) + self.assertEqual(report["pools"]["example-ci-01"]["samples"], 1) + redirected = Path(directory) / "redirected" + redirected.mkdir(mode=0o700) + other = Path(directory) / "other" + other.mkdir(mode=0o700) + redirected.rmdir() + redirected.symlink_to(other, target_is_directory=True) + with self.assertRaisesRegex(ValueError, "directory must be root-owned"): + health.record_capacity(redirected / "samples.jsonl", {"runners": {"current": 0}}, now=1_000_000) + dangling_target = Path(directory) / "outside.jsonl" + dangling = history.parent / "dangling.jsonl" + dangling.symlink_to(dangling_target) + with self.assertRaisesRegex(ValueError, "history must be root-owned"): + health.record_capacity(dangling, {"runners": {"current": 0}}, now=1_000_000) + self.assertFalse(dangling_target.exists()) + finally: + if previous_testing is None: + os.environ.pop("CI_FLEET_TESTING", None) + else: + os.environ["CI_FLEET_TESTING"] = previous_testing + + def test_capacity_surfaces_runner_stat_failures(self) -> None: + snapshot = healthy_snapshot() + snapshot.update({ + "cpu": {"logical": 8, "used_percent": 10, "total_ticks": 100, "idle_ticks": 40}, + "memory": {"total_bytes": 1000, "available_bytes": 600}, + "swap": {"total_bytes": 0, "used_bytes": 0}, + "runners": {"current": 1}, + }) + for value in snapshot["disks"].values(): + value.update(used_bytes=1, inode_used=1) + def run(args): + return health.subprocess.CompletedProcess(args, 0 if args[:3] == ["docker", "ps", "-q"] else 1, + "a" * 12 + "\n" if args[:3] == ["docker", "ps", "-q"] else "", "failed") + with tempfile.TemporaryDirectory() as directory: + history = Path(directory) / "capacity" / "samples.jsonl" + history.parent.mkdir(mode=0o700) + previous_testing = os.environ.get("CI_FLEET_TESTING") + os.environ["CI_FLEET_TESTING"] = "1" + try: + with self.assertRaisesRegex(ValueError, "stat collection failed"): + health.record_capacity(history, snapshot, run=run, now=1_000_000) + finally: + if previous_testing is None: + os.environ.pop("CI_FLEET_TESTING", None) + else: + os.environ["CI_FLEET_TESTING"] = previous_testing def test_external_heartbeats_detect_missing_and_stale_active_hosts(self) -> None: controllers = { @@ -180,6 +308,8 @@ def run(args): self.assertEqual((snapshot["load_per_cpu"], snapshot["swap_used_percent"]), (3.0, 50)) self.assertEqual(set(snapshot["services"]), {"cleanup", "drift"}) self.assertEqual(set(snapshot["timers"]), {"health", "cleanup", "drift"}) + (root / "etc/systemd/system").mkdir(parents=True) + (root / "etc/systemd/system/ci-fleet-capacity.timer").write_text("fixture\n") (root / "var/lib/ci-fleet/reconcile").mkdir(parents=True) (root / "var/lib/ci-fleet/reconcile/state.json").write_text('{"status":"rolled_back","desired_commit":"","applied_commit":"","health":"healthy"}\n') remote = health.collect_snapshot( @@ -191,11 +321,11 @@ def run(args): root=root, run=run, ) - self.assertEqual(set(remote["services"]), {"cleanup", "drift", "reconcile"}) + self.assertEqual(set(remote["services"]), {"capacity", "cleanup", "drift", "reconcile"}) self.assertEqual(set(remote["services"].values()), {"ok"}) - self.assertEqual(set(remote["timers"]), {"health", "cleanup", "drift", "reconcile"}) + self.assertEqual(set(remote["timers"]), {"health", "capacity", "cleanup", "drift", "reconcile"}) self.assertEqual(remote["reconciliation"]["status"], "bootstrap") - (root / "etc").mkdir() + (root / "etc").mkdir(exist_ok=True) (root / "etc/debian_version").write_text("13\n") debian = health.collect_snapshot({"CI_FLEET_CONTROLLER_STATE": "disabled", "CI_FLEET_HEALTH_BOOTSTRAP": "1"}, root=root, run=run) self.assertIn("updates", debian["services"]) @@ -296,7 +426,7 @@ def test_status_report_contract_redaction_and_disabled_ssh(self) -> None: "software_version": "1" * 40, "boot_time": 900, "ssh": "disabled", - "cpu": {"logical": 8, "used_percent": 25.0}, + "cpu": {"logical": 8, "used_percent": 25.0, "total_ticks": 0, "idle_ticks": 0}, "memory": {"total_bytes": 1024, "available_bytes": 768}, "swap": {"total_bytes": 512, "used_bytes": 0}, "load": {"one": 0.1, "five": 0.2, "fifteen": 0.3},