From 6f4bffe4b3541f6d9d3eb047f9464f8e3accad65 Mon Sep 17 00:00:00 2001 From: Jean Schmidt Date: Thu, 2 Jul 2026 12:05:38 -0700 Subject: [PATCH 01/16] sweep sim: node-size-sweep bin-packing simulator + prod-parity flags Real 2D CPU+memory+GPU bin-packing simulator that replays HUD workflow_job data against a Karpenter-style cluster. Warming-node model, ARC placeholder emulation, DaemonSet + phantom-pod prod-parity toggles, and per-fleet / cluster-wide utilization reports. --- osdc/.gitignore | 7 + osdc/scripts/node-size-sweep/build_csv.py | 270 ++++++++++ osdc/scripts/node-size-sweep/plan.md | 176 +++++++ osdc/scripts/node-size-sweep/pull_hud.py | 219 ++++++++ osdc/scripts/node-size-sweep/sim_load.py | 134 +++++ osdc/scripts/node-size-sweep/sim_nodes.py | 245 +++++++++ osdc/scripts/node-size-sweep/sim_report.py | 145 ++++++ osdc/scripts/node-size-sweep/simulate.py | 554 +++++++++++++++++++++ 8 files changed, 1750 insertions(+) create mode 100644 osdc/scripts/node-size-sweep/build_csv.py create mode 100644 osdc/scripts/node-size-sweep/plan.md create mode 100644 osdc/scripts/node-size-sweep/pull_hud.py create mode 100644 osdc/scripts/node-size-sweep/sim_load.py create mode 100644 osdc/scripts/node-size-sweep/sim_nodes.py create mode 100644 osdc/scripts/node-size-sweep/sim_report.py create mode 100644 osdc/scripts/node-size-sweep/simulate.py diff --git a/osdc/.gitignore b/osdc/.gitignore index 691e0ac9..f98bd255 100644 --- a/osdc/.gitignore +++ b/osdc/.gitignore @@ -24,3 +24,10 @@ modules/*/generated/ .DS_Store .env +# node-size-sweep: large HUD data + local backups (never commit) +scripts/node-size-sweep/*.csv +scripts/node-size-sweep/raw/ +scripts/node-size-sweep/*.bkp +scripts/node-size-sweep/*.bkp.* +scripts/node-size-sweep/raw.bkp/ + diff --git a/osdc/scripts/node-size-sweep/build_csv.py b/osdc/scripts/node-size-sweep/build_csv.py new file mode 100644 index 00000000..705f4576 --- /dev/null +++ b/osdc/scripts/node-size-sweep/build_csv.py @@ -0,0 +1,270 @@ +#!/usr/bin/env python3 +# /// script +# requires-python = ">=3.12" +# dependencies = ["pyyaml>=6.0"] +# /// +"""One-off: build label -> (nodepool, nodepool_fraction) table + join to HUD jobs. + +Hacky tool for the node-size sweep analysis. Splits into two subcommands: + + build-table Print label -> (nodepool, nodepool_fraction) mapping as JSON. + extract Read one or more raw ClickHouse JSON chunks (each is a JSON + list of [label, started_at, completed_at, runtime_s] rows, + which is what the ClickHouse MCP `run_clickhouse_query` + drops into `result_file`), filter, bucket, join, and write + a CSV. + +Columns in the CSV: + provider, label, nodepool, nodepool_fraction, start_time, end_time + +`provider` is either 'mt' or 'lf' — extracted from the HUD label prefix, +so the simulator can filter (e.g. drop 'lf' to simulate a single Meta cluster). + +`nodepool_fraction` is 1/N notation stored as the N — e.g. a 1/12 slice of a +c7a.24xlarge is stored as 12. `1` means "the runner takes the whole node". +""" + +from __future__ import annotations + +import argparse +import csv +import datetime as dt +import json +import re +import sys +from pathlib import Path + +import yaml + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(REPO_ROOT / "scripts" / "python")) + +from fleet_naming import derive_fleet_name # noqa: E402 +from instance_specs import INSTANCE_SPECS # noqa: E402 +from pytorch_workload_data import OLD_TO_NEW_LABEL # noqa: E402 + +RUNNER_DEF_DIRS = [ + REPO_ROOT / "modules" / "arc-runners" / "defs", + REPO_ROOT / "modules" / "arc-runners-h100" / "defs", + REPO_ROOT / "modules" / "arc-runners-b200" / "defs", +] + +# Runtime bucketing and filtering. +BUCKET_SECONDS = 300 # 5-min buckets on start_time +MAX_RUNTIME_S = 6 * 60 * 60 # drop >6h "bogus" +MIN_RUNTIME_S = 1 + + +def _parse_mem_gib(mem: str) -> float: + """Parse a k8s memory string ("225Gi", "1000Gi", "512Mi") into GiB (float).""" + m = re.match(r"^\s*(\d+(?:\.\d+)?)\s*([GMKgmk]i?)\s*$", mem) + if not m: + raise ValueError(f"unparseable memory: {mem!r}") + val = float(m.group(1)) + unit = m.group(2).lower() + if unit in ("gi", "g"): + return val + if unit in ("mi", "m"): + return val / 1024 + if unit in ("ki", "k"): + return val / (1024 * 1024) + raise ValueError(f"unknown unit in {mem!r}") + + +def _load_runner_defs() -> list[dict]: + """Read every runner def and return a list of parsed runner dicts.""" + out = [] + for d in RUNNER_DEF_DIRS: + if not d.is_dir(): + continue + for f in sorted(d.glob("*.yaml")): + data = yaml.safe_load(f.read_text()) or {} + r = data.get("runner") + if not isinstance(r, dict): + continue + out.append(r) + return out + + +def _nodepool_capacity(instance_type: str) -> tuple[float, float]: + """Return (vcpu, memory_gib) for the fleet's biggest instance. + + We index the fraction against the LARGEST instance in the fleet — that's + the size ARC/Karpenter tends to prefer (weight=100 in every fleet def). + This makes fraction comparisons cross-fleet consistent. + """ + spec = INSTANCE_SPECS.get(instance_type) + if spec is None: + raise KeyError(f"instance_type {instance_type!r} missing from INSTANCE_SPECS") + # memory_mi is real k8s capacity; convert to GiB for comparison with def mem. + return float(spec["vcpu"]), spec["memory_mi"] / 1024 + + +def _pick_fraction(runner_vcpu: float, runner_mem_gib: float, node_vcpu: float, node_mem_gib: float) -> int: + """Return the N (as in 1/N) that best fits this runner on this node. + + N = min(fits_by_cpu, fits_by_mem), integer floor. Overhead intentionally + ignored — this is a theoretical-maximum count, the simulator will layer + overhead in later if it wants to. + """ + if runner_vcpu <= 0 or runner_mem_gib <= 0: + return 1 + fits_cpu = int(node_vcpu // runner_vcpu) + fits_mem = int(node_mem_gib // runner_mem_gib) + return max(1, min(fits_cpu, fits_mem)) + + +def build_label_table() -> dict[str, dict]: + """Return { runner_label: {nodepool, nodepool_fraction, ...} }.""" + table: dict[str, dict] = {} + for r in _load_runner_defs(): + name = r["name"] + instance = r["instance_type"] + fleet = derive_fleet_name(instance, r.get("node_fleet")) + v = float(r["vcpu"]) + m = _parse_mem_gib(str(r["memory"])) + node_v, node_m = _nodepool_capacity(instance) + frac = _pick_fraction(v, m, node_v, node_m) + table[name] = { + "nodepool": fleet, + "instance_type": instance, + "nodepool_fraction": frac, + "vcpu": v, + "memory_gib": m, + "node_vcpu": node_v, + "node_mem_gib": node_m, + "gpu": int(r.get("gpu", 0)), + } + return table + + +def _strip_provider_prefix(label: str) -> tuple[str, str] | None: + """Given a HUD label ('mt-...', 'lf-...'), return (provider, osdc_runner_name). + + Handles: + * 'mt-l-x86...' -> ('mt', 'l-x86...') (new naming) + * 'lf-l-x86...' -> ('lf', 'l-x86...') + * 'mt-rel-l-...' -> ('mt', 'rel-l-...') + * 'mt-linux.2xl' -> ('mt', mapped via OLD_TO_NEW_LABEL) + * 'lf-linux.2xl' -> ('lf', mapped via OLD_TO_NEW_LABEL) + + Returns None if not a mt-/lf- label or the mapping is unknown. + """ + if label.startswith("mt-"): + provider = "mt" + elif label.startswith("lf-"): + provider = "lf" + else: + return None + rest = label.split("-", 1)[1] # drop 'mt' or 'lf' + if rest.startswith("l-") or rest.startswith("rel-l-"): + return provider, rest + # Legacy 'linux.*' names — translate via OLD_TO_NEW_LABEL. + mapped = OLD_TO_NEW_LABEL.get(rest) + if mapped is None: + return None + return provider, mapped + + +def _bucket_start(iso_str: str) -> str: + """Round an ISO timestamp DOWN to the nearest 5-min bucket.""" + # Accept 'YYYY-MM-DDTHH:MM:SS' (ClickHouse DateTime cast). + t = dt.datetime.fromisoformat(iso_str.replace("Z", "+00:00")) + if t.tzinfo is None: + t = t.replace(tzinfo=dt.UTC) + bucketed = t.replace(minute=(t.minute // 5) * 5, second=0, microsecond=0) + return bucketed.strftime("%Y-%m-%dT%H:%M:%S%z") + + +def _to_utc_str(iso_str: str) -> str: + t = dt.datetime.fromisoformat(iso_str.replace("Z", "+00:00")) + if t.tzinfo is None: + t = t.replace(tzinfo=dt.UTC) + return t.strftime("%Y-%m-%dT%H:%M:%S%z") + + +def _iter_chunk_rows(paths: list[Path]): + """Yield [label, started_at, completed_at, runtime_s] rows across all chunks.""" + for p in paths: + rows = json.loads(p.read_text()) + for row in rows: + yield row + + +def cmd_build_table(_args) -> int: + table = build_label_table() + json.dump(table, sys.stdout, indent=2, sort_keys=True) + print() + return 0 + + +def cmd_extract(args) -> int: + chunk_paths = [Path(p) for p in args.chunks] + for p in chunk_paths: + if not p.is_file(): + print(f"ERROR: chunk not found: {p}", file=sys.stderr) + return 2 + table = build_label_table() + unknown: dict[str, int] = {} + kept = 0 + skipped_runtime = 0 + skipped_unknown = 0 + + out = Path(args.out) + out.parent.mkdir(parents=True, exist_ok=True) + with out.open("w", newline="") as fh: + w = csv.writer(fh) + w.writerow(["provider", "label", "nodepool", "nodepool_fraction", "start_time", "end_time"]) + for hud_label, started, completed, runtime_s in _iter_chunk_rows(chunk_paths): + if not (MIN_RUNTIME_S <= int(runtime_s) <= MAX_RUNTIME_S): + skipped_runtime += 1 + continue + parsed = _strip_provider_prefix(hud_label) + if parsed is None or parsed[1] not in table: + unknown[hud_label] = unknown.get(hud_label, 0) + 1 + skipped_unknown += 1 + continue + provider, osdc_label = parsed + row = table[osdc_label] + w.writerow( + [ + provider, + osdc_label, + row["nodepool"], + row["nodepool_fraction"], + _bucket_start(started), + _to_utc_str(completed), + ] + ) + kept += 1 + + print(f"wrote {out}", file=sys.stderr) + print(f" kept: {kept}", file=sys.stderr) + print(f" skipped runtime: {skipped_runtime}", file=sys.stderr) + print(f" skipped unknown: {skipped_unknown}", file=sys.stderr) + if unknown: + top = sorted(unknown.items(), key=lambda kv: -kv[1])[:20] + print(f" top unknown labels ({len(unknown)} distinct):", file=sys.stderr) + for lbl, n in top: + print(f" {n:>6} {lbl}", file=sys.stderr) + return 0 + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + sub = ap.add_subparsers(dest="cmd", required=True) + + p_table = sub.add_parser("build-table", help="Print label -> nodepool mapping as JSON") + p_table.set_defaults(func=cmd_build_table) + + p_ext = sub.add_parser("extract", help="Filter + join HUD chunks -> CSV") + p_ext.add_argument("--out", required=True, help="output CSV path") + p_ext.add_argument("chunks", nargs="+", help="raw ClickHouse JSON chunk file(s)") + p_ext.set_defaults(func=cmd_extract) + + args = ap.parse_args() + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/osdc/scripts/node-size-sweep/plan.md b/osdc/scripts/node-size-sweep/plan.md new file mode 100644 index 00000000..465f090f --- /dev/null +++ b/osdc/scripts/node-size-sweep/plan.md @@ -0,0 +1,176 @@ +# Plan: real CPU + memory bin-packing in the sweep sim + +## Goal + +Replace `frac_n` (1D "fraction of biggest instance") with real 2D bin-packing on +`(cpu_millicores, memory_mib)` against per-node allocatable, computed properly +as `capacity − kubelet_reserved − daemonset_overhead`. Same for placement, +same for the util metric — matching prod's `node_compactor_node_utilization_ratio = +total_pod_requests / allocatable`. + +## What changes + +### A. Data-side (build_csv.py) + +**No CSV schema change.** The existing CSV already carries `label`, which is +enough — `build_label_table()`'s derivation logic (label → def → vcpu, memory, +instance_type, gpu, fleet) is what the sim needs, not a wider CSV. + +The stale `nodepool_fraction` column stays in the CSV but is ignored by the +sim. No re-extraction needed; the existing 200MB `pytorch_60d.csv` keeps +working. + +`build_csv.py` itself doesn't need changes for this pass. (Future cleanup: +drop `nodepool_fraction` since nothing reads it anymore.) + +### B. Sim-side (simulate.py) + +**New instance-model table** at the top of `simulate.py` (or imported from +`instance_specs.py`): +- Per instance type: `vcpu`, `memory_mib`, `gpu`. +- Per fleet name (as it appears in CSV): list of instance types Karpenter can + pick from (with weights, or in Karpenter's cheapest-first order). +- Cache `allocatable_cpu_m`, `allocatable_mem_mi`, `allocatable_gpu` per + (fleet, instance_type) after subtracting: + - Kubelet reserved: `analyze_node_utilization.kubelet_reserved(vcpu, mem_gib, max_pods)`. + Need `ENI_MAX_PODS` from `instance_specs.py` for `max_pods`. + - DaemonSet overhead: `daemonset_overhead.discover_daemonsets()` → filter + `gpu_only` by whether the fleet's instance is GPU. Sum cpu_m and mem_mib. + - Runner-hooks-warmer DaemonSet on the c7i-runner fleet only (already picked + up by the discovery walk if we point at the right root). + +Fleet-instance selection strategy for the sim (Karpenter emulation): +- For each pending pod, iterate instance types in the fleet's weighted order + (biggest weight first — that's how the current defs are configured). + Pick the first instance where `allocatable >= pod_request`. +- On placement, that instance's allocatable becomes the node's capacity. +- This is CLOSER to Karpenter's real behavior (which is + `price-capacity-optimized`) than "always biggest" — but not exact. The + simplification: we treat weight order as instance-selection order and + don't model price at all. Good enough for a first pass; can refine later. + +**Node dataclass changes**: +- Remove: `used: float` +- Add: `cpu_used_m: int`, `mem_used_mi: int`, `gpu_used: int` +- Add: `cpu_allocatable_m: int`, `mem_allocatable_mi: int`, `gpu_allocatable: int` +- Add: `instance_type: str` (for reporting + selecting overhead) + +`daemonset_frac` field stays gone. + +**Job dataclass changes**: +- Remove: `frac_n: int` +- Add: `cpu_m: int`, `mem_mi: int`, `gpu: int` (workflow pod requests) +- Add: `runner_cpu_m: int`, `runner_mem_mi: int` (runner pod requests, same + every row, but easier to keep with the job than as globals) + +`load_jobs` now reads the new columns. Runner pod becomes a synthesized paired +Job on the c7i-runner fleet with `cpu_m=runner_cpu_m`, `mem_mi=runner_mem_mi`, +`gpu=0`. Same lifetime as workflow job (unchanged behavior). + +**Placeholder dataclass changes**: +- Remove: `frac_n: int` +- Add: `cpu_m: int`, `mem_mi: int`, `gpu: int` (must match the job it reserves for) + +**`_place_free` / `_preempt_placeholder`**: +- Instead of `1.0 - n.used + EPS >= 1/frac_n`, check + `n.cpu_allocatable_m - n.cpu_used_m >= job.cpu_m` AND + `n.mem_allocatable_mi - n.mem_used_mi >= job.mem_mi` AND + `n.gpu_allocatable - n.gpu_used >= job.gpu`. +- MostAllocated: score = `max(cpu_used_frac, mem_used_frac, gpu_used_frac)` + where each is `used / allocatable`. Pick highest-scoring node that fits. + This matches Kubernetes MostAllocated (which uses max of resource + usage ratios, weighted). +- Preemption matches by exact (cpu_m, mem_mi, gpu) tuple (a placeholder must + match the job it's reserving for). Not "matching frac_n" anymore. + +**Fresh node creation** (step 2b, step 3, step 4c): +- Take the pending pod's `(cpu_m, mem_mi, gpu, fleet)`. +- Ask a `pick_instance(fleet, cpu_m, mem_mi, gpu)` helper for the smallest + instance that fits (Karpenter emulation). +- Materialize the Node with that instance's allocatable values. +- If NO instance in the fleet fits: sim error (this shouldn't happen with + real HUD data — but assert loudly so we know). + +**Step 6 metric**: +- Per-node util = `max(cpu_used/cpu_allocatable, mem_used/mem_allocatable, + gpu_used/gpu_allocatable)`. This matches the "which resource is most + constrained" view. +- Alternative simpler: report BOTH CPU and memory util separately (as prod + does — two panels). Prefer this. Two metrics: `cluster_cpu_util`, + `cluster_mem_util`. GPU util reported per-pool only. +- Aggregation for the cluster-wide line: allocatable-weighted mean (matching + prod's dashboard PromQL). Per-pool same. + +**Step 6 output**: report both CPU and memory average + p10..p90. Per-pool +table gets extra columns for cpu/mem/gpu util. + +### C. What DOESN'T change + +- Warmup model: unchanged. Warming nodes still filter out of `_place_free` + workload placement; still counted in denominator by their (empty) used. +- Placeholder timing: unchanged. +- 5-min bucketing: unchanged. +- Karpenter consolidation TTL: unchanged (`empty_ttl_buckets=2`). +- Shuffle: unchanged. +- CLI flags: `--no-warmup`, `--warmup-buckets*`, `--drop-provider`, + `--keep-fraction`, `--seed` all stay. +- CSV path stays a positional arg. + +### D. New CLI + +None needed. All the new data comes from the CSV. If we want to A/B against +the old model, keep old sim as `simulate.py.bkp` locally. + +## File impact + +- `build_csv.py`: substantial rewrite of `build_label_table`, `cmd_extract`. + CSV schema breaks (old CSVs won't work). +- `simulate.py`: substantial rewrite of Node, Job, Placeholder, `_place_free`, + `_preempt_placeholder`, `simulate()`, `report()`. ~200 lines changed. +- Need to import from `scripts/python/`: `instance_specs`, `daemonset_overhead`, + `analyze_node_utilization` (for `kubelet_reserved`), `runner_overhead`. + Already done in `build_csv.py` — extend the same pattern to `simulate.py`. + +## Fleet-to-instances mapping + +`build_csv.py`'s `derive_fleet_name(instance, node_fleet)` already picks the +fleet name for a runner def. But the sim needs the REVERSE: given a fleet +name (e.g. `c7i-runner`, `g5`), which instance types can Karpenter pick? + +Load from `modules/nodepools/defs/*.yaml` — these have the list of instance +types + weights. `scripts/python/nodepool_defs.py` probably already parses +these; if so, reuse. Otherwise, small YAML walker. + +## Consequences to expect + +- Cluster util WILL drop from the current sim number. Whole-node r7a jobs that + used to report 100% will now report `worker_cpu / (allocatable - DS - kubelet)` + = ~95-97%. That alone gets us closer to prod. +- Per-fleet packing accuracy improves. GPU fleets will show whichever resource + (CPU vs memory vs GPU) is actually the bottleneck — often not CPU. +- Node counts will change per fleet: fleets whose defs are memory-heavy will + provision more nodes (fewer fit per instance); CPU-heavy fewer nodes. +- Placeholders now match jobs on the tuple (cpu_m, mem_mi, gpu), so preemption + is slightly stricter. Should still work fine on real data because + placeholders are created from the same defs as jobs. + +## Validation plan + +1. Old sim on old CSV → baseline number (current: ~72.6% default warmup on + sample). +2. New sim on new CSV, `--no-warmup` → should be reasonably close to prod's + 61% since we've now removed the tile-perfection assumption AND added + real overhead accounting. +3. New sim with warmup → should be even lower. +4. Per-pool comparison against prod `dash.json` per-fleet breakdown. +5. Spot-check one pool: pick c7i, count max_nodes vs prod's observed peak. + +## Non-goals for THIS pass + +- Karpenter's actual `price-capacity-optimized` — we use "biggest weight + first" as approximation. +- Spare-capacity floor (compactor doesn't terminate, just taints — we're + not modeling that either yet). +- Batching latency (Karpenter batches for ~45s before provisioning). +- `karpenter.sh/do-not-disrupt` pinning of nodes. +- Multi-instance-type fleet load-balancing across sizes based on pod shape. diff --git a/osdc/scripts/node-size-sweep/pull_hud.py b/osdc/scripts/node-size-sweep/pull_hud.py new file mode 100644 index 00000000..e9fb1ffa --- /dev/null +++ b/osdc/scripts/node-size-sweep/pull_hud.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python3 +# /// script +# requires-python = ">=3.12" +# dependencies = ["requests>=2.31"] +# /// +"""Pull pytorch/pytorch workflow_job rows from HUD ClickHouse, weekly chunks. + +Hits ClickHouse Cloud directly. Reads credentials from env, in order of +precedence: + + CLICKHOUSE_URL / CLICKHOUSE_HUD_USER_URL + CLICKHOUSE_USER / CLICKHOUSE_HUD_USER_USERNAME + CLICKHOUSE_PASSWORD / CLICKHOUSE_HUD_USER_PASSWORD + +Writes one JSON file per week to /chunk__.json. +Each file is the raw list-of-lists returned by ClickHouse JSONCompact: + [ [label, started_at_iso, completed_at_iso, runtime_s], ... ] + +That's exactly what build_csv.py's `extract` subcommand consumes. + +Usage: + # 60 days back from now, 7-day windows + uv run pull_hud.py --out /tmp/pytorch_workload/raw --days 60 --window-days 7 + + # Explicit range + uv run pull_hud.py --out /tmp/pytorch_workload/raw \\ + --start 2026-05-01 --end 2026-07-01 --window-days 7 + + # Dry run: print the windows we WOULD fetch and skip the HTTP calls + uv run pull_hud.py --out /tmp/pytorch_workload/raw --days 60 --dry-run + +If a chunk file already exists and is non-empty, we skip it (idempotent). Pass +--force to re-fetch. +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import json +import os +import sys +import time +from pathlib import Path + +import requests + +DEFAULT_URL_ENV = "CLICKHOUSE_URL" +DEFAULT_USER_ENV = "CLICKHOUSE_USER" +DEFAULT_PW_ENV = "CLICKHOUSE_PASSWORD" + +FALLBACK_URL_ENV = "CLICKHOUSE_HUD_USER_URL" +FALLBACK_USER_ENV = "CLICKHOUSE_HUD_USER_USERNAME" +FALLBACK_PW_ENV = "CLICKHOUSE_HUD_USER_PASSWORD" + + +def _resolve_creds() -> tuple[str, str, str]: + """Read CH creds from env with a HUD_USER_* fallback. Exits on missing.""" + url = os.environ.get(DEFAULT_URL_ENV) or os.environ.get(FALLBACK_URL_ENV) + user = os.environ.get(DEFAULT_USER_ENV) or os.environ.get(FALLBACK_USER_ENV) + pw = os.environ.get(DEFAULT_PW_ENV) or os.environ.get(FALLBACK_PW_ENV) + missing = [] + if not url: + missing.append(f"{DEFAULT_URL_ENV} / {FALLBACK_URL_ENV}") + if not user: + missing.append(f"{DEFAULT_USER_ENV} / {FALLBACK_USER_ENV}") + if not pw: + missing.append(f"{DEFAULT_PW_ENV} / {FALLBACK_PW_ENV}") + if missing: + print(f"ERROR: missing env vars: {'; '.join(missing)}", file=sys.stderr) + sys.exit(2) + return url, user, pw + +# The query is intentionally kept simple and identical across chunks so any +# schema/filter change lives in ONE place. See build_csv.py for what happens +# to the returned rows. +QUERY_TEMPLATE = """ +SELECT + label, + toString(started_at, 'UTC') AS started_at_s, + toString(completed_at, 'UTC') AS completed_at_s, + runtime_s +FROM ( + SELECT + started_at, + completed_at, + dateDiff('second', started_at, completed_at) AS runtime_s, + arrayFilter(x -> startsWith(x, 'mt-') OR startsWith(x, 'lf-'), labels) AS mt_lf_labels + FROM default.workflow_job + WHERE started_at >= toDateTime64('{start}', 3, 'UTC') + AND started_at < toDateTime64('{end}', 3, 'UTC') + AND repository_full_name = 'pytorch/pytorch' + AND status = 'completed' + AND conclusion != '' + AND arrayExists(x -> startsWith(x, 'mt-') OR startsWith(x, 'lf-'), labels) + AND dateDiff('second', started_at, completed_at) BETWEEN 1 AND 21600 +) +ARRAY JOIN mt_lf_labels AS label +FORMAT JSONCompact +""" + + +def _iso_date(s: str) -> dt.date: + return dt.date.fromisoformat(s) + + +def _windows(start: dt.date, end: dt.date, window_days: int) -> list[tuple[dt.date, dt.date]]: + """Return [(win_start, win_end), ...] non-overlapping, half-open [start, end).""" + if window_days <= 0: + raise ValueError("window-days must be positive") + out = [] + cur = start + while cur < end: + nxt = min(cur + dt.timedelta(days=window_days), end) + out.append((cur, nxt)) + cur = nxt + return out + + +def _chunk_path(out_dir: Path, win_start: dt.date, win_end: dt.date) -> Path: + return out_dir / f"chunk_{win_start.isoformat()}_{win_end.isoformat()}.json" + + +def _fmt_ts(d: dt.date) -> str: + """Format a date as 'YYYY-MM-DD HH:MM:SS.000' for toDateTime64() in CH.""" + return f"{d.isoformat()} 00:00:00.000" + + +def _run_query(url: str, user: str, password: str, sql: str, timeout_s: int) -> dict: + """POST the query and return the parsed JSON body.""" + r = requests.post( + url, + params={"database": "default"}, + data=sql, + auth=(user, password), + timeout=timeout_s, + headers={"Content-Type": "text/plain; charset=utf-8"}, + ) + r.raise_for_status() + return r.json() + + +def _extract_rows(payload: dict) -> list[list]: + """ClickHouse JSONCompact -> plain list of rows. + + JSONCompact returns: + {"meta":[...], "data":[[...],[...]], "rows":N, "statistics":{...}} + """ + rows = payload.get("data") + if not isinstance(rows, list): + raise ValueError(f"unexpected response shape: keys={list(payload)}") + return rows + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--out", required=True, help="output directory for chunk files") + grp = ap.add_mutually_exclusive_group() + grp.add_argument("--days", type=int, default=60, help="days back from today (default 60)") + ap.add_argument("--start", type=_iso_date, help="explicit start date YYYY-MM-DD (UTC)") + ap.add_argument("--end", type=_iso_date, help="explicit end date YYYY-MM-DD (UTC, exclusive)") + ap.add_argument("--window-days", type=int, default=7, help="chunk size (default 7)") + ap.add_argument("--force", action="store_true", help="re-fetch even if chunk exists") + ap.add_argument("--dry-run", action="store_true", help="print the plan and exit") + ap.add_argument("--timeout", type=int, default=300, help="per-request timeout in seconds") + ap.add_argument("--sleep", type=float, default=0.0, help="sleep between chunks (throttle)") + args = ap.parse_args() + + today = dt.datetime.now(dt.UTC).date() + if args.start and args.end: + start, end = args.start, args.end + elif args.start or args.end: + print("ERROR: --start and --end must be given together", file=sys.stderr) + return 2 + else: + end = today + start = end - dt.timedelta(days=args.days) + if start >= end: + print(f"ERROR: start {start} >= end {end}", file=sys.stderr) + return 2 + + out_dir = Path(args.out) + out_dir.mkdir(parents=True, exist_ok=True) + + wins = _windows(start, end, args.window_days) + print(f"plan: {len(wins)} weekly chunk(s) covering [{start}, {end})", file=sys.stderr) + for a, b in wins: + print(f" {a} -> {b} ({_chunk_path(out_dir, a, b).name})", file=sys.stderr) + + if args.dry_run: + return 0 + + url, user, pw = _resolve_creds() + + for a, b in wins: + path = _chunk_path(out_dir, a, b) + if path.exists() and path.stat().st_size > 0 and not args.force: + print(f"skip (exists): {path.name}", file=sys.stderr) + continue + sql = QUERY_TEMPLATE.format(start=_fmt_ts(a), end=_fmt_ts(b)) + t0 = time.monotonic() + try: + payload = _run_query(url, user, pw, sql, args.timeout) + except requests.HTTPError as e: + body = e.response.text[:500] if e.response is not None else "" + print(f"ERROR: {path.name}: HTTP {e.response.status_code if e.response is not None else '?'}: {body}", file=sys.stderr) + return 3 + rows = _extract_rows(payload) + path.write_text(json.dumps(rows)) + dt_s = time.monotonic() - t0 + print(f"wrote {path.name} rows={len(rows):>7d} {dt_s:.1f}s", file=sys.stderr) + if args.sleep > 0: + time.sleep(args.sleep) + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/osdc/scripts/node-size-sweep/sim_load.py b/osdc/scripts/node-size-sweep/sim_load.py new file mode 100644 index 00000000..99081357 --- /dev/null +++ b/osdc/scripts/node-size-sweep/sim_load.py @@ -0,0 +1,134 @@ +"""Load workload CSV rows into Job objects with runner-pod overhead attached.""" + +from __future__ import annotations + +import csv +import datetime as dt +import random +import sys +from collections import defaultdict +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(REPO_ROOT / "scripts" / "python")) + +from build_csv import build_label_table # noqa: E402 +from analyze_node_utilization import HOOKS_OVERHEAD_CPU_M as FALLBACK_HOOKS_CPU_M # noqa: E402 +from analyze_node_utilization import HOOKS_OVERHEAD_MEM_MI as FALLBACK_HOOKS_MEM_MI # noqa: E402 +from sim_nodes import Job # noqa: E402 + +BUCKET_SEC = 300 + +RUNNER_POD_POOL = "c7i-runner" +RUNNER_POD_LABEL = "runner-pod" + +FALLBACK_RUNNER_CPU_M = 750 +FALLBACK_RUNNER_MEM_MI = 1024 + + +def _bucket(ts: dt.datetime) -> int: + return (int(ts.timestamp()) // BUCKET_SEC) * BUCKET_SEC + + +def _load_runner_overhead() -> tuple[int, int, int, int]: + """Return (workflow_extra_cpu_m, workflow_extra_mem_mi, runner_cpu_m, runner_mem_mi).""" + try: + from runner_overhead import load_runner_pod_overhead + + overhead = load_runner_pod_overhead(REPO_ROOT) + return ( + overhead.workflow_extra_cpu_m or FALLBACK_HOOKS_CPU_M, + overhead.workflow_extra_mem_mi or FALLBACK_HOOKS_MEM_MI, + overhead.runner_cpu_m, + overhead.runner_mem_mi, + ) + except Exception as e: + print( + f"warning: falling back to hooks/runner constants ({type(e).__name__}: {e})", + file=sys.stderr, + ) + return ( + FALLBACK_HOOKS_CPU_M, + FALLBACK_HOOKS_MEM_MI, + FALLBACK_RUNNER_CPU_M, + FALLBACK_RUNNER_MEM_MI, + ) + + +def load_jobs( + csv_path: Path, + add_runner_pods: bool = True, + runner_pool: str = RUNNER_POD_POOL, + drop_providers: set[str] | None = None, + keep_fraction: float = 1.0, + keep_seed: int = 12345, +) -> list[Job]: + if drop_providers is None: + drop_providers = set() + if not (0.0 < keep_fraction <= 1.0): + raise ValueError(f"keep_fraction must be in (0, 1], got {keep_fraction}") + rng = random.Random(keep_seed) + + label_table = build_label_table() + hooks_cpu, hooks_mem, runner_cpu, runner_mem = _load_runner_overhead() + + jobs: list[Job] = [] + dropped_provider = 0 + dropped_downsample = 0 + dropped_unknown_label: dict[str, int] = defaultdict(int) + with csv_path.open() as fh: + reader = csv.DictReader(fh) + has_provider = "provider" in (reader.fieldnames or []) + for row in reader: + if has_provider and row["provider"] in drop_providers: + dropped_provider += 1 + continue + if keep_fraction < 1.0 and rng.random() >= keep_fraction: + dropped_downsample += 1 + continue + label = row["label"] + spec = label_table.get(label) + if spec is None: + dropped_unknown_label[label] += 1 + continue + st = dt.datetime.fromisoformat(row["start_time"]) + en = dt.datetime.fromisoformat(row["end_time"]) + sb = _bucket(st) + eb = _bucket(en) + cpu_m = int(spec["vcpu"] * 1000) + hooks_cpu + mem_mi = int(spec["memory_gib"] * 1024) + hooks_mem + gpu = int(spec["gpu"]) + jobs.append( + Job( + label=label, + pool=spec["nodepool"], + cpu_m=cpu_m, + mem_mi=mem_mi, + gpu=gpu, + start_bucket=sb, + end_bucket=eb, + ) + ) + if add_runner_pods: + jobs.append( + Job( + label=RUNNER_POD_LABEL, + pool=runner_pool, + cpu_m=runner_cpu, + mem_mi=runner_mem, + gpu=0, + start_bucket=sb, + end_bucket=eb, + ) + ) + if dropped_provider: + print(f" dropped by provider filter: {dropped_provider:,}", file=sys.stderr) + if dropped_downsample: + print(f" dropped by downsample: {dropped_downsample:,}", file=sys.stderr) + if dropped_unknown_label: + total = sum(dropped_unknown_label.values()) + print( + f" dropped unknown labels: {total:,} ({len(dropped_unknown_label)} distinct)", + file=sys.stderr, + ) + return jobs diff --git a/osdc/scripts/node-size-sweep/sim_nodes.py b/osdc/scripts/node-size-sweep/sim_nodes.py new file mode 100644 index 00000000..bf799d0e --- /dev/null +++ b/osdc/scripts/node-size-sweep/sim_nodes.py @@ -0,0 +1,245 @@ +"""Data model + allocatable/instance-selection helpers for the sweep simulator. + +Loads fleet instance lists from `modules/nodepools*/defs/*.yaml`, computes per +(fleet, instance_type) allocatable capacity net of kubelet + DaemonSet overhead, +and picks the highest-weight instance in a fleet that fits a pod request. +""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass, field +from pathlib import Path + +import yaml + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(REPO_ROOT / "scripts" / "python")) + +from analyze_node_utilization import compute_allocatable, compute_daemonset_overhead # noqa: E402, F401 +from daemonset_overhead import DaemonSetOverhead, discover_daemonsets # noqa: E402 +from fleet_naming import derive_fleet_name # noqa: E402 +from instance_specs import INSTANCE_SPECS # noqa: E402 + +NODEPOOL_DEFS_DIRS = [ + REPO_ROOT / "modules" / "nodepools" / "defs", + REPO_ROOT / "modules" / "nodepools-h100" / "defs", + REPO_ROOT / "modules" / "nodepools-b200" / "defs", +] + +FLEET_SCOPED_DAEMONSETS: dict[str, set[str]] = { + "runner-hooks-warmer": {"c7i-runner"}, +} + + +def _daemonsets_for_fleet(daemonsets: list[DaemonSetOverhead], fleet_name: str) -> list[DaemonSetOverhead]: + """Filter DaemonSets whose nodeSelector restricts them to specific fleets. + + discover_daemonsets doesn't understand nodeSelector, so fleet-scoped DSes + like runner-hooks-warmer (node-fleet=c7i-runner) would otherwise be counted + on every fleet's allocatable calculation. + """ + return [ + ds + for ds in daemonsets + if fleet_name in FLEET_SCOPED_DAEMONSETS.get(ds.name, {fleet_name}) + ] + + +@dataclass(slots=True) +class Job: + label: str + pool: str + cpu_m: int + mem_mi: int + gpu: int + start_bucket: int + end_bucket: int + + +@dataclass(slots=True) +class Placeholder: + cpu_m: int + mem_mi: int + gpu: int + created_bucket: int + + +@dataclass(slots=True) +class Node: + pool: str + instance_type: str + cpu_allocatable_m: int + mem_allocatable_mi: int + gpu_allocatable: int + cpu_used_m: int = 0 + mem_used_mi: int = 0 + gpu_used: int = 0 + live_jobs: int = 0 + placeholders: list[Placeholder] = field(default_factory=list) + empty_since_bucket: int | None = None + warming_until_bucket: int | None = None + daemonset_cpu_m: int = 0 + daemonset_mem_mi: int = 0 + daemonset_gpu: int = 0 + phantom_cpu_m: int = 0 + phantom_mem_mi: int = 0 + phantom_gpu: int = 0 + + +@dataclass(frozen=True) +class Allocatable: + cpu_m: int + mem_mi: int + gpu: int + + +@dataclass(frozen=True) +class FleetSpec: + """A fleet's available instance types, ordered highest-weight first.""" + + name: str + is_gpu: bool + instances: tuple[str, ...] + + +def _load_fleet_specs(defs_dirs: list[Path]) -> dict[str, FleetSpec]: + """Read every nodepool def YAML and return {fleet_name: FleetSpec}. + + Ignores `exclude_regions` — the sim is region-agnostic. + """ + fleets: dict[str, FleetSpec] = {} + nodepool_files: list[Path] = [] + for defs_dir in defs_dirs: + if not defs_dir.is_dir(): + continue + for f in sorted(defs_dir.glob("*.yaml")): + data = yaml.safe_load(f.read_text()) or {} + fleet = data.get("fleet") + if isinstance(fleet, dict): + name = fleet.get("name") + if not isinstance(name, str): + continue + instances = fleet.get("instances") or [] + ordered = sorted( + (i for i in instances if isinstance(i, dict) and i.get("type")), + key=lambda i: -int(i.get("weight", 0)), + ) + types = tuple(i["type"] for i in ordered) + if not types: + continue + fleets[name] = FleetSpec(name=name, is_gpu=bool(fleet.get("gpu", False)), instances=types) + continue + if isinstance(data.get("nodepool"), dict): + nodepool_files.append(f) + + for f in nodepool_files: + data = yaml.safe_load(f.read_text()) or {} + nodepool = data.get("nodepool") or {} + instance_type = nodepool.get("instance_type") + if not isinstance(instance_type, str): + continue + # nodepool: defs are single-instance pools keyed by taint, not by node-fleet + # label; expose them as fleets under their instance-family name so the sim's + # fleet lookup (which uses derive_fleet_name) can resolve them. + derived = derive_fleet_name(instance_type) + if derived in fleets: + print( + f"warning: nodepool def {f} would collide with existing fleet " + f"{derived!r}; keeping explicit fleet: definition", + file=sys.stderr, + ) + continue + fleets[derived] = FleetSpec( + name=derived, + is_gpu=bool(nodepool.get("gpu", False)), + instances=(instance_type,), + ) + return fleets + + +class ClusterModel: + """Holds fleet specs + cached per-(fleet, instance) allocatable.""" + + def __init__(self, defs_dirs: list[Path] | None = None, upstream_dir: Path = REPO_ROOT): + if defs_dirs is None: + defs_dirs = NODEPOOL_DEFS_DIRS + self.fleets = _load_fleet_specs(defs_dirs) + self.daemonsets = discover_daemonsets(upstream_dir) + self._alloc_cache: dict[tuple[str, str], Allocatable] = {} + self._ds_cache: dict[tuple[str, str], tuple[int, int, int]] = {} + + def allocatable(self, fleet: str, instance_type: str) -> Allocatable: + key = (fleet, instance_type) + cached = self._alloc_cache.get(key) + if cached is not None: + return cached + scoped_ds = _daemonsets_for_fleet(self.daemonsets, fleet) + info = compute_allocatable(instance_type, scoped_ds) + alloc = Allocatable( + cpu_m=info["allocatable_cpu_m"], + mem_mi=info["allocatable_mem_mi"], + gpu=info["allocatable_gpu"], + ) + self._alloc_cache[key] = alloc + return alloc + + def daemonset_totals(self, fleet: str, instance_type: str) -> tuple[int, int, int]: + key = (fleet, instance_type) + cached = self._ds_cache.get(key) + if cached is not None: + return cached + spec = INSTANCE_SPECS.get(instance_type) + is_gpu = bool(spec and spec.get("gpu", 0) > 0) + scoped_ds = _daemonsets_for_fleet(self.daemonsets, fleet) + ds_cpu, ds_mem = compute_daemonset_overhead(scoped_ds, is_gpu=is_gpu) + totals = (ds_cpu, ds_mem, 0) + self._ds_cache[key] = totals + return totals + + def pick_instance(self, fleet: str, cpu_m: int, mem_mi: int, gpu: int) -> str: + """Highest-weight instance in `fleet` whose allocatable fits the pod.""" + fs = self.fleets.get(fleet) + if fs is None: + raise RuntimeError(f"unknown fleet {fleet!r} (not in modules/nodepools*/defs)") + for inst in fs.instances: + if inst not in INSTANCE_SPECS: + continue + alloc = self.allocatable(fleet, inst) + if alloc.cpu_m >= cpu_m and alloc.mem_mi >= mem_mi and alloc.gpu >= gpu: + return inst + raise RuntimeError( + f"no instance in fleet {fleet!r} fits pod " + f"(cpu_m={cpu_m}, mem_mi={mem_mi}, gpu={gpu}); tried {fs.instances}" + ) + + def make_node(self, pool: str, cpu_m: int, mem_mi: int, gpu: int) -> Node: + inst = self.pick_instance(pool, cpu_m, mem_mi, gpu) + alloc = self.allocatable(pool, inst) + ds_cpu, ds_mem, ds_gpu = self.daemonset_totals(pool, inst) + return Node( + pool=pool, + instance_type=inst, + cpu_allocatable_m=alloc.cpu_m, + mem_allocatable_mi=alloc.mem_mi, + gpu_allocatable=alloc.gpu, + daemonset_cpu_m=ds_cpu, + daemonset_mem_mi=ds_mem, + daemonset_gpu=ds_gpu, + ) + + +def fits(node: Node, cpu_m: int, mem_mi: int, gpu: int) -> bool: + return ( + node.cpu_allocatable_m - node.cpu_used_m >= cpu_m + and node.mem_allocatable_mi - node.mem_used_mi >= mem_mi + and node.gpu_allocatable - node.gpu_used >= gpu + ) + + +def most_allocated_score(node: Node) -> float: + """MostAllocated ranking: highest of (cpu, mem, gpu) usage ratios.""" + cpu_frac = node.cpu_used_m / node.cpu_allocatable_m if node.cpu_allocatable_m > 0 else 0.0 + mem_frac = node.mem_used_mi / node.mem_allocatable_mi if node.mem_allocatable_mi > 0 else 0.0 + gpu_frac = node.gpu_used / node.gpu_allocatable if node.gpu_allocatable > 0 else 0.0 + return max(cpu_frac, mem_frac, gpu_frac) diff --git a/osdc/scripts/node-size-sweep/sim_report.py b/osdc/scripts/node-size-sweep/sim_report.py new file mode 100644 index 00000000..a77b8e7c --- /dev/null +++ b/osdc/scripts/node-size-sweep/sim_report.py @@ -0,0 +1,145 @@ +"""Reporting for the sweep simulator: aggregation + pretty-printed output.""" + +from __future__ import annotations + +import statistics +from collections import defaultdict + + +def percentiles(vals: list[float], pcts: list[int]) -> dict[int, float]: + if not vals: + return {p: float("nan") for p in pcts} + sv = sorted(vals) + n = len(sv) + return {p: sv[max(0, min(n - 1, int(round(p / 100 * (n - 1)))))] for p in pcts} + + +def _weighted_series(per_bucket, used_key: str, alloc_key: str) -> list[float]: + out = [] + for _, per_pool in per_bucket: + num = 0 + den = 0 + for _name, sums in per_pool.items(): + num += sums[used_key] + den += sums[alloc_key] + if den > 0: + out.append(num / den) + return out + + +def _unweighted_series(per_bucket, used_key: str, alloc_key: str) -> list[float]: + out = [] + for _, per_pool in per_bucket: + ratios = [] + for _name, sums in per_pool.items(): + if sums[alloc_key] > 0: + ratios.append(sums[used_key] / sums[alloc_key]) + if ratios: + out.append(sum(ratios) / len(ratios)) + return out + + +def report(sim_out: dict) -> None: + per_bucket = sim_out["per_bucket"] + pool_max = sim_out["pool_max_nodes"] + pool_max_warming = sim_out.get("pool_max_warming", {}) + pool_created = sim_out["pool_total_created"] + pool_placeholders = sim_out["pool_total_placeholders"] + pool_preempted = sim_out["pool_total_preempted"] + pool_expired = sim_out["pool_total_expired"] + flags = sim_out.get("flags", {}) + + pcts = [10, 20, 30, 40, 50, 60, 70, 80, 90] + + print("==== Simulation flags ====") + print(f" daemonsets_in_metric: {'on' if flags.get('daemonsets_in_metric') else 'off'}") + if flags.get("phantom_pods_enabled"): + print( + f" phantom_pods: on(lookahead={flags.get('phantom_lookahead_buckets')}, " + f"cap={flags.get('phantom_cap'):.2f})" + ) + else: + print(" phantom_pods: off") + print(f" placeholders: {'on' if flags.get('placeholders_enabled') else 'off'}") + print( + f" warmup: default:{flags.get('warmup_buckets_default')}/" + f"gpu:{flags.get('warmup_buckets_gpu')}/" + f"baremetal:{flags.get('warmup_buckets_baremetal')}" + ) + print(f" empty_ttl_buckets: {flags.get('empty_ttl_buckets')}") + print(f" placeholder_max_age: {flags.get('placeholder_max_age')}") + + def _print_series(title: str, series: list[float]) -> None: + print(f"\n==== Cluster-wide {title} ====") + print(f" buckets measured: {len(series):,}") + if not series: + print(" (no data)") + return + print(f" average: {statistics.mean(series):.1%}") + p = percentiles(series, pcts) + print(" " + " ".join(f"p{q:>2}: {p[q]:.1%}" for q in pcts)) + + weighted_label = ( + "allocatable-weighted, matches prod PromQL" + if flags.get("daemonsets_in_metric") + else "allocatable-weighted (excludes DaemonSets)" + ) + _print_series( + f"CPU utilization ({weighted_label})", + _weighted_series(per_bucket, "cpu_used_m", "cpu_alloc_m"), + ) + _print_series( + "CPU utilization (unweighted mean of per-pool ratios)", + _unweighted_series(per_bucket, "cpu_used_m", "cpu_alloc_m"), + ) + _print_series( + f"memory utilization ({weighted_label})", + _weighted_series(per_bucket, "mem_used_mi", "mem_alloc_mi"), + ) + _print_series( + "memory utilization (unweighted mean of per-pool ratios)", + _unweighted_series(per_bucket, "mem_used_mi", "mem_alloc_mi"), + ) + + pool_cpu: dict[str, list[float]] = defaultdict(list) + pool_mem: dict[str, list[float]] = defaultdict(list) + pool_gpu: dict[str, list[float]] = defaultdict(list) + for _, per_pool in per_bucket: + for name, sums in per_pool.items(): + if sums["cpu_alloc_m"] > 0: + pool_cpu[name].append(sums["cpu_used_m"] / sums["cpu_alloc_m"]) + else: + pool_cpu[name].append(0.0) + if sums["mem_alloc_mi"] > 0: + pool_mem[name].append(sums["mem_used_mi"] / sums["mem_alloc_mi"]) + else: + pool_mem[name].append(0.0) + if sums["gpu_alloc"] > 0: + pool_gpu[name].append(sums["gpu_used"] / sums["gpu_alloc"]) + + print("\n==== Per-pool utilization (placeholders included) ====") + header = ( + f" {'pool':<16} {'buckets':>8} {'max_nodes':>10} {'warming_max':>12} {'created':>8} " + f"{'ph_new':>7} {'ph_pre':>7} {'ph_exp':>7} " + f"{'cpu_avg':>8} {'cpu_p50':>8} {'cpu_p90':>8} " + f"{'mem_avg':>8} {'mem_p50':>8} {'mem_p90':>8} " + f"{'gpu_avg':>8} {'gpu_p50':>8} {'gpu_p90':>8}" + ) + print(header) + print(" " + "-" * (len(header) - 2)) + for name in sorted(pool_cpu, key=lambda n: -statistics.mean(pool_cpu[n])): + cpu = pool_cpu[name] + mem = pool_mem[name] + gpu = pool_gpu.get(name, []) + cp = percentiles(cpu, pcts) + mp = percentiles(mem, pcts) + gp = percentiles(gpu, pcts) if gpu else {50: float("nan"), 90: float("nan")} + gpu_avg = statistics.mean(gpu) if gpu else float("nan") + print( + f" {name:<16} {len(cpu):>8} {pool_max.get(name, 0):>10} {pool_max_warming.get(name, 0):>12} " + f"{pool_created.get(name, 0):>8} " + f"{pool_placeholders.get(name, 0):>7} {pool_preempted.get(name, 0):>7} {pool_expired.get(name, 0):>7} " + f"{statistics.mean(cpu):>7.1%} {cp[50]:>7.1%} {cp[90]:>7.1%} " + f"{statistics.mean(mem):>7.1%} {mp[50]:>7.1%} {mp[90]:>7.1%} " + f"{gpu_avg:>7.1%} {gp[50]:>7.1%} {gp[90]:>7.1%}" + ) diff --git a/osdc/scripts/node-size-sweep/simulate.py b/osdc/scripts/node-size-sweep/simulate.py new file mode 100644 index 00000000..c04b7370 --- /dev/null +++ b/osdc/scripts/node-size-sweep/simulate.py @@ -0,0 +1,554 @@ +#!/usr/bin/env python3 +# /// script +# requires-python = ">=3.12" +# dependencies = ["pyyaml>=6.0"] +# /// +"""Replay a workload CSV onto a simulated cluster, bucket by bucket. + +Step 3 (placeholder top-up) and step 4b (placeholder preemption) are gated by +``placeholders_enabled``; when False, jobs go directly from 4a to 4c. + +Read a CSV of jobs (label, nodepool, nodepool_fraction, start_time, end_time), +join each label to its runner def (vcpu, memory, gpu, fleet, instance_type), +and simulate a Karpenter-style bin-packed cluster on real 3D +(cpu_millicores, memory_mib, gpu) allocatable capacity. + +Per-node allocatable = instance_capacity - kubelet_reserved - daemonset_overhead. +Karpenter emulation: for each pod, pick the highest-weight instance in the +fleet whose allocatable fits the request. + +Fresh nodes are modeled as `warming` for a per-pool window (default 5 min, GPU +10 min, baremetal 15 min) during which they cannot accept workload jobs. +Look-ahead pre-provisioning creates warming nodes for future arrivals so jobs +still start on-time via the normal placement path once the node flips to warm. + +Model, per 5-min bucket iterated contiguously: + 0. Promote warming nodes whose warming_until_bucket <= bucket_idx. + 1. Expire placeholders older than PLACEHOLDER_MAX_AGE_BUCKETS (default 2). + 2. Deprovision finishers whose end_bucket == this bucket. + 2b. Look-ahead: create warming nodes for future arrivals that will not fit + the currently-free capacity. + 3. Top up placeholders per (pool, cpu_m, mem_mi, gpu) to match arrivals. + Each new placeholder is bin-packed via MostAllocated on free space + (including warming nodes — placeholders are pause containers with no + wait-for-hooks gate); creates a new node if no candidate has room. + 4. Place arrivals in shuffled order. For each job: + a. MostAllocated on WARM nodes with truly-free 3D room. + b. Else preempt a matching-shape placeholder on the MostAllocated + warm node. + c. Else create a new WARM node (unforecasted demand cannot be deferred). + 5. Deprovision same-bucket start=end finishers. + 6. Measure: allocatable-weighted CPU and memory util per pool + cluster, + GPU util per pool where applicable. + 7. Consolidate: drop nodes empty >= KARPENTER_EMPTY_TTL_BUCKETS. + +Every workload job in the CSV is paired with an ARC runner pod on the +c7i-runner fleet (750m/1Gi), same lifetime. Placeholders apply to both. + +Usage: + uv run simulate.py path/to/workload.csv --seed 42 +""" + +from __future__ import annotations + +import argparse +import random +import sys +from collections import defaultdict +from datetime import UTC, datetime +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(REPO_ROOT / "scripts" / "python")) + +from sim_load import RUNNER_POD_POOL, load_jobs # noqa: E402 +from sim_nodes import ClusterModel, Job, Node, Placeholder, fits, most_allocated_score # noqa: E402 +from sim_report import report # noqa: E402 + +BUCKET_SEC = 300 +KARPENTER_EMPTY_TTL_BUCKETS = 2 +PLACEHOLDER_MAX_AGE_BUCKETS = 2 + +WARMUP_BUCKETS_DEFAULT = 1 +WARMUP_BUCKETS_GPU = 2 +WARMUP_BUCKETS_BAREMETAL = 3 +GPU_POOL_PREFIXES = ("g5", "g6", "g4dn") +BAREMETAL_POOL_PREFIXES = ("p4d", "p5", "p6-b200") + + +def _pool_class(pool: str) -> str: + for prefix in BAREMETAL_POOL_PREFIXES: + if pool.startswith(prefix): + return "baremetal" + for prefix in GPU_POOL_PREFIXES: + if pool.startswith(prefix): + return "gpu" + return "default" + + +def _warmup_buckets(pool: str, default: int, gpu: int, baremetal: int) -> int: + cls = _pool_class(pool) + if cls == "baremetal": + return baremetal + if cls == "gpu": + return gpu + return default + + +def _mark_empty_if_needed(node: Node, bucket_idx: int) -> None: + if node.warming_until_bucket is not None: + return + if node.live_jobs == 0 and not node.placeholders: + node.cpu_used_m = 0 + node.mem_used_mi = 0 + node.gpu_used = 0 + if node.empty_since_bucket is None: + node.empty_since_bucket = bucket_idx + + +def _place_free( + pool_nodes: list[Node], cpu_m: int, mem_mi: int, gpu: int, include_warming: bool = False +) -> Node | None: + best: Node | None = None + best_score = -1.0 + for n in pool_nodes: + if not include_warming and n.warming_until_bucket is not None: + continue + if not fits(n, cpu_m, mem_mi, gpu): + continue + score = most_allocated_score(n) + if score > best_score: + best = n + best_score = score + return best + + +def _preempt_placeholder( + pool_nodes: list[Node], cpu_m: int, mem_mi: int, gpu: int, include_warming: bool = False +) -> tuple[Node, Placeholder] | None: + best: tuple[Node, Placeholder] | None = None + best_score = -1.0 + for n in pool_nodes: + if not include_warming and n.warming_until_bucket is not None: + continue + for p in n.placeholders: + if p.cpu_m == cpu_m and p.mem_mi == mem_mi and p.gpu == gpu: + score = most_allocated_score(n) + if score > best_score: + best = (n, p) + best_score = score + break + return best + + +def _place_phantom(pool_nodes: list[Node], job: Job, cap: float) -> None: + candidates = [n for n in pool_nodes if n.warming_until_bucket is None] + candidates.sort(key=most_allocated_score) + for n in candidates: + if n.cpu_used_m + n.phantom_cpu_m + job.cpu_m > n.cpu_allocatable_m: + continue + if n.mem_used_mi + n.phantom_mem_mi + job.mem_mi > n.mem_allocatable_mi: + continue + if n.gpu_used + n.phantom_gpu + job.gpu > n.gpu_allocatable: + continue + cpu_basis = n.cpu_allocatable_m + n.daemonset_cpu_m + mem_basis = n.mem_allocatable_mi + n.daemonset_mem_mi + gpu_basis = n.gpu_allocatable + n.daemonset_gpu + if cpu_basis > 0 and (n.phantom_cpu_m + job.cpu_m) / cpu_basis > cap: + continue + if mem_basis > 0 and (n.phantom_mem_mi + job.mem_mi) / mem_basis > cap: + continue + if gpu_basis > 0 and (n.phantom_gpu + job.gpu) / gpu_basis > cap: + continue + n.phantom_cpu_m += job.cpu_m + n.phantom_mem_mi += job.mem_mi + n.phantom_gpu += job.gpu + return + + +def simulate( + jobs: list[Job], + model: ClusterModel, + seed: int, + empty_ttl_buckets: int = KARPENTER_EMPTY_TTL_BUCKETS, + placeholder_max_age: int = PLACEHOLDER_MAX_AGE_BUCKETS, + warmup_buckets_default: int = WARMUP_BUCKETS_DEFAULT, + warmup_buckets_gpu: int = WARMUP_BUCKETS_GPU, + warmup_buckets_baremetal: int = WARMUP_BUCKETS_BAREMETAL, + placeholders_enabled: bool = True, + daemonsets_in_metric: bool = False, + phantom_pods_enabled: bool = False, + phantom_lookahead_buckets: int = 1, + phantom_cap: float = 0.30, + progress: bool = True, +) -> dict: + rng = random.Random(seed) + + def warmup_for(pool: str) -> int: + return _warmup_buckets(pool, warmup_buckets_default, warmup_buckets_gpu, warmup_buckets_baremetal) + + arrivals: dict[int, list[Job]] = defaultdict(list) + for j in jobs: + arrivals[j.start_bucket].append(j) + + t_first = min(j.start_bucket for j in jobs) + t_last = max(j.end_bucket for j in jobs) + total_buckets = (t_last - t_first) // BUCKET_SEC + 1 + + pools: dict[str, list[Node]] = defaultdict(list) + finishers: dict[int, list[tuple[Job, Node]]] = defaultdict(list) + per_bucket: list[tuple[int, dict[str, dict]]] = [] + pool_max_nodes: dict[str, int] = defaultdict(int) + pool_max_warming: dict[str, int] = defaultdict(int) + pool_total_created: dict[str, int] = defaultdict(int) + pool_total_placeholders: dict[str, int] = defaultdict(int) + pool_total_preempted: dict[str, int] = defaultdict(int) + pool_total_expired: dict[str, int] = defaultdict(int) + + max_wu = max(warmup_buckets_default, warmup_buckets_gpu, warmup_buckets_baremetal) + + bucket_idx = 0 + for t in range(t_first, t_last + BUCKET_SEC, BUCKET_SEC): + # 0. Promote warming nodes. + for ns in pools.values(): + for n in ns: + if n.warming_until_bucket is not None and bucket_idx >= n.warming_until_bucket: + n.warming_until_bucket = None + _mark_empty_if_needed(n, bucket_idx) + + # 1. Expire placeholders. + for name, ns in pools.items(): + for n in ns: + kept: list[Placeholder] = [] + for p in n.placeholders: + if bucket_idx - p.created_bucket >= placeholder_max_age: + n.cpu_used_m -= p.cpu_m + n.mem_used_mi -= p.mem_mi + n.gpu_used -= p.gpu + pool_total_expired[name] += 1 + else: + kept.append(p) + n.placeholders = kept + _mark_empty_if_needed(n, bucket_idx) + + # 2. Deprovision finishers. + for j, node in finishers.pop(t, ()): + node.cpu_used_m -= j.cpu_m + node.mem_used_mi -= j.mem_mi + node.gpu_used -= j.gpu + node.live_jobs -= 1 + _mark_empty_if_needed(node, bucket_idx) + + # 2b. Look-ahead pre-provisioning per future pod. + if max_wu > 0: + pool_future: dict[str, list[Job]] = defaultdict(list) + for offset in range(1, max_wu + 1): + future_t = t + offset * BUCKET_SEC + for fj in arrivals.get(future_t, ()): + if offset > warmup_for(fj.pool): + continue + pool_future[fj.pool].append(fj) + for pool, fjs in pool_future.items(): + free_cpu = sum(n.cpu_allocatable_m - n.cpu_used_m for n in pools[pool]) + free_mem = sum(n.mem_allocatable_mi - n.mem_used_mi for n in pools[pool]) + free_gpu = sum(n.gpu_allocatable - n.gpu_used for n in pools[pool]) + for fj in fjs: + if free_cpu >= fj.cpu_m and free_mem >= fj.mem_mi and free_gpu >= fj.gpu: + free_cpu -= fj.cpu_m + free_mem -= fj.mem_mi + free_gpu -= fj.gpu + continue + node = model.make_node(pool, fj.cpu_m, fj.mem_mi, fj.gpu) + node.warming_until_bucket = bucket_idx + warmup_for(pool) + pools[pool].append(node) + pool_total_created[pool] += 1 + free_cpu += node.cpu_allocatable_m - fj.cpu_m + free_mem += node.mem_allocatable_mi - fj.mem_mi + free_gpu += node.gpu_allocatable - fj.gpu + + # 3. Top up placeholders per (pool, cpu_m, mem_mi, gpu). + arr = arrivals.get(t, []) + if placeholders_enabled: + needed: dict[tuple[str, int, int, int], int] = defaultdict(int) + for j in arr: + needed[(j.pool, j.cpu_m, j.mem_mi, j.gpu)] += 1 + for (pool, cpu_m, mem_mi, gpu), n_needed in needed.items(): + have = 0 + for node in pools[pool]: + for p in node.placeholders: + if p.cpu_m == cpu_m and p.mem_mi == mem_mi and p.gpu == gpu: + have += 1 + to_create = n_needed - have + if to_create <= 0: + continue + for _ in range(to_create): + node = _place_free(pools[pool], cpu_m, mem_mi, gpu, include_warming=True) + if node is None: + wu = warmup_for(pool) + node = model.make_node(pool, cpu_m, mem_mi, gpu) + node.warming_until_bucket = (bucket_idx + wu) if wu > 0 else None + pools[pool].append(node) + pool_total_created[pool] += 1 + node.cpu_used_m += cpu_m + node.mem_used_mi += mem_mi + node.gpu_used += gpu + node.placeholders.append( + Placeholder(cpu_m=cpu_m, mem_mi=mem_mi, gpu=gpu, created_bucket=bucket_idx) + ) + node.empty_since_bucket = None + pool_total_placeholders[pool] += 1 + + # 4. Place arrivals in shuffled order. + rng.shuffle(arr) + for j in arr: + # (a) Truly-free room on a warm node. + node = _place_free(pools[j.pool], j.cpu_m, j.mem_mi, j.gpu) + if node is not None: + node.cpu_used_m += j.cpu_m + node.mem_used_mi += j.mem_mi + node.gpu_used += j.gpu + node.live_jobs += 1 + node.empty_since_bucket = None + finishers[j.end_bucket].append((j, node)) + continue + # (b) Preempt a matching-shape placeholder on a warm node. + picked = _preempt_placeholder(pools[j.pool], j.cpu_m, j.mem_mi, j.gpu) if placeholders_enabled else None + if picked is not None: + node, ph = picked + node.placeholders.remove(ph) + pool_total_preempted[j.pool] += 1 + node.live_jobs += 1 + node.empty_since_bucket = None + finishers[j.end_bucket].append((j, node)) + continue + # (c) Fresh warm node — unforecasted arrival cannot be deferred. + node = model.make_node(j.pool, j.cpu_m, j.mem_mi, j.gpu) + pools[j.pool].append(node) + pool_total_created[j.pool] += 1 + node.cpu_used_m += j.cpu_m + node.mem_used_mi += j.mem_mi + node.gpu_used += j.gpu + node.live_jobs += 1 + finishers[j.end_bucket].append((j, node)) + + # 5. Deprovision same-bucket start=end finishers. + for j, node in finishers.pop(t, ()): + node.cpu_used_m -= j.cpu_m + node.mem_used_mi -= j.mem_mi + node.gpu_used -= j.gpu + node.live_jobs -= 1 + _mark_empty_if_needed(node, bucket_idx) + + # 5b. Phantom load: pre-count next-bucket arrivals on candidate warm nodes. + if phantom_pods_enabled: + for ns in pools.values(): + for n in ns: + n.phantom_cpu_m = 0 + n.phantom_mem_mi = 0 + n.phantom_gpu = 0 + for offset in range(1, phantom_lookahead_buckets + 1): + for pj in arrivals.get(t + offset * BUCKET_SEC, ()): + _place_phantom(pools[pj.pool], pj, phantom_cap) + + # 6. Measure. + ds_factor = 1 if daemonsets_in_metric else 0 + per_pool: dict[str, dict] = {} + for name, ns in pools.items(): + if not ns: + continue + cpu_used = sum(n.cpu_used_m + ds_factor * n.daemonset_cpu_m + n.phantom_cpu_m for n in ns) + cpu_alloc = sum(n.cpu_allocatable_m + ds_factor * n.daemonset_cpu_m for n in ns) + mem_used = sum(n.mem_used_mi + ds_factor * n.daemonset_mem_mi + n.phantom_mem_mi for n in ns) + mem_alloc = sum(n.mem_allocatable_mi + ds_factor * n.daemonset_mem_mi for n in ns) + gpu_used = sum(n.gpu_used + ds_factor * n.daemonset_gpu + n.phantom_gpu for n in ns) + gpu_alloc = sum(n.gpu_allocatable + ds_factor * n.daemonset_gpu for n in ns) + per_pool[name] = { + "cpu_used_m": cpu_used, + "cpu_alloc_m": cpu_alloc, + "mem_used_mi": mem_used, + "mem_alloc_mi": mem_alloc, + "gpu_used": gpu_used, + "gpu_alloc": gpu_alloc, + } + if len(ns) > pool_max_nodes[name]: + pool_max_nodes[name] = len(ns) + warming_count = sum(1 for n in ns if n.warming_until_bucket is not None) + if warming_count > pool_max_warming[name]: + pool_max_warming[name] = warming_count + + per_bucket.append((t, per_pool)) + + # 7. Consolidate. + for name in list(pools): + kept_nodes = [] + for n in pools[name]: + if n.empty_since_bucket is not None and bucket_idx - n.empty_since_bucket >= empty_ttl_buckets: + continue + kept_nodes.append(n) + if kept_nodes: + pools[name] = kept_nodes + else: + del pools[name] + + if progress and bucket_idx % 1000 == 0: + live = sum(len(ns) for ns in pools.values()) + print( + f" bucket {bucket_idx:>6}/{total_buckets} " + f"{datetime.fromtimestamp(t, UTC):%Y-%m-%d %H:%M} nodes={live:>5}", + file=sys.stderr, + ) + bucket_idx += 1 + + return { + "per_bucket": per_bucket, + "pool_max_nodes": dict(pool_max_nodes), + "pool_max_warming": dict(pool_max_warming), + "pool_total_created": dict(pool_total_created), + "pool_total_placeholders": dict(pool_total_placeholders), + "pool_total_preempted": dict(pool_total_preempted), + "pool_total_expired": dict(pool_total_expired), + "flags": { + "daemonsets_in_metric": daemonsets_in_metric, + "phantom_pods_enabled": phantom_pods_enabled, + "phantom_lookahead_buckets": phantom_lookahead_buckets, + "phantom_cap": phantom_cap, + "placeholders_enabled": placeholders_enabled, + "empty_ttl_buckets": empty_ttl_buckets, + "placeholder_max_age": placeholder_max_age, + "warmup_buckets_default": warmup_buckets_default, + "warmup_buckets_gpu": warmup_buckets_gpu, + "warmup_buckets_baremetal": warmup_buckets_baremetal, + }, + } + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("csv", help="workload CSV produced by build_csv.py extract") + ap.add_argument("--seed", type=int, default=42, help="RNG seed for arrival shuffle") + ap.add_argument( + "--empty-ttl-buckets", + type=int, + default=KARPENTER_EMPTY_TTL_BUCKETS, + help="drop nodes empty for >= this many consecutive 5min buckets (default 2 = 10min)", + ) + ap.add_argument( + "--placeholder-max-age", + type=int, + default=PLACEHOLDER_MAX_AGE_BUCKETS, + help="expire placeholders at start of the bucket that's >= this many buckets after creation", + ) + ap.add_argument( + "--no-runner-pods", + action="store_true", + help=f"disable modeling the ARC runner-pod overhead on the {RUNNER_POD_POOL} fleet", + ) + ap.add_argument("--runner-pod-pool", default=RUNNER_POD_POOL, help="pool name for runner-pod overhead") + ap.add_argument( + "--drop-provider", + action="append", + default=[], + choices=["mt", "lf"], + help="drop all rows with this provider (repeatable)", + ) + ap.add_argument( + "--keep-fraction", + type=float, + default=1.0, + help="deterministically keep this fraction of rows (0 < f <= 1)", + ) + ap.add_argument( + "--no-warmup", + action="store_true", + help="disable the warming-node model: fresh nodes are instantly warm", + ) + ap.add_argument( + "--no-placeholders", + action="store_true", + help="disable ARC placeholder pods entirely (step 3 skipped, step 4b never fires)", + ) + ap.add_argument("--warmup-buckets", type=int, default=WARMUP_BUCKETS_DEFAULT) + ap.add_argument("--warmup-buckets-gpu", type=int, default=WARMUP_BUCKETS_GPU) + ap.add_argument("--warmup-buckets-baremetal", type=int, default=WARMUP_BUCKETS_BAREMETAL) + ap.add_argument( + "--daemonsets-in-metric", + action="store_true", + help="include DaemonSet requests in numerator+denominator (matches prod node_compactor_node_utilization_ratio)", + ) + ap.add_argument( + "--phantom-pods", + action="store_true", + help="pre-count next-bucket arrivals as phantom load on warm nodes (matches prod compactor)", + ) + ap.add_argument( + "--phantom-lookahead-buckets", + type=int, + default=1, + help="how many future buckets of arrivals to phantom-load (default 1 = next bucket only)", + ) + ap.add_argument( + "--phantom-cap", + type=float, + default=0.30, + help="max fraction of node allocatable consumed by phantoms (default 0.30 matches prod PHANTOM_LOAD_CAP)", + ) + ap.add_argument("--no-progress", action="store_true") + args = ap.parse_args() + + print(f"loading {args.csv}...", file=sys.stderr) + jobs = load_jobs( + Path(args.csv), + add_runner_pods=not args.no_runner_pods, + runner_pool=args.runner_pod_pool, + drop_providers=set(args.drop_provider), + keep_fraction=args.keep_fraction, + ) + print(f" {len(jobs):,} pod-lifetimes (including runner pods)", file=sys.stderr) + + if not jobs: + print("no jobs - nothing to simulate", file=sys.stderr) + return 1 + + if args.no_warmup: + wu_default = wu_gpu = wu_baremetal = 0 + else: + wu_default = args.warmup_buckets + wu_gpu = args.warmup_buckets_gpu + wu_baremetal = args.warmup_buckets_baremetal + + placeholders_enabled = not args.no_placeholders + print( + f"simulating (seed={args.seed}, empty_ttl_buckets={args.empty_ttl_buckets}, " + f"placeholder_max_age={args.placeholder_max_age}, " + f"warmup=default:{wu_default}/gpu:{wu_gpu}/bm:{wu_baremetal} buckets, " + f"placeholders={placeholders_enabled}, " + f"daemonsets_in_metric={args.daemonsets_in_metric}, " + f"phantom_pods={args.phantom_pods}" + f"{f' (lookahead={args.phantom_lookahead_buckets}, cap={args.phantom_cap})' if args.phantom_pods else ''}" + f")...", + file=sys.stderr, + ) + model = ClusterModel() + sim = simulate( + jobs, + model=model, + seed=args.seed, + empty_ttl_buckets=args.empty_ttl_buckets, + placeholder_max_age=args.placeholder_max_age, + warmup_buckets_default=wu_default, + warmup_buckets_gpu=wu_gpu, + warmup_buckets_baremetal=wu_baremetal, + placeholders_enabled=placeholders_enabled, + daemonsets_in_metric=args.daemonsets_in_metric, + phantom_pods_enabled=args.phantom_pods, + phantom_lookahead_buckets=args.phantom_lookahead_buckets, + phantom_cap=args.phantom_cap, + progress=not args.no_progress, + ) + + report(sim) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From 0ca133e835c8afb2337ba16cdf9fdeb2b4f4c52b Mon Sep 17 00:00:00 2001 From: Jean Schmidt Date: Thu, 2 Jul 2026 12:53:15 -0700 Subject: [PATCH 02/16] sweep: node/pod sizing optimizer design - Add optimize.md: design for a per-fleet-family node/pod sizing optimizer targeting max(CPU, mem) allocatable-weighted utilization on HUD data - Define objective split: opt metric (workload-only over post-kubelet capacity) for ranking, cal metric matching prod PromQL for calibration - Specify shape catalog + eligibility (data-driven N, not power-of-2) and a checkpointed, resumable multi-restart hill-climb search - Document 8 design decisions (family-locked, virtual sub-fleets, GPU/ baremetal/reserved scope-out) and 11 risks with mitigations - Lay out phased plan (Phase 0 calibration -> catalog -> search -> sensitivity -> git-apply-able patch deliverables) and file layout Notes: design only, no executable code yet. Scope excludes p4d/p5/p6, -large/-metal variants, reserved-capacity fleets, and c7i-runner (fixed ARC pod shape). Coupling across families is treated as zero on the workload side, so each family is searched independently. Absolute util numbers stay ungrounded until the Phase 0 sim-vs-prod delta is measured; if it exceeds 5pp, recs are reframed as deltas vs baseline. Signed-off-by: Jean Schmidt --- osdc/scripts/node-size-sweep/optimize.md | 743 +++++++++++++++++++++++ 1 file changed, 743 insertions(+) create mode 100644 osdc/scripts/node-size-sweep/optimize.md diff --git a/osdc/scripts/node-size-sweep/optimize.md b/osdc/scripts/node-size-sweep/optimize.md new file mode 100644 index 00000000..36c85de0 --- /dev/null +++ b/osdc/scripts/node-size-sweep/optimize.md @@ -0,0 +1,743 @@ +# Node/pod sizing optimizer — design + +## Goal + +For each fleet family (r7a, c7i, m7i, c7a, m8g, m7g, m6i, r7i, g5, g6, g4dn), +find the combination of (1) AWS instance sizes offered by the fleet and (2) +per-def pod shapes that maximizes **allocatable-weighted utilization of the +binding resource** — `max(CPU util, memory util)` — on real HUD workload data. + +Skip fleets that are already right-sized or out of scope: `p4d`, `p5`, +`p5-large`, `p6-b200`, `p6-b200-large`, all `-metal` variants, the +reserved-capacity fleets, `-large` variants (whole-node-per-pod by design), and +`c7i-runner`. + +## Why this is the right optimization target + +Two failure modes we're steering between: + +- **Fragmentation waste** dominates when instance size is too large. A 8c/64Gi + pod on a r7a.48xl (192c/1536Gi) with 1/1 packing wastes ~184c of the node. + Small pods on big nodes underutilize. +- **Overhead waste** dominates when instance size is too small. A 8c/64Gi pod + on a r7a.2xl (8c/64Gi) with 1/1 packing pays ~1c kubelet + ~0.5c DaemonSets + = ~15-20% of the node lost to fixed overhead before any workload. Small + instances have terrible overhead ratios. + +Sweet spot is usually 2-3 instance sizes per fleet with pod shapes tuned to +fit cleanly. + +CPU util alone rewards packing patterns that strand expensive memory on +memory-optimized families (r7a, r7i). Memory util alone rewards patterns that +strand CPU on compute-optimized families (c7a, c7i). Picking `max(cpu, mem)` +optimizes for the resource that will actually run out first — the binding +constraint. The sim already reports both dimensions; the ranking function +takes the max. + +## Logic — the search space + +### Shape catalog + +For each AWS instance size in each family, enumerate every valid split N: + +``` +N_max = min( + alloc_cpu_m // req_cpu_m, + alloc_mem_mi // req_mem_mi, + alloc_gpu // req_gpu if req_gpu > 0 else infinity, +) +enumerate N in 1..N_max +``` + +then compute per-slot allocatable capacity: + +``` +slot_cpu_m = (alloc_cpu_m - N * 0) // N # even split +slot_mem_mi = alloc_mem_mi // N +slot_gpu = alloc_gpu // N # only valid if gpu % N == 0 +``` + +Reject shapes where `slot_gpu == 0` but the def requires a GPU, and shapes +where any slot dimension is <= 0. + +This admits current production shapes (N ∈ {3, 5, 6, 12, 24, 96}) that a +power-of-2 constraint would exclude. Catalog size per family is typically +50-150 shapes, still tractable. + +### Per-def eligibility + +For each runner def, filter the family's catalog to shapes where +`slot >= request` on ALL 3 dimensions. Compute per-eligible-shape waste on +each axis (`slot - request`) so operators can eyeball the tradeoffs before any +sim runs. + +### Objective + +Two distinct metrics with distinct roles: one for optimization ranking, one +for calibration against prod. + +**Ranking metric — optimization objective:** + +``` +opt_cpu = sum(cpu_used_m) / sum(cpu_allocatable_m + ds_cpu_m) +opt_mem = sum(mem_used_mi) / sum(mem_allocatable_mi + ds_mem_mi) + +objective(sim_result) = allocatable_weighted( max(opt_cpu, opt_mem) ) +``` + +Numerator is workload requests only (excludes DaemonSets). Denominator is +capacity net of kubelet reserved (post-kubelet, pre-DS) — the physical space +available for real work if DS overhead were zero. DS appears in the +denominator as a fixed per-node tax, so configs that reduce per-node DS +fraction (bigger instances, fewer nodes) win by shrinking the denominator; +configs that pack workload tighter win by growing the numerator. Both are +real optimization axes. + +**Calibration metric — matches prod dashboard PromQL:** + +``` +cal_cpu = sum(cpu_used_m + ds_cpu_m) / sum(cpu_allocatable_m + ds_cpu_m) +cal_mem = sum(mem_used_mi + ds_mem_mi) / sum(mem_allocatable_mi + ds_mem_mi) +``` + +Matches `node_compactor_node_utilization_ratio` (all pod requests including +DaemonSets over `node.status.allocatable`, i.e. post-kubelet-pre-DS). Sim's +`--daemonsets-in-metric` flag already emits this. Used ONLY for Phase 0 +sim-vs-prod calibration and for final reports so a human can cross-check the +sim number against Grafana. Not used for ranking. + +Tie-breaker on the ranking metric: total node-hours (lower is better). Report +opt AND cal for both CPU and memory in every output; ranking uses +`max(opt_cpu, opt_mem)`. + +### Search structure + +The problem factors along fleet-family boundaries. r7a workloads don't +consume c7i nodes. `c7i-runner` receives a fixed-shape 750m/1Gi pod per +workflow job regardless of the workflow's fleet (see runner.yaml.tpl — +`CAPACITY_AWARE_RUNNER_CPU` / `RUNNER_MEMORY` are hard-coded), so the +workflow-fleet optimization does not perturb c7i-runner load at all. Coupling +between families is zero at this metric. Outer loop: **for each family, run +an independent search**. + +## Decisions made + +### D1: Two metrics — opt for ranking, cal for calibration + +**Ranking (opt_cpu / opt_mem)**: numerator is workload requests only, +denominator is `allocatable + ds` (physical space post-kubelet, pre-DS). +Ranks configs by `max(opt_cpu, opt_mem)`. This is the true optimization +target: it rewards both packing workload tighter (grow numerator) and +reducing per-node DS fraction by using bigger instances (shrink denominator). + +**Calibration (cal_cpu / cal_mem)**: matches prod's +`node_compactor_node_utilization_ratio` — DS included in both num and denom. +Used ONLY for Phase 0 sim-vs-prod delta measurement and for reports so the +number is cross-checkable against Grafana. NOT used for ranking. + +Node-hours is the tie-breaker on the ranking metric. + +### D2: Family-locked mapping + +Every def stays in its current fleet family. This is a scoping decision, not +a technical limit. Reasons: + +- ARC runner labels are addressable identities from GitHub workflows + (`runs-on: linux.arm64.m7g.4xlarge`, etc.). The family name is encoded in + the label; cross-family swaps break label semantics and require coordinated + PRs across `pytorch/pytorch` `.github/workflows/`. +- Family choice encodes NUMA/network topology assumptions and arch (amd64 vs + arm64) that pod `cpu_m` / `mem_mi` requests do not capture. +- Family boundaries match reserved-capacity boundaries in the account. + +Cross-family moves are a separate optimization pass we are not doing here. + +### D3: Fleet independence — zero coupling on the workload side + +Per-fleet search. Other fleets held at current config. Coupling to +`c7i-runner` is zero because the runner-pod shape is fixed by the ARC +template, not derived from workflow fleet choice. No cross-fleet +re-optimization phase is required. + +### D4: N enumeration is data-driven, not power-of-2 + +`N ∈ 1..N_max` where `N_max` is derived per (def, instance) from the +allocatable-vs-request division. This admits the shapes production actually +runs (N=3 on r7a.24xl for a 24c def; N=12 on r7a.48xl for the 8c amx def) +that a `{1,2,4,8,16,32}` constraint would exclude. + +### D5: Shape represents pod REQUESTS, not real usage + +Sim optimizes on requests, same as Karpenter's scheduler. A workload that +requests 46c but uses 15c is packed as 46c. Prod's Karpenter runs with +`WhenEmptyOrUnderutilized` consolidation, which reclaims based on real +utilization — so prod may see additional headroom sim does not model. +Expect a systematic gap between sim util and Grafana util; Phase 0 +calibrates it. + +### D6: Deliverable is git-apply-able patches + +Recommendations are emitted as: + +- Per-def unified diff against `modules/arc-runners/defs/