2058 lines
109 KiB
Python
Executable File
2058 lines
109 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""exam - Kubernetes certification exam simulator (CKA / CKAD / CKS tracks).
|
|
Hands-on tasks graded per-check against the live cluster, exam clock, attempt history,
|
|
weak-spot log, drill mode, and a browser front end modelled on the real exam."""
|
|
import http.server
|
|
import importlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
from collections.abc import Callable
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
VERSION = "2026.10.05.1"
|
|
ROOT = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, ROOT)
|
|
|
|
# ------------------------------------------------------------ track ----
|
|
ARGV = sys.argv[1:]
|
|
TRACKS = ["cka", "ckad", "cks"]
|
|
TRACK_ID = ""
|
|
TRACK = {}
|
|
QUESTIONS = []
|
|
DOMAINS = {}
|
|
QUOTA = {}
|
|
QUOTA_HARD = {}
|
|
QBYID = {}
|
|
SETS = {}
|
|
POOL = []
|
|
EXAM_SECONDS = 0
|
|
PASS_PCT = 0
|
|
STATE_FILE = ""
|
|
HISTORY_FILE = ""
|
|
WEAK_FILE = ""
|
|
STUDY_FILE = "" # tasks solved in study mode (the objective checklist reads it; history never does)
|
|
OBJECTIVES = [] # the track's curriculum, objective by objective (tracks/<id>/curriculum.py), if it has one
|
|
STUDY_ORDER = [] # domains in the order the path walks them, when that differs from the curriculum's own
|
|
K8S = "" # Kubernetes minor the track's real exam runs, e.g. "v1.35"
|
|
CLUSTER = "" # kind cluster for that minor
|
|
CONTEXT = ""
|
|
NODES = {}
|
|
|
|
# One kind cluster per Kubernetes minor an exam runs on, so tracks on the same minor share one
|
|
# (CKA and CKS today) and moving an exam to a new minor is a one-line change to its TRACK["k8s"].
|
|
# The images are the ones the kind v0.33.0 release notes publish; kind only guarantees an image
|
|
# built for its own release when it is pinned by digest.
|
|
KIND_IMAGES = {
|
|
"v1.35": "kindest/node:v1.35.8@sha256:07b2536e30b803ed61d1677a79df6115f798ce64c80f9e22f6ed45afd09323c0",
|
|
"v1.37": "kindest/node:v1.37.0@sha256:a1ed56cfb0e7b93589bdf97c8cd566405a265939e3620fc4f5de89adff580ae5",
|
|
}
|
|
|
|
def kind_cluster(k8s):
|
|
"""Name of the kind cluster for a Kubernetes minor: v1.35 -> cka-sim-135."""
|
|
return "cka-sim-" + k8s.lstrip("v").replace(".", "")
|
|
|
|
|
|
def load_track(track_id):
|
|
"""Bind every track-dependent global. Called at startup and when the UI switches track."""
|
|
try:
|
|
mod = importlib.import_module(f"tracks.{track_id}.questions")
|
|
except ModuleNotFoundError:
|
|
sys.exit(f"unknown track '{track_id}' (no tracks/{track_id}/questions.py)")
|
|
importlib.reload(mod)
|
|
track, questions = mod.TRACK, list(mod.QUESTIONS)
|
|
try:
|
|
cur = importlib.import_module(f"tracks.{track_id}.curriculum")
|
|
cur = importlib.reload(cur)
|
|
objectives, study_order = list(cur.OBJECTIVES), list(getattr(cur, "STUDY_ORDER", []))
|
|
except ModuleNotFoundError as e:
|
|
if e.name != f"tracks.{track_id}.curriculum":
|
|
raise
|
|
objectives, study_order = [], [] # no objective map for this track: study/drill by domain only
|
|
for q in questions:
|
|
q["weight"] = sum(c["pts"] for c in q["checks"])
|
|
q.setdefault("difficulty", "normal")
|
|
q.setdefault("set", None)
|
|
cluster = kind_cluster(track["k8s"])
|
|
g = globals()
|
|
g.update(
|
|
TRACK_ID=track_id, TRACK=track, QUESTIONS=questions, DOMAINS=track["domains"],
|
|
QUOTA=track.get("quota", {}), QUOTA_HARD=track.get("quota_hard", track.get("quota", {})),
|
|
QBYID={q["id"]: q for q in questions},
|
|
SETS=track.get("sets", {}),
|
|
POOL=[q for q in questions if not q.get("set")],
|
|
EXAM_SECONDS=track["minutes"] * 60, PASS_PCT=track["pass_pct"],
|
|
STATE_FILE=os.path.join(ROOT, ".state", f"state-{track_id}.json"),
|
|
HISTORY_FILE=os.path.join(ROOT, f"history-{track_id}.jsonl"),
|
|
WEAK_FILE=os.path.join(ROOT, f"weak-spots-{track_id}.md"),
|
|
STUDY_FILE=os.path.join(ROOT, f"study-{track_id}.json"), OBJECTIVES=objectives, STUDY_ORDER=study_order,
|
|
# The CKA_* variables point every track at your own cluster instead (see the README).
|
|
K8S=track["k8s"], CLUSTER=cluster,
|
|
CONTEXT=os.environ.get("CKA_CONTEXT", f"kind-{cluster}"),
|
|
NODES={"CP": os.environ.get("CKA_CP", f"{cluster}-control-plane"),
|
|
"W1": os.environ.get("CKA_W1", f"{cluster}-worker"),
|
|
"W2": os.environ.get("CKA_W2", f"{cluster}-worker2"),
|
|
"W3": os.environ.get("CKA_W3", f"{cluster}-worker3")},
|
|
)
|
|
|
|
|
|
def set_questions(name):
|
|
return [q for q in QUESTIONS if q.get("set") == name]
|
|
|
|
|
|
STATE_DIR = os.path.join(ROOT, ".state")
|
|
WORKSPACE = os.path.join(ROOT, "workspace")
|
|
# How {WS} is shown in task text and solutions. Relative to the repo root by default, since that
|
|
# is where you are working; graders always use the absolute $WS internally.
|
|
WS_DISPLAY = os.environ.get("CKA_WS_DISPLAY") or os.path.relpath(WORKSPACE, ROOT)
|
|
DRILL_SECONDS_PER_TASK = 6 * 60
|
|
DRILL_SECONDS_PER_TASK_HARD = 5 * 60
|
|
DRILL_SIZE = 5
|
|
import random
|
|
NAMESPACES_TO_SETTLE = ["apps", "netpol", "svc", "ingress-app", "dns-check", "storage"]
|
|
|
|
# --- cluster targeting: CONTEXT and NODES follow the track (load_track); env vars override ---
|
|
def _default_access():
|
|
try:
|
|
return open(os.path.join(STATE_DIR, "runtime")).read().strip() # written by setup.sh
|
|
except OSError:
|
|
return "podman"
|
|
NODE_ACCESS = os.environ.get("CKA_NODE_ACCESS", _default_access()) # podman | docker | ssh | vagrant
|
|
# kubectl matching each cluster's server version, fetched by setup.sh: .state/kubectl/v1.35/kubectl.
|
|
# .state/kubectl/current links to the one for the cluster of the latest session; examrc puts it
|
|
# first on PATH, so your terminals switch kubectl version together with the cluster.
|
|
KUBECTL_DIR = os.path.join(STATE_DIR, "kubectl")
|
|
def shell_hint(node):
|
|
return {"podman": f"podman exec -it {node} bash",
|
|
"docker": f"docker exec -it {node} bash",
|
|
"ssh": f"ssh {node}",
|
|
"vagrant": f"vagrant ssh {node}"}.get(NODE_ACCESS, f"ssh {node}")
|
|
|
|
PRELUDE = r"""
|
|
set -uo pipefail
|
|
K="kubectl --context ${CKA_CONTEXT:?}" # grading never depends on the candidate's current context
|
|
WS="$WORKSPACE"
|
|
# j [kubectl get args...] '{jsonpath}' -> prints the jsonpath result ("" on error)
|
|
j() { local n=$#; local path="${!n}"; $K get "${@:1:n-1}" -o "jsonpath=$path" 2>/dev/null; }
|
|
eq() { [ "$1" = "$2" ]; }
|
|
CP="${CKA_CP:?}"; W1="${CKA_W1:?}"; W2="${CKA_W2:?}"; W3="${CKA_W3:?}" # run() exports the track's nodes
|
|
# nx NODE 'command' -> run a shell command on a node (stdin is forwarded)
|
|
nx() {
|
|
local n="$1"; shift
|
|
case "${CKA_NODE_ACCESS:-podman}" in
|
|
ssh) ssh -o BatchMode=yes -o StrictHostKeyChecking=accept-new "$n" "$@" ;;
|
|
vagrant) vagrant ssh "$n" -c "$*" -- -o LogLevel=QUIET ;;
|
|
docker) docker exec -i "$n" sh -c "$*" ;;
|
|
*) podman exec -i "$n" sh -c "$*" ;;
|
|
esac
|
|
}
|
|
# ensure_etcd_tools -> etcdctl and etcdutl on the control-plane node, as on the real exam.
|
|
# The etcd image is distroless, so they cannot be copied out of the pod; etcd-tools.sh takes
|
|
# them from the image in the node's containerd (and falls back to a download / apt).
|
|
ensure_etcd_tools() {
|
|
nx "$CP" 'command -v etcdctl >/dev/null 2>&1 && command -v etcdutl >/dev/null 2>&1' </dev/null && return 0
|
|
bash "${CKA_ROOT:-.}/etcd-tools.sh" --emit | nx "$CP" 'sh -s'
|
|
}
|
|
# ensure_gateway_crds -> the Gateway API CRDs (standard channel), which kind does not ship. Q56 and
|
|
# Q74 need them at seed time; a cluster rebuilt by ./setup.sh has none until one of them installs it.
|
|
ensure_gateway_crds() {
|
|
$K get crd gateways.gateway.networking.k8s.io >/dev/null 2>&1 && return 0
|
|
curl -sfL --max-time 60 https://github.com/kubernetes-sigs/gateway-api/releases/download/v1.2.1/standard-install.yaml \
|
|
| $K apply -f - >/dev/null 2>&1 \
|
|
&& $K wait --for=condition=Established --timeout=60s crd/gatewayclasses.gateway.networking.k8s.io \
|
|
crd/gateways.gateway.networking.k8s.io crd/httproutes.gateway.networking.k8s.io >/dev/null 2>&1
|
|
}
|
|
# ensure_metrics_server -> a working metrics API for `kubectl top` (CKAD Q23). kind ships none, and
|
|
# kind's kubelet serving certs have no IP SANs, so it only becomes Ready with --kubelet-insecure-tls.
|
|
# CKA Q23 strips that flag on purpose; the baseline puts it back before every session.
|
|
# Pinned: v0.9.x supports Kubernetes 1.34+, so it serves the v1.35 and v1.37 clusters alike, and a
|
|
# future "latest" that drops 1.35 cannot break the CKA track.
|
|
METRICS_SERVER_URL=https://github.com/kubernetes-sigs/metrics-server/releases/download/v0.9.0/components.yaml
|
|
ensure_metrics_server() {
|
|
if ! $K -n kube-system get deploy metrics-server >/dev/null 2>&1; then
|
|
curl -sfL --max-time 60 "$METRICS_SERVER_URL" | $K apply -f - >/dev/null 2>&1 || return 1
|
|
fi
|
|
if ! $K -n kube-system get deploy metrics-server -o jsonpath='{.spec.template.spec.containers[0].args}' | grep -q kubelet-insecure-tls; then
|
|
$K -n kube-system patch deploy metrics-server --type=json \
|
|
-p '[{"op":"add","path":"/spec/template/spec/containers/0/args/-","value":"--kubelet-insecure-tls"}]' >/dev/null 2>&1
|
|
fi
|
|
$K -n kube-system rollout status deploy metrics-server --timeout=150s >/dev/null 2>&1
|
|
}
|
|
"""
|
|
|
|
def _pick_track_arg(argv):
|
|
t = os.environ.get("CKA_TRACK", "cka")
|
|
if "--track" in argv:
|
|
i = argv.index("--track"); t = argv[i + 1]; del argv[i:i + 2]
|
|
return t
|
|
|
|
|
|
load_track(_pick_track_arg(ARGV))
|
|
|
|
TTY = sys.stdout.isatty()
|
|
def c(code, s): return f"\033[{code}m{s}\033[0m" if TTY else s
|
|
def green(s): return c("32", s)
|
|
def red(s): return c("31", s)
|
|
def yellow(s): return c("33", s)
|
|
def bold(s): return c("1", s)
|
|
def dim(s): return c("2", s)
|
|
class Die(SystemExit):
|
|
"""SystemExit that carries its message, so the web UI can show why a start was refused."""
|
|
|
|
def die(msg, code=1):
|
|
print(red(msg))
|
|
e = Die(code); e.msg = msg
|
|
raise e
|
|
|
|
# ------------------------------------------------------------- state ----
|
|
def load_state() -> dict[str, Any] | None:
|
|
if not os.path.exists(STATE_FILE):
|
|
return None
|
|
with open(STATE_FILE) as f:
|
|
return json.load(f)
|
|
|
|
def save_state(st: dict[str, Any]) -> None:
|
|
os.makedirs(STATE_DIR, exist_ok=True)
|
|
with open(STATE_FILE, "w") as f:
|
|
json.dump(st, f, indent=2)
|
|
|
|
def paused_total(st: dict[str, Any] | None) -> float:
|
|
"""Seconds spent paused so far, including the current pause if any."""
|
|
if not st:
|
|
return 0.0
|
|
t = float(st.get("paused_total", 0))
|
|
if st.get("paused_at"):
|
|
t += time.time() - float(st["paused_at"])
|
|
return t
|
|
|
|
def untimed(st: dict[str, Any] | None) -> bool:
|
|
return bool(st and st.get("untimed"))
|
|
|
|
def remaining(st: dict[str, Any] | None) -> float:
|
|
if not st:
|
|
return 0.0
|
|
if untimed(st):
|
|
return float("inf")
|
|
return float(st.get("started_at", 0)) + float(st.get("duration", EXAM_SECONDS)) + paused_total(st) - time.time()
|
|
|
|
def set_paused(st: dict[str, Any], on: bool) -> None:
|
|
if on and not st.get("paused_at"):
|
|
st["paused_at"] = time.time()
|
|
elif not on and st.get("paused_at"):
|
|
st["paused_total"] = float(st.get("paused_total", 0)) + (time.time() - float(st["paused_at"])); st["paused_at"] = None
|
|
save_state(st)
|
|
|
|
def fmt(secs: float | int) -> str:
|
|
secs_int = max(0, int(secs))
|
|
return f"{secs_int // 3600:02d}:{(secs_int % 3600) // 60:02d}:{secs_int % 60:02d}"
|
|
|
|
def require_started() -> dict[str, Any]:
|
|
st = load_state()
|
|
if not st:
|
|
die("No session in progress. Run: ./exam start (or ./exam drill weak)")
|
|
assert st is not None
|
|
return st
|
|
|
|
_TRACK_K8S: dict[str, str] = {}
|
|
def track_k8s():
|
|
"""Each track's Kubernetes minor, for the lobby's track picker (looked up once per process)."""
|
|
if not _TRACK_K8S:
|
|
for t in TRACKS:
|
|
try:
|
|
_TRACK_K8S[t] = importlib.import_module(f"tracks.{t}.questions").TRACK.get("k8s", "?")
|
|
except Exception: # noqa: BLE001 - a bank mid-edit must not take the other tracks' UI down
|
|
_TRACK_K8S[t] = "?"
|
|
return _TRACK_K8S
|
|
|
|
def session_questions(st: dict[str, Any] | None) -> list[dict[str, Any]]:
|
|
if not st:
|
|
return list(QUESTIONS)
|
|
ids = st.get("ids") or [q["id"] for q in QUESTIONS]
|
|
# Preserve the session's own order. Filtering QUESTIONS (which is sorted by id) threw it away,
|
|
# so compose_exam's rng.shuffle never reached the candidate and every exam arrived grouped by
|
|
# domain and difficulty - base tasks first, then hard, then the sim sets. Real exams are not
|
|
# ordered by difficulty, and practising triage on a sorted list teaches the wrong instinct.
|
|
by_id = {q["id"]: q for q in QUESTIONS}
|
|
return [by_id[i] for i in ids if i in by_id]
|
|
|
|
def q_by_id(n: int) -> dict[str, Any]:
|
|
for q in QUESTIONS:
|
|
if q["id"] == n:
|
|
return q
|
|
# ids run 1..N without gaps today, but print the real ranges rather than "1-<count>" so a
|
|
# future gap can never make this message name ids that do not exist.
|
|
ids = sorted(q["id"] for q in QUESTIONS)
|
|
ranges, start, prev = [], ids[0], ids[0]
|
|
for i in ids[1:] + [None]:
|
|
if i != prev + 1:
|
|
ranges.append(str(start) if start == prev else f"{start}-{prev}")
|
|
start = i
|
|
prev = i
|
|
die(f"No question {n}. Valid: {', '.join(ranges)}")
|
|
return {}
|
|
|
|
def render(text, params):
|
|
"""Substitute %{param}, {WS}, {CP}.. and {SHELL_CP}.. placeholders."""
|
|
t = text.replace("{WS}", WS_DISPLAY)
|
|
for k, v in (params or {}).items():
|
|
t = t.replace("%{" + k + "}", str(v))
|
|
for k, v in NODES.items():
|
|
t = t.replace("{" + k + "}", v).replace("{SHELL_" + k + "}", shell_hint(v))
|
|
return t
|
|
|
|
def session_params(st, q):
|
|
"""Params drawn for this session; outside a session, a fixed sample so text still reads naturally."""
|
|
p = (st or {}).get("params", {}).get(str(q["id"]))
|
|
return p if p is not None else make_params(q, random.Random(q["id"]))
|
|
|
|
def task_text(q, params=None):
|
|
return render(q["task"].strip(), params if params is not None else session_params(load_state(), q))
|
|
|
|
def make_params(q, rng):
|
|
fn = q.get("params")
|
|
return fn(rng) if fn else {}
|
|
|
|
# -------------------------------------------------------------- shell ----
|
|
def kubectl_path():
|
|
"""PATH with the kubectl matching this track's Kubernetes minor first, when setup.sh fetched one."""
|
|
d = os.path.join(KUBECTL_DIR, K8S)
|
|
path = os.environ.get("PATH", "")
|
|
return f"{d}{os.pathsep}{path}" if os.path.isfile(os.path.join(d, "kubectl")) else path
|
|
|
|
def run(script, timeout=180, params=None):
|
|
env = dict(os.environ, WORKSPACE=WORKSPACE, CKA_ROOT=ROOT, CKA_NODE_ACCESS=NODE_ACCESS, CKA_CONTEXT=CONTEXT,
|
|
PATH=kubectl_path(), **{f"CKA_{k}": v for k, v in NODES.items()})
|
|
p = subprocess.run(["bash", "-c", PRELUDE + render(script, params)], env=env, capture_output=True, text=True,
|
|
timeout=timeout, stdin=subprocess.DEVNULL)
|
|
return p.returncode, (p.stdout + p.stderr).strip()
|
|
|
|
def server_version():
|
|
"""The cluster's gitVersion (e.g. v1.35.8), or None when it cannot be reached."""
|
|
try:
|
|
p = subprocess.run(["kubectl", "--context", CONTEXT, "get", "--raw=/version"],
|
|
capture_output=True, text=True, timeout=15)
|
|
return json.loads(p.stdout).get("gitVersion") if p.returncode == 0 else None
|
|
except Exception:
|
|
return None
|
|
|
|
def kubectl_ok():
|
|
return server_version() is not None
|
|
|
|
def setup_hint():
|
|
"""How to (re)build this track's cluster."""
|
|
return f"./setup.sh {TRACK_ID}"
|
|
|
|
def grade_one(q, params=None):
|
|
"""Run every check independently. Returns a result dict with partial credit."""
|
|
if params is None:
|
|
params = session_params(load_state(), q)
|
|
checks = []
|
|
for ch in q["checks"]:
|
|
try:
|
|
rc, out = run(ch["script"], timeout=240, params=params)
|
|
except subprocess.TimeoutExpired:
|
|
rc, out = 1, "timed out"
|
|
checks.append({"name": render(ch["name"], params), "pts": ch["pts"], "pass": rc == 0, "out": out[-400:]})
|
|
earned = sum(ch["pts"] for ch in checks if ch["pass"])
|
|
return {"pass": earned == q["weight"], "earned": earned, "total": q["weight"],
|
|
"checks": checks, "at": time.time(), "difficulty": q["difficulty"]}
|
|
|
|
def result_detail(r):
|
|
return "\n".join((" ✓ " if ch["pass"] else " ✗ ") + ch["name"] + f" ({ch['pts']} pt)" for ch in r["checks"])
|
|
|
|
def score(st, results=None):
|
|
results = results if results is not None else st["results"]
|
|
qs = session_questions(st)
|
|
total = sum(q["weight"] for q in qs)
|
|
earned = sum(results.get(str(q["id"]), {}).get("earned", 0) for q in qs)
|
|
return earned, total, (round(100 * earned / total, 1) if total else 0)
|
|
|
|
# ----------------------------------------------------------- history ----
|
|
def load_history(include_synthetic=False):
|
|
"""Attempts from history-<track>.jsonl. Entries marked "synthetic" (a scripted verification
|
|
run, not a person sitting the tasks) stay in the file for the record but are left out of every
|
|
statistic, weak list and "latest exam" unless explicitly asked for."""
|
|
if not os.path.exists(HISTORY_FILE):
|
|
return []
|
|
with open(HISTORY_FILE) as f:
|
|
entries = [json.loads(l) for l in f if l.strip()]
|
|
return entries if include_synthetic else [e for e in entries if not e.get("synthetic")]
|
|
|
|
def record_attempt(st):
|
|
earned, total, pct = score(st)
|
|
entry = {"at": datetime.now().isoformat(timespec="minutes"), "mode": st.get("mode", "exam"), "hard": st.get("hard", False),
|
|
"k8s": st.get("k8s"), # server version it ran on; older entries predate per-track clusters
|
|
"ids": [q["id"] for q in session_questions(st)], "score_pct": pct, "earned": earned, "total": total,
|
|
"time_used": int(st["ended_at"] - st["started_at"] - paused_total(st)),
|
|
"results": {k: {"pass": v["pass"], "earned": v["earned"], "total": v["total"],
|
|
"failed": [ch["name"] for ch in v["checks"] if not ch["pass"]]}
|
|
for k, v in st["results"].items()}}
|
|
with open(HISTORY_FILE, "a") as f:
|
|
f.write(json.dumps(entry) + "\n")
|
|
write_weak_spots()
|
|
return entry
|
|
|
|
class AggEntry:
|
|
def __init__(self) -> None:
|
|
self.attempts: int = 0
|
|
self.fails: int = 0
|
|
self.last: Any = None
|
|
self.last_fail_at: str = ""
|
|
self.failed: list[str] = []
|
|
self.last_earned: str = ""
|
|
self.streak: int = 0 # consecutive full passes at the end of the history
|
|
|
|
def __getitem__(self, item: str) -> Any:
|
|
return getattr(self, item)
|
|
|
|
def __setitem__(self, item: str, value: Any) -> None:
|
|
setattr(self, item, value)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"attempts": self.attempts,
|
|
"fails": self.fails,
|
|
"last": self.last,
|
|
"last_fail_at": self.last_fail_at,
|
|
"failed": self.failed,
|
|
"last_earned": self.last_earned,
|
|
"streak": self.streak,
|
|
}
|
|
|
|
|
|
def stats() -> dict[int, AggEntry]:
|
|
agg: dict[int, AggEntry] = {int(q["id"]): AggEntry() for q in QUESTIONS}
|
|
for e in load_history():
|
|
for k, r in e.get("results", {}).items():
|
|
a = agg.get(int(k))
|
|
if a is None:
|
|
continue
|
|
a.attempts += 1
|
|
a.last = r.get("pass")
|
|
a.last_earned = f"{r.get('earned', 0)}/{r.get('total', 0)}"
|
|
a.streak = a.streak + 1 if r.get("pass") else 0
|
|
if not r.get("pass"):
|
|
a.fails += 1
|
|
a.last_fail_at = str(e.get("at", ""))
|
|
a.failed = list(r.get("failed", []))
|
|
return agg
|
|
|
|
def last_seen():
|
|
"""question id -> index of the most recent session it appeared in (higher = more recent)."""
|
|
seen = {}
|
|
for i, e in enumerate(load_history()):
|
|
for k in e["results"]:
|
|
seen[int(k)] = i
|
|
return seen
|
|
|
|
def drilled_domains():
|
|
out = {d: 0 for d in DOMAINS}
|
|
for e in load_history():
|
|
if e["mode"] != "drill":
|
|
continue
|
|
for k in e["results"]:
|
|
q = QBYID.get(int(k))
|
|
if q:
|
|
out[q["domain"]] += 1
|
|
return out
|
|
|
|
def breaks_of(q):
|
|
"""A task's "breaks" as a set. It is a string or a list: Q61 damages the scheduler AND the
|
|
controller-manager, and declaring only one let it share a session with Q38."""
|
|
b = q.get("breaks")
|
|
return set() if not b else ({b} if isinstance(b, str) else set(b))
|
|
|
|
def no_conflicts(cands, n, taken=()):
|
|
"""Take up to n tasks, never two that break the same thing (or clash with tasks already taken).
|
|
A "solo" task (Q45's etcd restore rolls the WHOLE cluster back to its snapshot, wiping every
|
|
other task's seed and answers) is only ever taken on its own."""
|
|
used = set().union(*(breaks_of(q) for q in taken)) if taken else set()
|
|
out = []
|
|
for q in cands:
|
|
if q.get("solo") and (taken or out):
|
|
continue
|
|
if any(x.get("solo") for x in list(taken) + out):
|
|
break
|
|
b = breaks_of(q)
|
|
if b & used:
|
|
continue
|
|
out.append(q)
|
|
used |= b
|
|
if len(out) == n:
|
|
break
|
|
return out
|
|
|
|
def split_solo(qs):
|
|
"""Pull solo tasks out of a multi-task selection; they have to be drilled alone."""
|
|
if len(qs) <= 1:
|
|
return list(qs), []
|
|
return [q for q in qs if not q.get("solo")], [q for q in qs if q.get("solo")]
|
|
|
|
def pick(domain, n, hard, rng, exclude=(), taken=(), unseen_only=False):
|
|
"""n tasks from a domain: unseen first, then least recently seen; hard sessions prefer hard tasks."""
|
|
seen = last_seen()
|
|
|
|
def build(allow_hard):
|
|
pool = [q for q in POOL if q["domain"] == domain and q["id"] not in exclude and not q.get("solo")
|
|
and (allow_hard or q["difficulty"] == "normal") and not (unseen_only and q["id"] in seen)]
|
|
rng.shuffle(pool)
|
|
pool.sort(key=lambda q: (0 if (hard and q["difficulty"] == "hard") else 1, seen.get(q["id"], -1)))
|
|
return pool
|
|
|
|
got = no_conflicts(build(hard), n, taken)
|
|
if len(got) < n and not hard:
|
|
# A domain whose only tasks are "hard" contributes NOTHING to a normal session and drops
|
|
# out of the blueprint entirely - CKS SUPPLY and RUNTIME (20% each) were both unreachable
|
|
# via ./exam start. Better to hand over a hard task than to silently skip the domain.
|
|
got = no_conflicts(build(True), n, taken)
|
|
return got
|
|
|
|
def compose_exam(hard, rng):
|
|
quota = QUOTA_HARD if hard else QUOTA
|
|
# Scarcest domain first. Two tasks that "breaks" the same thing can't share a session, so a
|
|
# one-task domain loses every such tie to a bigger domain that claimed the shared slot earlier
|
|
# and then contributes nothing: CKS RUNTIME (only Q9, apiserver-flag) was never drawn in ANY
|
|
# mode because SETUP was iterated first and took that slot with Q2.
|
|
order = sorted(quota, key=lambda d: len([q for q in POOL if q["domain"] == d]))
|
|
qs, short = [], {}
|
|
for d in order:
|
|
got = pick(d, quota[d], hard, rng, taken=qs)
|
|
if len(got) < quota[d]:
|
|
short[d] = (len(got), quota[d])
|
|
qs += got
|
|
if short:
|
|
# Never truncate in silence - a short session still gets scored out of its own total, so
|
|
# the missing domains look like they were simply not on the blueprint.
|
|
print(yellow(" warning: bank too small to fill the blueprint - "
|
|
+ ", ".join(f"{d} {g}/{w}" for d, (g, w) in short.items())))
|
|
rng.shuffle(qs)
|
|
return qs
|
|
|
|
def campaign_batch(hard, rng):
|
|
"""'drill all': DRILL_SIZE tasks you have never attempted, round-robin across domains,
|
|
hard tasks included when --hard. When everything has been seen once, falls back to weak tasks."""
|
|
seen = last_seen()
|
|
pool = [q for q in POOL if q["id"] not in seen and not q.get("solo") and (hard or q["difficulty"] == "normal")]
|
|
if not pool:
|
|
return None, []
|
|
qs, domains = [], list(DOMAINS)
|
|
rng.shuffle(pool)
|
|
pool.sort(key=lambda q: 0 if (hard and q["difficulty"] == "hard") else 1)
|
|
while len(qs) < DRILL_SIZE and pool:
|
|
progressed = False
|
|
for d in domains:
|
|
cand = [q for q in pool if q["domain"] == d and q not in qs]
|
|
got = no_conflicts(cand, 1, qs)
|
|
if got:
|
|
qs += got; progressed = True
|
|
if len(qs) == DRILL_SIZE:
|
|
break
|
|
if not progressed:
|
|
break
|
|
total = len([q for q in POOL if hard or q["difficulty"] == "normal"])
|
|
return f"{total - len(pool)}/{total} attempted", qs
|
|
|
|
def solution_allowed(st: dict[str, Any] | None, qid: int | str) -> bool:
|
|
"""Study: always. Drill: once the task has been graded. Exam: only after submit."""
|
|
if not st or st.get("ended_at") or untimed(st):
|
|
return True
|
|
return bool(st.get("mode") == "drill" and str(qid) in st.get("results", {}))
|
|
|
|
def cmd_solution(args):
|
|
if not args:
|
|
die("usage: ./exam solution N (drills: after grading the task; exams: after submit; --force to peek anyway)")
|
|
force = "--force" in args; args = [a for a in args if a != "--force"]
|
|
q = q_by_id(int(args[0])); st = load_state()
|
|
if not solution_allowed(st, int(q["id"])) and not force:
|
|
die("Not yet: in a drill, grade the task first (./exam grade N); in an exam, submit first. --force overrides.")
|
|
print(bold(f"Solution - Q{q['id']} {q['title']}")); print("-" * 64)
|
|
sol = str(q.get("solution", ""))
|
|
print(render(sol, session_params(st, q))); print("-" * 64)
|
|
print(dim(f"One correct way, not the only one. $do = --dry-run=client -o yaml, k = kubectl. Paths are relative to {ROOT}"))
|
|
|
|
def cmd_progress(args):
|
|
seen = last_seen(); agg = stats()
|
|
print(bold(f"{'Domain':<8}{'attempted':>11}{'passed':>9} {'hard attempted':>15}{'hard passed':>13}"))
|
|
for d in DOMAINS:
|
|
dq = [q for q in QUESTIONS if q["domain"] == d]
|
|
n = [q for q in dq if q["difficulty"] == "normal"]; h = [q for q in dq if q["difficulty"] == "hard"]
|
|
att = lambda L: sum(1 for q in L if q["id"] in seen); ok = lambda L: sum(1 for q in L if agg.get(int(q["id"])) and agg[int(q["id"])].last is True)
|
|
print(f"{d:<8}{att(n):>6}/{len(n):<4}{ok(n):>6}/{len(n):<4} {att(h):>9}/{len(h):<4}{ok(h):>8}/{len(h):<4}")
|
|
left = [q["id"] for q in QUESTIONS if q["id"] not in seen]
|
|
print(dim(f"\n{len(QUESTIONS) - len(left)}/{len(QUESTIONS)} tasks attempted at least once." + (f" Never seen: Q{', Q'.join(map(str, left))}" if left and len(left) <= 25 else "")))
|
|
print(dim("passed = latest attempt was a full pass. ./exam drill all [--hard] works through what's left."))
|
|
|
|
def is_weak(q, a):
|
|
"""Weak = failed on the latest attempt, or failed at least twice and not yet passed twice in a
|
|
row since - a flaky task has to prove itself - or never attempted (outside the fixed sets).
|
|
The old rule ("failed twice, ever") kept a task weak forever: 13 of 29 listed were passing."""
|
|
if a.attempts == 0:
|
|
return not q.get("set")
|
|
return a.last is False or (a.fails >= 2 and a.streak < 2)
|
|
|
|
def weak_ids(agg=None):
|
|
agg = agg or stats()
|
|
return [q["id"] for q in QUESTIONS if is_weak(q, agg[q["id"]])]
|
|
|
|
def weak_lines(a):
|
|
"""What to show under a weak task: the failing checks if it failed last time, otherwise how
|
|
close it is to dropping off the list - not the ✗ list of some older, since-fixed attempt."""
|
|
if a.last is False:
|
|
return [f"✗ {n}" for n in (a.failed or [])]
|
|
return [f"passed last time ({a.last_earned}) - one more clean pass takes it off this list"]
|
|
|
|
def exam_domain_scores(e):
|
|
"""domain -> (earned, total) over the tasks that were actually IN that attempt. Summing the
|
|
weight of every task in the bank instead made a 100% run read "ARCH 38/266"."""
|
|
out = {}
|
|
for k, r in e.get("results", {}).items():
|
|
q = QBYID.get(int(k))
|
|
if q:
|
|
de, dt = out.get(q["domain"], (0, 0))
|
|
out[q["domain"]] = (de + r.get("earned", 0), dt + r.get("total", 0))
|
|
return out
|
|
|
|
def write_weak_spots():
|
|
agg = stats(); hist = load_history()
|
|
L = [f"# {TRACK_ID.upper()} weak spots", "", f"_Generated by exam from {len(hist)} attempt(s); regenerated on every `end`._", ""]
|
|
exams = [e for e in hist if e["mode"] == "exam"]
|
|
if exams:
|
|
e = exams[-1]; ds = exam_domain_scores(e)
|
|
L += [f"## Domain scores (latest full exam, {e['at']}, {e['score_pct']}%)", ""]
|
|
for d, name in DOMAINS.items():
|
|
if d in ds:
|
|
L.append(f"- **{d}** {ds[d][0]}/{ds[d][1]} — {name}")
|
|
L.append("")
|
|
L += ["## Drill next (`./exam drill weak`)", ""]
|
|
weak = [q for q in QUESTIONS if q["id"] in weak_ids(agg) and agg[q["id"]]["attempts"] > 0]
|
|
if not weak:
|
|
L.append("Nothing failing right now.")
|
|
for q in weak:
|
|
a = agg[q["id"]]
|
|
L.append(f"### Q{q['id']:02d} · {q['domain']} · {q['title']} ({a['fails']}/{a['attempts']} failed, last {a['last_earned']} pts, last failed {a['last_fail_at']})")
|
|
L += [f"- {n}" for n in weak_lines(a)]
|
|
L.append("")
|
|
L += ["## Attempt matrix", "", "| Q | Domain | Title | " + " | ".join(f"{e['mode'][0].upper()}{i + 1}" for i, e in enumerate(hist)) + " |",
|
|
"|---|---|---|" + "---|" * len(hist)]
|
|
for q in QUESTIONS:
|
|
cells = []
|
|
for e in hist:
|
|
r = e["results"].get(str(q["id"]))
|
|
cells.append("·" if r is None else ("✓" if r["pass"] else f"{r['earned']}/{r['total']}"))
|
|
L.append(f"| {q['id']} | {q['domain']} | {q['title']} | " + " | ".join(cells) + " |")
|
|
L += ["", "_E = full exam, D = drill, · = not in that session, n/m = partial credit._", ""]
|
|
with open(WEAK_FILE, "w") as f:
|
|
f.write("\n".join(L))
|
|
|
|
# --------------------------------------------------------- curriculum ----
|
|
# A track may ship tracks/<id>/curriculum.py: the exam's objectives in order, each with study
|
|
# notes and the tasks that practise it. It turns the bank into a path - study an objective with
|
|
# the solutions open, then prove it against the clock - and `./exam objectives` shows where you are.
|
|
def find_objective(arg):
|
|
"""An objective by id (3.2) or key (probes, or any unambiguous start of it); None if none fits."""
|
|
a = str(arg).lower()
|
|
for o in OBJECTIVES:
|
|
if a in (o["id"], o["key"]):
|
|
return o
|
|
hits = [o for o in OBJECTIVES if len(a) >= 3 and o["key"].startswith(a)]
|
|
return hits[0] if len(hits) == 1 else None
|
|
|
|
def objective_tasks(o, with_sets=False):
|
|
"""An objective's tasks in its own order. Tasks of the fixed simulator sets stay out unless
|
|
asked for: meeting them one objective at a time would spoil the mock they are kept for."""
|
|
return [QBYID[i] for i in o["tasks"] if i in QBYID and (with_sets or not QBYID[i].get("set"))]
|
|
|
|
def objectives_of(qid):
|
|
return [o for o in OBJECTIVES if qid in o["tasks"]]
|
|
|
|
def objective_domains():
|
|
"""Domain codes in the order the curriculum lists them (TRACK["domains"] may order them otherwise)."""
|
|
return list(dict.fromkeys(o["domain"] for o in OBJECTIVES))
|
|
|
|
def path_objectives():
|
|
"""Objectives in the order the path walks them: the curriculum's order, unless the track names
|
|
a STUDY_ORDER of domains (CKA leaves Troubleshooting for last - it builds on all the rest)."""
|
|
order = [d for d in STUDY_ORDER if d in DOMAINS] + [d for d in objective_domains() if d not in STUDY_ORDER]
|
|
return sorted(OBJECTIVES, key=lambda o: order.index(o["domain"])) # stable: curriculum order within a domain
|
|
|
|
def objective_session(o, how, with_sets=False, n=None):
|
|
"""The tasks of one sitting on an objective: what is still open first ("study": never solved;
|
|
"drill": not yet passed against the clock), at most n, never two that break the same thing.
|
|
A big objective therefore takes several sittings, each on new ground."""
|
|
agg, studied = stats(), load_study()
|
|
closed = ((lambda q: task_solved(q["id"], agg, studied)) if how == "study"
|
|
else (lambda q: task_state(q["id"], agg, studied) == "passed"))
|
|
qs = objective_tasks(o, with_sets)
|
|
todo = [q for q in qs if not closed(q)] or qs # nothing open: go over all of it again
|
|
return no_conflicts(todo, n or DRILL_SIZE), len(todo)
|
|
|
|
def load_progress():
|
|
"""study-<track>.json, next to the history so it survives ./teardown.sh as the history does:
|
|
tasks task id -> its latest result in a study session ("passed" sticks once earned)
|
|
objectives objective id -> when it was first studied and first completed
|
|
Study results are kept apart from the history on purpose: a pass with the solution open is
|
|
practice, not proof, so it never reaches the statistics, the weak list or the exam gate."""
|
|
try:
|
|
with open(STUDY_FILE) as f:
|
|
d = json.load(f)
|
|
except (OSError, ValueError):
|
|
d = {}
|
|
return {"tasks": {int(k): v for k, v in d.get("tasks", {}).items()}, "objectives": dict(d.get("objectives", {}))}
|
|
|
|
def save_progress(prog):
|
|
with open(STUDY_FILE, "w") as f:
|
|
json.dump({"tasks": {str(k): prog["tasks"][k] for k in sorted(prog["tasks"])},
|
|
"objectives": {k: prog["objectives"][k] for k in sorted(prog["objectives"])}}, f, indent=1)
|
|
f.write("\n")
|
|
|
|
def load_study():
|
|
return load_progress()["tasks"]
|
|
|
|
def record_study(st, ids=None):
|
|
"""Note what a study session graded (all of it, or just `ids`)."""
|
|
if not untimed(st):
|
|
return
|
|
prog = load_progress(); log = prog["tasks"]
|
|
for k, r in st.get("results", {}).items():
|
|
if ids is not None and int(k) not in ids:
|
|
continue
|
|
log[int(k)] = {"at": datetime.now().isoformat(timespec="minutes"), "pass": r["pass"], "earned": r["earned"],
|
|
"total": r["total"], "passed": bool(r["pass"] or log.get(int(k), {}).get("passed"))}
|
|
save_progress(prog)
|
|
|
|
def task_state(qid, agg, studied):
|
|
"""passed = the latest attempt against the clock was a full pass; failed = it was not;
|
|
studied = solved in study mode but never attempted against the clock; new = neither."""
|
|
a = agg.get(int(qid))
|
|
if a is not None and a.attempts:
|
|
return "passed" if a.last else "failed"
|
|
return "studied" if studied.get(int(qid), {}).get("passed") else "new"
|
|
|
|
def task_solved(qid, agg, studied):
|
|
"""Solved at least once, with the solution open or against the clock."""
|
|
a = agg.get(int(qid))
|
|
return bool(studied.get(int(qid), {}).get("passed") or (a is not None and a.attempts > a.fails))
|
|
|
|
MARKS = {"passed": ("✓", green), "studied": ("~", yellow), "failed": ("✗", red), "new": ("·", dim), "started": ("-", yellow)}
|
|
|
|
def mark(state):
|
|
ch, col = MARKS[state]
|
|
return col(ch)
|
|
|
|
def objective_state(o, agg, studied):
|
|
"""The two stages every objective goes through, and where it is:
|
|
studied every pool task solved at least once (study mode counts)
|
|
done every pool task passed against the clock on its latest attempt - exam conditions
|
|
before that: started (something attempted) or new."""
|
|
qs = objective_tasks(o)
|
|
states = [task_state(q["id"], agg, studied) for q in qs]
|
|
if qs and all(x == "passed" for x in states):
|
|
return "done"
|
|
if qs and all(task_solved(q["id"], agg, studied) for q in qs):
|
|
return "studied"
|
|
return "started" if any(x != "new" for x in states) else "new"
|
|
|
|
OBJECTIVE_MARK = {"done": "passed", "studied": "studied", "started": "started", "new": "new"}
|
|
|
|
def next_objective(agg=None, studied=None):
|
|
"""(objective, how) for the first objective in curriculum order that is not done: "study" it
|
|
until every task has been solved once, then "drill" it against the clock. (None, None) when
|
|
every objective is done."""
|
|
agg = stats() if agg is None else agg
|
|
studied = load_study() if studied is None else studied
|
|
for o in path_objectives():
|
|
state = objective_state(o, agg, studied)
|
|
if state != "done":
|
|
return o, ("drill" if state == "studied" else "study")
|
|
return None, None
|
|
|
|
def mock_scores():
|
|
"""Scores of the full mocks and of each fixed simulator set, from the history."""
|
|
full, fixed = [], {k: [] for k in SETS}
|
|
members = {k: {q["id"] for q in set_questions(k)} for k in SETS}
|
|
for e in load_history():
|
|
if e["mode"] != "exam":
|
|
continue
|
|
hit = next((k for k in SETS if set(e["ids"]) == members[k]), None)
|
|
(fixed[hit] if hit else full).append(e["score_pct"])
|
|
return full, fixed
|
|
|
|
def next_step(agg=None, studied=None):
|
|
"""The next move on the path to the exam, as {kind, cmd, why[, objective | set]}:
|
|
study an objective -> drill it against the clock -> ... every objective -> a full mock ->
|
|
the fixed simulator sets. Each stage is closer to the real exam than the one before."""
|
|
agg = stats() if agg is None else agg
|
|
studied = load_study() if studied is None else studied
|
|
o, how = next_objective(agg, studied)
|
|
if o and how == "study":
|
|
return {"kind": "study", "objective": o, "cmd": f"./exam study {o['id']}",
|
|
"why": f"{o['id']} {o['text']} - notes, then its tasks with no clock and the solutions open"}
|
|
if o:
|
|
return {"kind": "drill", "objective": o, "cmd": f"./exam drill {o['id']}",
|
|
"why": f"{o['id']} {o['text']} - studied; now the same tasks against the clock, no solution until graded"}
|
|
full, fixed = mock_scores()
|
|
if not any(x >= PASS_PCT for x in full):
|
|
return {"kind": "mock", "cmd": "./exam start",
|
|
"why": f"every objective is done - a full mock: {sum(QUOTA.values())} tasks, {TRACK['minutes']} minutes, no feedback until you submit"}
|
|
for k in SETS:
|
|
if not any(x >= PASS_PCT for x in fixed[k]):
|
|
return {"kind": "sim", "set": k, "cmd": f"./exam sim {k}",
|
|
"why": f"{SETS[k]}: {len(set_questions(k))} fixed tasks, {TRACK['minutes']} minutes, exam rules"}
|
|
return {"kind": "done", "cmd": "./exam start --hard",
|
|
"why": f"every objective done and every mock passed at {PASS_PCT}% or better - keep sharp with a fresh mock"}
|
|
|
|
def update_milestones():
|
|
"""Stamp each objective the first time it is studied and the first time it is done, and
|
|
return what was newly reached as [(objective, "studied" | "done")]. Called whenever results
|
|
are recorded, so the checklist keeps the date you got there."""
|
|
if not OBJECTIVES:
|
|
return []
|
|
prog = load_progress(); agg = stats(); new = []
|
|
now = datetime.now().isoformat(timespec="minutes")
|
|
for o in OBJECTIVES:
|
|
state = objective_state(o, agg, prog["tasks"])
|
|
seen = prog["objectives"].setdefault(o["id"], {})
|
|
for level in ("studied", "done"):
|
|
if state in (level, "done") and level not in seen:
|
|
seen[level] = now
|
|
new.append((o, level))
|
|
if not seen:
|
|
del prog["objectives"][o["id"]]
|
|
if new:
|
|
save_progress(prog)
|
|
# reaching "done" in one go also stamps "studied"; only the higher one is news
|
|
return [(o, lv) for o, lv in new if not (lv == "studied" and (o, "done") in new)]
|
|
|
|
def milestone_lines(new):
|
|
out = []
|
|
for o, level in new:
|
|
n = len(objective_tasks(o))
|
|
if level == "done":
|
|
out.append(f"✓ Objective {o['id']} {o['key']} complete - {n}/{n} tasks passed against the clock.")
|
|
else:
|
|
out.append(f"~ Objective {o['id']} {o['key']} studied - every task solved once. Prove it against the clock: ./exam drill {o['id']}")
|
|
return out
|
|
|
|
def curriculum_gaps():
|
|
"""What would make the path lie: a task no objective lists, a listed id that does not exist,
|
|
an objective too thin to study."""
|
|
listed = {i for o in OBJECTIVES for i in o["tasks"]}
|
|
out = []
|
|
orphans = [q["id"] for q in QUESTIONS if q["id"] not in listed]
|
|
if orphans:
|
|
out.append(f"not listed under any objective: Q{', Q'.join(map(str, orphans))}")
|
|
ghosts = sorted(listed - set(QBYID))
|
|
if ghosts:
|
|
out.append(f"listed but not in the bank: Q{', Q'.join(map(str, ghosts))}")
|
|
thin = [o["id"] for o in OBJECTIVES if len(objective_tasks(o)) < 2]
|
|
if thin:
|
|
out.append(f"fewer than two pool tasks: {', '.join(thin)}")
|
|
wrong = [f"{o['id']} ({o['domain']})" for o in OBJECTIVES if o["domain"] not in DOMAINS]
|
|
if wrong:
|
|
out.append(f"unknown domain: {', '.join(wrong)}")
|
|
return out
|
|
|
|
def require_objectives():
|
|
if not OBJECTIVES:
|
|
die(f"Track '{TRACK_ID}' has no objective map (tracks/{TRACK_ID}/curriculum.py).\n"
|
|
f" Study it by domain instead: ./exam study {next(iter(DOMAINS))}")
|
|
|
|
def cmd_objectives(args):
|
|
"""The curriculum as a checklist: every objective, its tasks, and how far you are."""
|
|
require_objectives()
|
|
agg, studied = stats(), load_study()
|
|
cols = shutil.get_terminal_size((120, 24)).columns
|
|
sets = len(QUESTIONS) - len(POOL)
|
|
print(bold(f"{TRACK['name']} - study path") + dim(f" {len(OBJECTIVES)} objectives · {len(POOL)} tasks"
|
|
+ (f" · {sets} more in the fixed sets [in brackets]" if sets else "")))
|
|
print(dim("tasks: ✓ passed against the clock ~ solved in study mode ✗ failed last time · not yet"
|
|
" objectives: ✓ complete ~ studied - started"))
|
|
def cell(q):
|
|
return f"{MARKS[task_state(q['id'], agg, studied)][0]}{q['id']}", f"{mark(task_state(q['id'], agg, studied))}{q['id']}"
|
|
rows = []
|
|
for o in OBJECTIVES:
|
|
pool = [cell(q) for q in objective_tasks(o)]
|
|
fixed = [cell(q) for q in objective_tasks(o, True) if q.get("set")]
|
|
plain = " ".join(c[0] for c in pool) + (" [" + " ".join(c[0] for c in fixed) + "]" if fixed else "")
|
|
shown = " ".join(c[1] for c in pool) + (dim(" [") + " ".join(c[1] for c in fixed) + dim("]") if fixed else "")
|
|
rows.append((o, plain, shown))
|
|
tw = max(28, min(max(len(o["text"]) for o in OBJECTIVES), cols - 26 - max(len(r[1]) for r in rows)))
|
|
for d in objective_domains():
|
|
print(bold(f"\n{DOMAINS[d]}"))
|
|
for o, plain, shown in rows:
|
|
if o["domain"] != d:
|
|
continue
|
|
state = objective_state(o, agg, studied)
|
|
text = o["text"] if len(o["text"]) <= tw else o["text"][:tw - 1] + "…"
|
|
print(f" {mark(OBJECTIVE_MARK[state])} {o['id']:<4}{o['key']:<16}{text:<{tw}} {shown}")
|
|
print_path(agg, studied)
|
|
for g in curriculum_gaps():
|
|
print(yellow(f"curriculum.py: {g}"))
|
|
|
|
def print_path(agg=None, studied=None):
|
|
"""The ladder from first reading to exam conditions, how far up you are, and the next rung."""
|
|
agg = stats() if agg is None else agg
|
|
studied = load_study() if studied is None else studied
|
|
states = [objective_state(o, agg, studied) for o in OBJECTIVES]
|
|
n, done = len(OBJECTIVES), states.count("done")
|
|
full, fixed = mock_scores()
|
|
step = next_step(agg, studied)
|
|
def best(scores):
|
|
if not scores:
|
|
return dim("not sat yet")
|
|
b = max(scores)
|
|
return (green if b >= PASS_PCT else red)(f"best {b}%") + dim(f" of {len(scores)}, pass mark {PASS_PCT}%")
|
|
rows = [("study", "Study", "each objective's tasks, notes and solutions open", f"{done + states.count('studied')}/{n} objectives"),
|
|
("drill", "Clock", "the same tasks as a timed drill, graded per check", f"{done}/{n} objectives")]
|
|
rows.append(("mock", "Mock exam", f"{sum(QUOTA.values())} random tasks, {TRACK['minutes']} min, no feedback until submit", best(full)))
|
|
rows += [("sim", f"Set {k.upper()}", f"{len(set_questions(k))} fixed tasks, {TRACK['minutes']} min, exam rules", best(fixed[k])) for k in SETS]
|
|
print(bold("\nPath to the exam"))
|
|
for i, (kind, name, what, where) in enumerate(rows, 1):
|
|
here = step["kind"] == kind and (kind != "sim" or name == f"Set {step['set'].upper()}")
|
|
print(f" {green('→') if here else ' '} {i} {name:<11}{dim(f'{what:<54}')}{where}")
|
|
print(bold(f"\nNext: {step['cmd']}") + dim(" (or just ./exam next)"))
|
|
print(dim(f" {step['why']}"))
|
|
if step.get("objective"):
|
|
print(dim(f" ./exam notes {step['objective']['id']} what to know, the fastest commands, the traps, the docs to open"))
|
|
|
|
def cmd_next(args):
|
|
"""Start whatever is next on the path: study an objective, drill it, a mock, a simulator set."""
|
|
require_objectives()
|
|
step = next_step()
|
|
print(bold(f"Next on the path: {step['cmd']}") + dim(f"\n {step['why']}"))
|
|
if step["kind"] == "study":
|
|
cmd_study([step["objective"]["id"]] + args)
|
|
elif step["kind"] == "drill":
|
|
cmd_drill([step["objective"]["id"]] + args)
|
|
elif step["kind"] == "sim":
|
|
cmd_sim([step["set"]] + args)
|
|
else:
|
|
cmd_start((["--hard"] if step["kind"] == "done" else []) + args)
|
|
|
|
def paint_code(text):
|
|
"""`code` in a note: highlighted on a terminal, left as written (it is Markdown) in a pipe."""
|
|
return re.sub(r"`([^`]+)`", lambda m: bold(m.group(1)), text) if TTY else text
|
|
|
|
def print_notes(o, agg=None, studied=None):
|
|
agg = stats() if agg is None else agg
|
|
studied = load_study() if studied is None else studied
|
|
w = min(shutil.get_terminal_size((100, 24)).columns, 104)
|
|
def bullets(items):
|
|
for it in items:
|
|
print(paint_code("\n".join(textwrap.wrap(it, w, initial_indent=" - ", subsequent_indent=" "))))
|
|
print(bold(f"{o['id']} {o['key']}") + dim(f" · {DOMAINS[o['domain']]}"))
|
|
print(bold(o["text"])); print("-" * 64)
|
|
print(bold("Know")); bullets(o["know"])
|
|
print(bold("\nFast")); print("\n".join(" " + line for line in o["fast"].strip("\n").splitlines()))
|
|
print(bold("\nTraps")); bullets(o["traps"])
|
|
print(bold("\nDocs") + dim(" (pages you may open during the exam)"))
|
|
for u in o["docs"]:
|
|
print(" " + u)
|
|
print(bold("\nTasks"))
|
|
for q in objective_tasks(o, True):
|
|
print(f" {mark(task_state(q['id'], agg, studied))} Q{q['id']:<3} {q['title']}" + (dim(f" [{SETS.get(q['set'], q['set'])}]") if q.get("set") else ""))
|
|
print("-" * 64)
|
|
|
|
def notes_markdown(objs):
|
|
"""The same notes as a Markdown document (./exam notes all --md > CKAD_STUDY_GUIDE.md)."""
|
|
L = [f"# {TRACK_ID.upper()} study guide", "",
|
|
f"The {len(OBJECTIVES)} objectives of the {TRACK['name']} curriculum, in its own words and order: what to know, "
|
|
"the fastest commands, the traps, the documentation pages the exam lets you open, and the tasks in this "
|
|
"simulator that practise each one.", "",
|
|
f"_Generated by `./exam --track {TRACK_ID} notes all --md` from `tracks/{TRACK_ID}/curriculum.py` - edit that file, not this one._", "",
|
|
"```", f"./exam objectives # the path and your progress", "./exam study 3.2 # one objective: these notes, then its tasks, no clock, solutions open",
|
|
"./exam drill 3.2 # the same tasks against the clock", "```", ""]
|
|
for d in objective_domains():
|
|
mine = [o for o in objs if o["domain"] == d]
|
|
if not mine:
|
|
continue
|
|
L += [f"## {DOMAINS[d]}", ""]
|
|
for o in mine:
|
|
L += [f"### {o['id']} {o['text']}", "", f"`./exam study {o['id']}` (or `{o['key']}`)", "", "**Know**", ""]
|
|
L += [f"- {x}" for x in o["know"]]
|
|
L += ["", "**Fast**", "", "```", o["fast"].strip("\n"), "```", "", "**Traps**", ""]
|
|
L += [f"- {x}" for x in o["traps"]]
|
|
L += ["", "**Docs**", ""] + [f"- <{u}>" for u in o["docs"]]
|
|
L += ["", "**Tasks**", ""]
|
|
L += [f"- Q{q['id']} {q['title']}" + (f" _({SETS.get(q['set'], q['set'])})_" if q.get("set") else "") for q in objective_tasks(o, True)]
|
|
L.append("")
|
|
return "\n".join(L)
|
|
|
|
def cmd_notes(args):
|
|
"""Study notes for an objective, a domain or the whole curriculum."""
|
|
require_objectives()
|
|
rest = [a for a in args if not a.startswith("--")]
|
|
if not rest:
|
|
st = load_state()
|
|
o = find_objective(st.get("objective")) if st and not st.get("ended_at") and st.get("objective") else None
|
|
objs = [o or next_objective()[0] or OBJECTIVES[0]]
|
|
elif rest[0] == "all":
|
|
objs = list(OBJECTIVES)
|
|
elif rest[0].upper() in DOMAINS:
|
|
objs = [o for o in OBJECTIVES if o["domain"] == rest[0].upper()]
|
|
else:
|
|
objs = [find_objective(a) for a in rest]
|
|
if None in objs:
|
|
die(f"'{rest[objs.index(None)]}' is not an objective. Ids and keys:\n "
|
|
+ "\n ".join(f"{o['id']:<5}{o['key']:<16}{o['text']}" for o in OBJECTIVES))
|
|
if "--md" in args:
|
|
print(notes_markdown(objs)); return
|
|
agg, studied = stats(), load_study()
|
|
for i, o in enumerate(objs):
|
|
if i:
|
|
print()
|
|
print_notes(o, agg, studied)
|
|
if len(objs) == 1:
|
|
print(dim(f"./exam study {objs[0]['id']} its tasks, no clock, solutions open ./exam drill {objs[0]['id']} against the clock"))
|
|
|
|
# ------------------------------------------------------------ session ----
|
|
def preflight():
|
|
out = subprocess.run(["kubectl", "--context", CONTEXT, "get", "nodes", "-o", "jsonpath={.items[*].metadata.name}"],
|
|
capture_output=True, text=True).stdout.split()
|
|
missing = [v for v in NODES.values() if v not in out]
|
|
if missing:
|
|
die(f"Nodes not found in cluster: {missing}\n cluster has: {out}\n"
|
|
" Set CKA_CP, CKA_W1, CKA_W2, CKA_W3 to your node names (needs 1 control-plane + 3 workers).")
|
|
rc, _ = run('nx "$CP" "true" </dev/null && nx "$W1" "true" </dev/null && nx "$W2" "true" </dev/null && nx "$W3" "true" </dev/null', timeout=60)
|
|
if rc != 0:
|
|
die(f"Cannot run commands on nodes with CKA_NODE_ACCESS={NODE_ACCESS}.\n"
|
|
f" podman/docker -> node names must be kind container names (podman ps --filter label=io.x-k8s.kind.cluster={CLUSTER})\n"
|
|
" ssh -> passwordless ssh to each node name must work\n"
|
|
" vagrant -> run from the Vagrantfile directory; names are vagrant machine names")
|
|
ver = server_version() or "?"
|
|
print(dim(f" track: {TRACK_ID} cluster: {CONTEXT} ({ver}) node access: {NODE_ACCESS} nodes: {', '.join(NODES.values())}"))
|
|
# The real exams pin one minor each; practising on another one teaches the wrong defaults.
|
|
if not ver.startswith(K8S + "."):
|
|
fix = f"rebuild it: {setup_hint()} --recreate" if CONTEXT == f"kind-{CLUSTER}" else "point CKA_CONTEXT at a cluster on that minor"
|
|
print(yellow(f" warning: this cluster runs {ver}, but the real {TRACK_ID.upper()} exam runs Kubernetes {K8S} - {fix}"))
|
|
for tool, why in (("helm", "Helm tasks (the real exam has helm on the terminal)"), ("openssl", "certificate tasks"), ("curl", "tasks that fetch manifests")):
|
|
if not shutil.which(str(tool)):
|
|
print(yellow(f" warning: '{tool}' not found on this machine - needed for {why}"))
|
|
|
|
HARD_CONTEXTS = {"k8s-c1-H": "default", "k8s-c2-AC": "%{ns}", "k8s-c3-CCC": "kube-system"}
|
|
|
|
def context_namespace(ctx, template, params_by_id):
|
|
"""Default namespace for one exam context.
|
|
|
|
Several tasks can name the same context while working in different namespaces, but a
|
|
context has only one default namespace - so it belongs to whichever task actually reads it
|
|
("ctx_ns": True, i.e. the task says to work in the context's namespace without -n). Every
|
|
other task states its namespace in its own text, so the value does not matter to them.
|
|
"""
|
|
if "%{ns}" not in template:
|
|
return template
|
|
def ns_of(pred):
|
|
return [p["ns"] for i, p in params_by_id.items()
|
|
if p.get("ctx") == ctx and p.get("ns") and pred(QBYID.get(int(i)) or {})]
|
|
return (ns_of(lambda q: q.get("ctx_ns")) or ns_of(lambda q: True)
|
|
or [p["ns"] for p in params_by_id.values() if p.get("ns")] or ["default"])[0]
|
|
|
|
def setup_contexts(hard, params_by_id):
|
|
"""Exam-style contexts on the same cluster with different default namespaces.
|
|
|
|
Created whenever a task in the session asks for one - 24 tasks open with
|
|
`kubectl config use-context k8s-c?-?` and are unfollowable without it - not only when the
|
|
whole session was started with --hard. A drill that names hard task ids explicitly
|
|
('./exam drill 121', './exam drill weak') pulls those tasks in with hard=False.
|
|
|
|
Only tracks with TRACK["exam_contexts"] (CKA) use them, so only those tracks create or delete
|
|
them: the others run on another cluster, and must not repoint or remove the CKA contexts.
|
|
"""
|
|
if not TRACK.get("exam_contexts"):
|
|
subprocess.run(["kubectl", "config", "use-context", CONTEXT], capture_output=True)
|
|
return
|
|
cluster = subprocess.run(["kubectl", "config", "view", "-o", f"jsonpath={{.contexts[?(@.name=='{CONTEXT}')].context.cluster}}"],
|
|
capture_output=True, text=True).stdout.strip()
|
|
user = subprocess.run(["kubectl", "config", "view", "-o", f"jsonpath={{.contexts[?(@.name=='{CONTEXT}')].context.user}}"],
|
|
capture_output=True, text=True).stdout.strip()
|
|
for name in HARD_CONTEXTS:
|
|
subprocess.run(["kubectl", "config", "delete-context", name], capture_output=True)
|
|
wanted = set(HARD_CONTEXTS) if hard else {p.get("ctx") for p in params_by_id.values()} & set(HARD_CONTEXTS)
|
|
if not wanted:
|
|
subprocess.run(["kubectl", "config", "use-context", CONTEXT], capture_output=True)
|
|
return
|
|
for name in sorted(wanted):
|
|
subprocess.run(["kubectl", "config", "set-context", name, f"--cluster={cluster}", f"--user={user}",
|
|
f"--namespace={context_namespace(name, HARD_CONTEXTS[name], params_by_id)}"], capture_output=True)
|
|
# Hard mode starts you in c1 like the real exam; a mixed drill stays on the normal context
|
|
# so its non-hard tasks behave as usual and the hard ones still switch as instructed.
|
|
start = "k8s-c1-H" if hard and "k8s-c1-H" in wanted else CONTEXT
|
|
subprocess.run(["kubectl", "config", "use-context", start], capture_output=True)
|
|
|
|
def _log_msg(*args: Any, **kwargs: Any) -> None:
|
|
print(*args, **kwargs)
|
|
|
|
|
|
# Disruptive tasks (see "breaks": ... in the question bank) corrupt shared control-plane /
|
|
# node state. A session's setup only touches the questions it actually contains, so damage
|
|
# from a task in an earlier session lingers if that task isn't picked again - e.g. Q39 wrecks
|
|
# W1's kubelet config, and a later session with different questions never undoes it. This
|
|
# restores every known "breaks" target to a good baseline before each session seeds, so
|
|
# leftover damage never leaks across sessions regardless of which questions get sampled.
|
|
RESTORE_BASELINE = r"""
|
|
# Static-pod manifests are only ever rewritten when an edit actually changes them, and through a
|
|
# hidden temp file the kubelet ignores. `sed -i` writes its temp file (sedXXXXXX) into the SAME
|
|
# directory, the kubelet loads that as one more static pod and kills the running one: every session
|
|
# start used to knock the scheduler and controller-manager into CrashLoopBackOff for up to minutes
|
|
# while the session's tasks were being seeded (new Namespaces got no ServiceAccount, so bare Pods
|
|
# were rejected; Pods sat unscheduled). The user's cluster had 10 such restarts by 2026-09-24.
|
|
MEDIT='medit() { f=$1; shift; t=$(mktemp "${f%/*}/.medit.XXXXXX") || return 1
|
|
if sed "$@" "$f" > "$t" 2>/dev/null && [ "$(md5sum < "$t")" != "$(md5sum < "$f")" ]; then
|
|
chmod --reference="$f" "$t" 2>/dev/null; mv -f "$t" "$f"; else rm -f "$t"; fi; }'
|
|
|
|
# apiserver FIRST: Q66 renames the command to "kube-apiserverr", which takes the whole API
|
|
# down - every kubectl-based step below would fail until this is undone. The repair goes over
|
|
# podman/ssh (nx), not kubectl, so it works precisely when the cluster is unreachable.
|
|
if nx "$CP" 'grep -q "^ *- kube-apiserverr$" /etc/kubernetes/manifests/kube-apiserver.yaml' </dev/null 2>/dev/null; then
|
|
nx "$CP" "$MEDIT"'
|
|
medit /etc/kubernetes/manifests/kube-apiserver.yaml -e "s|^\( *\)- kube-apiserverr$|\1- kube-apiserver|"' </dev/null
|
|
# the kubelet re-reads the manifest on its own; wait for the API to answer again
|
|
for _ in $(seq 1 30); do $K get --raw=/readyz >/dev/null 2>&1 && break; sleep 2; done
|
|
fi
|
|
|
|
# apiserver audit flags (CKS Q9, whose track shares this cluster with CKA): a half-finished answer -
|
|
# flags without the volume, or a hostPath of type File whose policy file is gone - keeps the API server
|
|
# down for every later session. Touched only when the API server is down and audit flags are present,
|
|
# so a working audit setup is left alone.
|
|
if nx "$CP" 'grep -q -- "--audit-" /etc/kubernetes/manifests/kube-apiserver.yaml' </dev/null 2>/dev/null; then
|
|
for _ in $(seq 1 10); do $K get --raw=/readyz >/dev/null 2>&1 && break; sleep 3; done
|
|
if ! $K get --raw=/readyz >/dev/null 2>&1; then
|
|
nx "$CP" 'mkdir -p /var/log/kubernetes
|
|
[ -f /etc/kubernetes/audit-policy.yaml ] || printf "apiVersion: audit.k8s.io/v1\nkind: Policy\nrules:\n- level: None\n" > /etc/kubernetes/audit-policy.yaml' </dev/null
|
|
nx "$CP" "$MEDIT"'
|
|
medit /etc/kubernetes/manifests/kube-apiserver.yaml -e "/--audit-/d"' </dev/null
|
|
for _ in $(seq 1 45); do $K get --raw=/readyz >/dev/null 2>&1 && break; sleep 2; done
|
|
fi
|
|
fi
|
|
|
|
# scheduler: undo the "kube-scheduler" -> "kube-schedulerr" command typo (Q20/Q61), and put the
|
|
# manifest back if a task had the candidate stop the scheduler by moving it out (Q97) and they
|
|
# never moved it back - that setup keeps a pristine copy outside the static pod directory.
|
|
nx "$CP" "$MEDIT"'
|
|
f=/etc/kubernetes/manifests/kube-scheduler.yaml
|
|
[ -f $f ] || { [ -f /etc/kubernetes/.cka-sim/kube-scheduler.yaml ] && cp /etc/kubernetes/.cka-sim/kube-scheduler.yaml $f; }
|
|
medit $f -e "s|^\( *\)- kube-schedulerr$|\1- kube-scheduler|"
|
|
' </dev/null
|
|
|
|
# controller-manager: undo Q38's injected flag and Q61's broken image tag. The last two lines
|
|
# undo the retired set B's Q317 (the CSR signing controller switched off in --controllers, or in
|
|
# its older form --cluster-signing-cert-file pointed at a missing file, which crash-looped the WHOLE
|
|
# controller-manager); a cluster that ever ran that set may still carry either.
|
|
nx "$CP" "$MEDIT"'
|
|
medit /etc/kubernetes/manifests/kube-controller-manager.yaml \
|
|
-e "/--leader-elect-lease-duration=1s/d" \
|
|
-e "s|\(image: registry.k8s.io/kube-controller-manager:v[0-9.]*\)-broken|\1|" \
|
|
-e "s|--cluster-signing-cert-file=/etc/kubernetes/pki/ca-signing.crt|--cluster-signing-cert-file=/etc/kubernetes/pki/ca.crt|" \
|
|
-e "s|,-certificatesigningrequest-signing-controller||; s|,-csrsigning||"
|
|
' </dev/null
|
|
|
|
# coredns: restore the default upstream forward (Q37/Q57 point it somewhere bogus)
|
|
cf=$($K -n kube-system get cm coredns -o jsonpath='{.data.Corefile}' 2>/dev/null)
|
|
if [ -n "$cf" ] && ! printf '%s' "$cf" | grep -Eq '^[[:space:]]*forward \. /etc/resolv\.conf'; then
|
|
printf '%s' "$cf" | sed -E 's|^([[:space:]]*forward \. )[^[:space:]{]+|\1/etc/resolv.conf|' > /tmp/Corefile.restore
|
|
$K -n kube-system create cm coredns --from-file=Corefile=/tmp/Corefile.restore --dry-run=client -o yaml | $K apply -f - >/dev/null 2>&1
|
|
$K -n kube-system rollout restart deploy coredns >/dev/null 2>&1
|
|
fi
|
|
# ...and Q57's other two faults. kindnet ENFORCES NetworkPolicy, so a leftover lockdown policy on
|
|
# the CoreDNS pods kills name resolution for the whole cluster in every later session, and so does
|
|
# a Deployment left at 0 replicas.
|
|
$K -n kube-system delete netpol maintenance-lockdown --ignore-not-found >/dev/null 2>&1
|
|
r=$($K -n kube-system get deploy coredns -o jsonpath='{.spec.replicas}' 2>/dev/null)
|
|
if [ -n "$r" ] && [ "$r" -lt 2 ]; then $K -n kube-system scale deploy coredns --replicas=2 >/dev/null 2>&1; fi
|
|
|
|
# w1/w2/w3 kubelet: undo Q17/Q39-style tampering, then make sure it's enabled and running.
|
|
# staticPodPath is normalised unconditionally: Q46 relocates it to /etc/kubernetes/static, and
|
|
# the config.yaml.orig copy below only helps if something made that backup first - nothing does,
|
|
# so a relocated path used to survive every later session and silently break Q05 (whose check
|
|
# looks in /etc/kubernetes/manifests). Only restart the kubelet if we actually changed the file.
|
|
# clientCAFile gets the same treatment: Q39 points it at ca-bundle.crt and Q60 at ca-wrong.crt, and
|
|
# only W1 ever gets a config.yaml.orig (Q39 makes it), so an unfinished Q60 left W2's kubelet
|
|
# crash-looping in every later session - enable/start below cannot help a kubelet that exits.
|
|
# The systemd drop-in is checked too: Q92 points ExecStart at a kubelet binary that does not exist.
|
|
for n in "$W1" "$W2" "$W3"; do
|
|
nx "$n" '
|
|
c=/var/lib/kubelet/config.yaml; d=/etc/systemd/system/kubelet.service.d/10-kubeadm.conf
|
|
[ -f $c.orig ] && cp -f $c.orig $c
|
|
restart=0
|
|
if grep -q "^staticPodPath:" $c 2>/dev/null && ! grep -q "^staticPodPath: /etc/kubernetes/manifests$" $c; then
|
|
sed -i "s|^staticPodPath: .*|staticPodPath: /etc/kubernetes/manifests|" $c; restart=1
|
|
fi
|
|
ca=$(awk "/clientCAFile:/{print \$2}" $c 2>/dev/null)
|
|
if [ -n "$ca" ] && [ ! -f "$ca" ]; then
|
|
sed -i "s|clientCAFile: .*|clientCAFile: /etc/kubernetes/pki/ca.crt|" $c; restart=1
|
|
fi
|
|
bin=$(sed -n "s|^ExecStart=\([^ ]*kubelet\) .*|\1|p" $d 2>/dev/null | tail -1)
|
|
if [ -n "$bin" ] && [ ! -x "$bin" ]; then
|
|
sed -i "s|^ExecStart=$bin |ExecStart=/usr/bin/kubelet |" $d; systemctl daemon-reload; restart=1
|
|
fi
|
|
[ $restart = 1 ] && systemctl restart kubelet >/dev/null 2>&1
|
|
systemctl enable kubelet >/dev/null 2>&1
|
|
systemctl start kubelet >/dev/null 2>&1
|
|
' </dev/null
|
|
done
|
|
|
|
# static pods the bank drops on nodes: Q05's static-web on W3, Q46's relocated copies, Q94's
|
|
# ops-beacon on the control plane - and edge-cache, left on W3 by the retired set B (it sat in
|
|
# ContainerCreating for days, its hostPath deliberately missing). Each setup recreates what its
|
|
# own session needs.
|
|
nx "$W3" 'rm -f /etc/kubernetes/manifests/static-web*.yaml /etc/kubernetes/manifests/edge-cache.yaml /etc/kubernetes/static/*.yaml' </dev/null
|
|
nx "$CP" 'rm -f /etc/kubernetes/manifests/ops-beacon*.yaml' </dev/null
|
|
|
|
# default StorageClass: Q16 makes fast-local the default and strips `standard`. Leave exactly one
|
|
# default - `standard` - so "the default class" means the same thing in every session.
|
|
if $K get sc standard >/dev/null 2>&1; then
|
|
for sc in $($K get sc -o jsonpath='{.items[*].metadata.name}'); do
|
|
[ "$sc" = standard ] && continue
|
|
[ "$($K get sc "$sc" -o jsonpath='{.metadata.annotations.storageclass\.kubernetes\.io/is-default-class}')" = true ] \
|
|
&& $K annotate sc "$sc" storageclass.kubernetes.io/is-default-class- >/dev/null 2>&1
|
|
done
|
|
$K annotate sc standard storageclass.kubernetes.io/is-default-class=true --overwrite >/dev/null 2>&1
|
|
fi
|
|
|
|
# w1-taint: clear every randomized demo taint key the bank can apply. Q25 draws from
|
|
# dedicated/team/workload and Q50 from pool/lane; `maintenance` was the retired set B's and may
|
|
# still be on a cluster that ran it. Miss one and that node silently stops accepting pods.
|
|
# One key per call: `kubectl taint node N a- b-` aborts on the first key that is not present and
|
|
# removes NOTHING, so the single taint that is actually set survives the "cleanup" and W1 keeps
|
|
# rejecting pods in every later session - the exact failure this block exists to prevent.
|
|
for _t in dedicated workload team pool lane maintenance; do
|
|
$K taint node "$W1" "$_t-" >/dev/null 2>&1 || true
|
|
done
|
|
|
|
# metrics-server: Q23 seeds its scenario by stripping --kubelet-insecure-tls, and kind's kubelet
|
|
# serving certs have no IP SANs, so without that flag metrics-server never becomes Ready. Left
|
|
# broken it does far more than disable `kubectl top`: the v1beta1.metrics.k8s.io APIService has no
|
|
# endpoints, aggregated discovery goes stale, and then EVERY namespace deletion in the cluster
|
|
# hangs on NamespaceDeletionDiscoveryFailure. Baseline runs before seeding, so a session that
|
|
# actually contains Q23 re-breaks it a moment later, exactly as intended.
|
|
if $K -n kube-system get deploy metrics-server >/dev/null 2>&1 \
|
|
&& ! $K -n kube-system get deploy metrics-server -o jsonpath='{.spec.template.spec.containers[0].args}' | grep -q 'kubelet-insecure-tls'; then
|
|
$K -n kube-system patch deploy metrics-server --type=json \
|
|
-p '[{"op":"add","path":"/spec/template/spec/containers/0/args/-","value":"--kubelet-insecure-tls"}]' >/dev/null 2>&1 || true
|
|
fi
|
|
|
|
# cordons: Q60 cordons W2, Q109 has the control-plane node drained, and a drain task left half-done
|
|
# leaves a node cordoned. A cordon left behind fails any later task that needs every worker (Q26's
|
|
# DaemonSet coverage, Q50's spread) or a Pod on the control-plane node (Q96, Q109 itself).
|
|
for n in "$CP" "$W1" "$W2" "$W3"; do $K uncordon "$n" >/dev/null 2>&1 || true; done
|
|
|
|
# Deliberately NOT restored, and why:
|
|
# etcd (Q45) and certs (Q65). Neither setup damages anything - Q45 only takes a snapshot
|
|
# and deletes a ConfigMap, Q65 only records the current notBefore. Any damage comes from a
|
|
# half-finished candidate attempt (etcd.yaml left pointing at a restored data-dir, static pod
|
|
# manifests moved to /tmp and never moved back), which is not deterministic: there is no way
|
|
# to tell "mid-attempt" from "correctly completed" without guessing, and guessing wrong here
|
|
# rewrites etcd's data-dir or re-issues certs on a healthy cluster. Re-run ./setup.sh (or
|
|
# ./teardown.sh && ./setup.sh) if one of those leaves the control plane wedged.
|
|
"""
|
|
|
|
def restore_baseline(log: Callable[..., Any] = _log_msg):
|
|
log(dim(" resetting control-plane / node state to a clean baseline..."))
|
|
rc, out = run(RESTORE_BASELINE, timeout=90)
|
|
if rc != 0:
|
|
log(yellow(f" warning: baseline restore had issues (continuing anyway):\n{dim(out[-500:])}"))
|
|
|
|
|
|
def seed(qs, log: Callable[..., Any] = _log_msg, params_by_id=None):
|
|
params_by_id = params_by_id or {}
|
|
normal = [q for q in qs if not q.get("disruptive")]
|
|
disruptive = [q for q in qs if q.get("disruptive")]
|
|
def one(q):
|
|
try:
|
|
rc, out = run(q["setup"], params=params_by_id.get(str(q["id"])), timeout=240)
|
|
except subprocess.TimeoutExpired:
|
|
rc, out = 1, "setup timed out after 240s"
|
|
log(f" {'✓' if rc == 0 else '!'} Q{q['id']:02d} {q['title']}" + ("" if rc == 0 else f"\n{dim(out[-300:])}"))
|
|
for q in normal:
|
|
one(q)
|
|
if disruptive:
|
|
log(dim(" waiting up to 45s for seeded pods to be scheduled..."))
|
|
for _ in range(15):
|
|
pending = subprocess.run(["kubectl", "--context", CONTEXT, "get", "pods", "-A", "--field-selector=spec.nodeName=",
|
|
"-o", "name"], capture_output=True, text=True).stdout.strip()
|
|
if not pending:
|
|
break
|
|
time.sleep(3)
|
|
for q in disruptive:
|
|
one(q)
|
|
|
|
def open_sessions_elsewhere():
|
|
"""Tracks other than this one with a session that has not ended. Sessions share the kubeconfig's
|
|
current context (and kubectl version), so starting one would pull the other's terminal over to
|
|
another cluster while its grading stays on the old one: silent zeros."""
|
|
out = []
|
|
for t in TRACKS:
|
|
if t == TRACK_ID:
|
|
continue
|
|
try:
|
|
with open(os.path.join(STATE_DIR, f"state-{t}.json")) as f:
|
|
if not json.load(f).get("ended_at"):
|
|
out.append(t)
|
|
except (OSError, ValueError):
|
|
pass
|
|
return out
|
|
|
|
def use_kubectl():
|
|
"""Point .state/kubectl/current at the kubectl for this track's minor, or at your own kubectl
|
|
('host') when setup.sh could not fetch one. examrc puts current/ first on PATH; it always exists
|
|
once setup.sh has run, so a shell that already hashed current/kubectl follows the switch."""
|
|
if not os.path.isdir(KUBECTL_DIR):
|
|
return
|
|
target = K8S
|
|
if not os.path.isfile(os.path.join(KUBECTL_DIR, K8S, "kubectl")):
|
|
target = "host"
|
|
mine = os.path.realpath(KUBECTL_DIR) # never our own current/ - PATH may spell it via a symlink
|
|
own = shutil.which("kubectl", path=os.pathsep.join(
|
|
p for p in os.environ.get("PATH", "").split(os.pathsep) if not os.path.realpath(p).startswith(mine)))
|
|
if not own:
|
|
return
|
|
link = os.path.join(KUBECTL_DIR, "host", "kubectl")
|
|
os.makedirs(os.path.dirname(link), exist_ok=True)
|
|
if os.path.realpath(link) != os.path.realpath(own):
|
|
if os.path.lexists(link):
|
|
os.remove(link)
|
|
os.symlink(os.path.realpath(own), link)
|
|
cur = os.path.join(KUBECTL_DIR, "current")
|
|
tmp = cur + ".new"
|
|
try:
|
|
if os.path.lexists(tmp):
|
|
os.remove(tmp)
|
|
os.symlink(target, tmp)
|
|
os.replace(tmp, cur) # atomic: a terminal never sees current/ missing
|
|
except OSError:
|
|
pass
|
|
|
|
def begin_session(qs, mode, duration, force, log: Callable[..., Any] = _log_msg, hard=False, untimed_session=False, other_ok=None,
|
|
objective=None):
|
|
st = load_state()
|
|
if st and not st.get("ended_at") and not force:
|
|
die("A session is already in progress. './exam end' it first, or add --force.")
|
|
others = open_sessions_elsewhere()
|
|
if others and not (force if other_ok is None else other_ok):
|
|
t = others[0]
|
|
how = (f"switch to {t.upper()} in the lobby and submit it, or run ./exam --track {t} end (abort --yes discards it)"
|
|
if other_ok is False else f"./exam --track {t} end (or abort --yes), or add --force here")
|
|
die(f"A {t.upper()} session is still open. Tracks run on their own clusters but share your kubectl context,\n"
|
|
f" so finish that one first: {how}.")
|
|
if others:
|
|
log(yellow(f" warning: the open {'/'.join(o.upper() for o in others)} session loses its kubectl context and kubectl"
|
|
f" version to this {TRACK_ID.upper()} session; end it before going back to it."))
|
|
if not kubectl_ok():
|
|
die(f"Cannot reach the {TRACK_ID.upper()} cluster through kubectl context '{CONTEXT}' (Kubernetes {K8S}).\n"
|
|
f" Create or restart it: {setup_hint()} (or set CKA_CONTEXT for your own cluster)")
|
|
if not qs:
|
|
die(f"Track '{TRACK_ID}' has no questions yet.")
|
|
preflight()
|
|
update_milestones() # stamp what the history already proves, so this session only announces its own
|
|
os.makedirs(WORKSPACE, exist_ok=True)
|
|
rng = random.Random()
|
|
params_by_id = {str(q["id"]): make_params(q, rng) for q in qs}
|
|
setup_contexts(hard, params_by_id)
|
|
use_kubectl()
|
|
restore_baseline(log)
|
|
log(bold(f"Seeding {len(qs)} scenario(s){' [HARD]' if hard else ''}..."))
|
|
seed(qs, log, params_by_id)
|
|
st = {"mode": mode, "track": TRACK_ID, "hard": hard, "untimed": untimed_session, "ids": [q["id"] for q in qs],
|
|
"params": params_by_id, "duration": duration, "started_at": time.time(), "ended_at": None,
|
|
"k8s": server_version(), "results": {}, "flags": [], "notes": "",
|
|
"objective": objective} # the curriculum objective a study/drill session was started for, if any
|
|
save_state(st)
|
|
log("")
|
|
log(bold("=" * 64))
|
|
kind = "STUDY" if untimed_session else ("EXAM" if mode == "exam" else "DRILL")
|
|
log(bold(f" {TRACK_ID.upper()} {kind}{' · HARD' if hard else ''} - {'no clock, solutions open' if untimed_session else 'clock started'}"))
|
|
log(f" Cluster : {CONTEXT} (Kubernetes {st['k8s'] or K8S}, as the real {TRACK_ID.upper()} runs {K8S})")
|
|
if hard and TRACK.get("exam_contexts"):
|
|
log(yellow(" Contexts: k8s-c1-H (default) · k8s-c2-AC · k8s-c3-CCC - each task tells you which to use; you are on k8s-c1-H now"))
|
|
if not untimed_session:
|
|
log(f" Ends : {datetime.fromtimestamp(st['started_at'] + duration).strftime('%H:%M:%S')} ({duration // 60} min)")
|
|
log(f" Tasks : {', '.join('Q%02d' % q['id'] for q in qs) if mode == 'drill' else len(qs)}"
|
|
+ ("" if untimed_session else f" Pass mark: {PASS_PCT}%"))
|
|
log(bold("=" * 64))
|
|
log(" ./exam web exam-style browser UI (task pane + terminal + clock)\n"
|
|
+ (" ./exam list / show N / grade N / solution N / end terminal workflow\n" if untimed_session else
|
|
" ./exam list / show N / grade N / timer / end terminal workflow\n")
|
|
+ (" Tip: read ALL tasks first. Some are cluster-wide failures that block others.\n" if mode == "exam" else "")
|
|
+ f" Files the tasks ask for go under {WS_DISPLAY}/ (relative to this directory)\n"
|
|
+ " Your own terminal: source ./examrc (fish: source ./examrc.fish) -> k, $do, $now, completion")
|
|
return st
|
|
|
|
def finish_session(st, log: Callable[..., Any] = _log_msg):
|
|
qs = session_questions(st)
|
|
over = remaining(st) <= 0
|
|
set_paused(st, False)
|
|
log(bold(f"Submitting - grading {len(qs)} task(s)..."))
|
|
for q in qs:
|
|
r = grade_one(q)
|
|
st["results"][str(q["id"])] = r
|
|
tag = green("PASS") if r["pass"] else (yellow(f"{r['earned']}/{r['total']}") if r["earned"] else red("FAIL"))
|
|
log(f"Q{q['id']:02d} {tag} {q['title']}" + ("" if r["pass"] else f"\n{dim(result_detail(r))}"))
|
|
st["ended_at"] = time.time()
|
|
st["over_time"] = over
|
|
save_state(st)
|
|
setup_contexts(False, {})
|
|
entry = None
|
|
if untimed(st):
|
|
record_study(st)
|
|
else:
|
|
entry = record_attempt(st)
|
|
# objectives this session got over a line: shown by ./exam end and by the results page
|
|
st["milestones"] = st.get("milestones", []) + milestone_lines(update_milestones())
|
|
save_state(st)
|
|
return entry
|
|
|
|
# ----------------------------------------------------------- commands ----
|
|
def cmd_start(args):
|
|
hard = "--hard" in args; force = "--force" in args
|
|
undrilled = [d for d, n in drilled_domains().items() if n == 0]
|
|
if undrilled and not force:
|
|
die("Drill every domain at least once before a full exam (that is the point of the drills).\n"
|
|
f" Not yet drilled: {', '.join(undrilled)}\n Run: ./exam drill (picks the next domain)\n"
|
|
" or: ./exam start --force to skip the gate.")
|
|
begin_session(compose_exam(hard, random.Random()), "exam", EXAM_SECONDS, force, hard=hard)
|
|
|
|
def next_drill(hard, rng):
|
|
"""No-arg drill: the least-drilled domain, DRILL_SIZE tasks, least-recently-seen first."""
|
|
cov = drilled_domains()
|
|
domain = min(DOMAINS, key=lambda d: (cov[d], list(DOMAINS).index(d)))
|
|
return domain, pick(domain, DRILL_SIZE, hard, rng)
|
|
|
|
def cmd_drill(args):
|
|
force = "--force" in args; hard = "--hard" in args
|
|
args = [a for a in args if a not in ("--force", "--hard")]
|
|
minutes = None
|
|
if "--time" in args:
|
|
i = args.index("--time"); minutes = int(args[i + 1]); del args[i:i + 2]
|
|
per_task = DRILL_SECONDS_PER_TASK_HARD if hard else DRILL_SECONDS_PER_TASK
|
|
if not args:
|
|
domain, qs = next_drill(hard, random.Random())
|
|
cov = drilled_domains()
|
|
print(bold(f"Drill: {DOMAINS[domain]}") + dim(f" ({sum(1 for v in cov.values() if v)}/{len(DOMAINS)} domains drilled so far)"))
|
|
begin_session(qs, "drill", (minutes * 60) if minutes else per_task * len(qs), force, hard=hard)
|
|
return
|
|
if args[0] == "all":
|
|
label, qs = campaign_batch(hard, random.Random())
|
|
if not qs:
|
|
print(bold("Every task in the bank has been attempted." ) + " Switching to weak tasks.")
|
|
qs, _ = split_solo([q for q in QUESTIONS if q["id"] in weak_ids()])
|
|
qs = no_conflicts(qs, DRILL_SIZE)
|
|
if not qs:
|
|
die("Nothing weak either - you have passed everything on its latest attempt. Time for ./exam start --hard.")
|
|
else:
|
|
print(bold(f"Campaign: {label}") + dim(f" this batch: Q{', Q'.join(str(q['id']) for q in qs)}"))
|
|
begin_session(qs, "drill", (minutes * 60) if minutes else per_task * len(qs), force, hard=hard)
|
|
return
|
|
objective = None
|
|
if args[0] == "weak":
|
|
ids = weak_ids()
|
|
if not ids:
|
|
die("Nothing marked weak - finish a session with './exam end' first, or name tasks: ./exam drill 2 17")
|
|
else:
|
|
ids = []
|
|
for a in args:
|
|
if a.lower().replace("set", "") in SETS and a.lower().startswith("set"):
|
|
ids += [q["id"] for q in set_questions(a.lower().replace("set", ""))]
|
|
elif a.upper() in DOMAINS:
|
|
ids += [q["id"] for q in QUESTIONS if q["domain"] == a.upper()]
|
|
elif a.isdigit():
|
|
ids.append(q_by_id(int(a))["id"])
|
|
elif find_objective(a):
|
|
sitting, open_n = objective_session(find_objective(a), "drill")
|
|
ids += [q["id"] for q in sitting]
|
|
objective = objective or find_objective(a)["id"]
|
|
if open_n > len(sitting):
|
|
print(dim(f" {find_objective(a)['id']}: {len(sitting)} of its {open_n} open tasks in this sitting - run it again for the rest."))
|
|
else:
|
|
die(f"'{a}' is not a task id, a domain ({', '.join(DOMAINS)}), a set "
|
|
f"({', '.join('set' + k for k in SETS) or 'none'})"
|
|
+ (" or an objective (./exam objectives)." if OBJECTIVES else "."))
|
|
ids = sorted(set(ids))
|
|
qs, solo = split_solo([q for q in QUESTIONS if q["id"] in ids])
|
|
if solo:
|
|
print(yellow(f"Dropped Q{', Q'.join(str(q['id']) for q in solo)}: restoring it rolls the whole cluster back and would "
|
|
f"wipe the other tasks. Drill it on its own: ./exam drill {solo[0]['id']}"))
|
|
kept = no_conflicts(qs, len(qs))
|
|
if len(kept) < len(qs):
|
|
dropped = [q["id"] for q in qs if q not in kept]
|
|
print(yellow(f"Dropped Q{', Q'.join(map(str, dropped))}: they break the same thing as another task in this set. Drill them separately."))
|
|
# Shuffle AFTER resolving conflicts, so which task gets dropped stays deterministic while the
|
|
# order you meet them in does not. A drill listed by id walks you from easy to hard in
|
|
# lockstep, which is the one ordering the real exam will never give you.
|
|
random.Random().shuffle(kept)
|
|
begin_session(kept, "drill", (minutes * 60) if minutes else per_task * len(kept), force, hard=hard,
|
|
objective=objective if len(args) == 1 else None)
|
|
|
|
def cmd_study(args):
|
|
"""No clock, solutions unlocked from the start, nothing written to history."""
|
|
hard = "--hard" in args; force = "--force" in args; with_sets = "--all" in args
|
|
rest = [a for a in args if not a.startswith("--")]
|
|
n = DRILL_SIZE
|
|
if "--count" in args:
|
|
n = int(args[args.index("--count") + 1]); rest = [a for a in rest if a != str(n)]
|
|
rng = random.Random()
|
|
obj = None
|
|
if rest and rest[0].upper() in DOMAINS:
|
|
domain = rest[0].upper()
|
|
qs = pick(domain, n, hard, rng)
|
|
print(bold(f"Study: {DOMAINS[domain]}"))
|
|
elif rest and rest[0] == "weak":
|
|
qs = no_conflicts(split_solo([q for q in QUESTIONS if q["id"] in weak_ids()])[0], n)
|
|
print(bold("Study: tasks you have been failing"))
|
|
elif OBJECTIVES and (not rest or rest[0] == "next" or find_objective(rest[0])):
|
|
# one objective of the curriculum: the one named, else the next one on the path
|
|
if rest and rest[0] != "next":
|
|
obj = find_objective(rest[0])
|
|
else:
|
|
agg, studied = stats(), load_study()
|
|
obj = next((o for o in path_objectives() if objective_state(o, agg, studied) in ("new", "started")), None)
|
|
if obj is None:
|
|
die("Every objective has been studied - name one to revisit (./exam objectives lists them),\n"
|
|
f" or carry on up the path: {next_step(agg, studied)['cmd']} (./exam next)")
|
|
qs, open_n = objective_session(obj, "study", with_sets, n if "--count" in args else None)
|
|
print(bold(f"Study: {obj['id']} {obj['text']}") + dim(f" {DOMAINS[obj['domain']]}"))
|
|
if open_n > len(qs):
|
|
print(dim(f" {len(qs)} of its {open_n} open tasks in this sitting - run the same command again for the rest."))
|
|
elif rest:
|
|
qs = no_conflicts(split_solo([q_by_id(int(a)) if a.isdigit() else die(
|
|
f"'{a}' is not a task id, a domain ({', '.join(DOMAINS)}), 'weak'"
|
|
+ (" or an objective (./exam objectives)." if OBJECTIVES else ".")) for a in rest])[0], len(rest))
|
|
print(bold("Study: " + ", ".join(f"Q{q['id']}" for q in qs)))
|
|
else:
|
|
domain, qs = next_drill(hard, rng)
|
|
print(bold(f"Study: {DOMAINS[domain]}"))
|
|
begin_session(qs, "drill", EXAM_SECONDS, force, hard=hard, untimed_session=True, objective=obj and obj["id"])
|
|
if obj:
|
|
print(); print_notes(obj)
|
|
print(dim(" ./exam show N read a task ./exam grade N check it ./exam solution N reference answer\n"
|
|
" ./exam notes these notes again ./exam end finish (./exam objectives then shows what you solved)"))
|
|
|
|
def cmd_sim(args):
|
|
"""A fixed simulator set, full exam rules and clock."""
|
|
force = "--force" in args
|
|
names = [a.lower() for a in args if not a.startswith("--")]
|
|
if not names or names[0] not in SETS:
|
|
die(f"usage: ./exam sim {'|'.join(SETS) or '<set>'} [--force]\n "
|
|
+ "\n ".join(f"{k}: {v} ({len(set_questions(k))} tasks)" for k, v in SETS.items()))
|
|
name = names[0]
|
|
qs = set_questions(name)
|
|
print(bold(f"{SETS[name]} - {len(qs)} tasks, {TRACK['minutes']} minutes, exam rules (no feedback until you submit)"))
|
|
begin_session(qs, "exam", EXAM_SECONDS, force, hard=True)
|
|
|
|
def cmd_list(args):
|
|
st = load_state()
|
|
results = st["results"] if st and "results" in st else {}
|
|
qs = session_questions(st) if st and not st.get("ended_at") else list(QUESTIONS)
|
|
agg = stats()
|
|
print(bold(f"{'#':>3} {'Domain':<8}{'Pts':>4} {'Now':<8} {'History':<9} Title"))
|
|
for q in qs:
|
|
r = results.get(str(q["id"]))
|
|
now = "-" if r is None else (green("PASS") if r.get("pass") else yellow(f"{r.get('earned')}/{r.get('total')}"))
|
|
a = agg.get(int(q["id"])) or AggEntry()
|
|
hist = "" if a.attempts == 0 else f"{a.attempts - a.fails}/{a.attempts}"
|
|
if a.attempts and a.last is False:
|
|
hist = red(hist + " ✗")
|
|
print(f"{q['id']:>3} {q['domain']:<8}{q['weight']:>4} {now:<17} {hist:<18} {q['title']}{' [hard]' if q['difficulty'] == 'hard' else ''}")
|
|
print(dim("\nHistory = full passes/attempts across sessions, ✗ = failed most recently."))
|
|
if st and not st.get("ended_at"):
|
|
print(status_line(st))
|
|
|
|
def cmd_show(args):
|
|
if not args:
|
|
die("usage: ./exam show N")
|
|
q = q_by_id(int(args[0]))
|
|
st0 = load_state()
|
|
sq = session_questions(st0) if st0 and not st0.get("ended_at") else []
|
|
pos = f"{sq.index(q) + 1} of {len(sq)} | " if q in sq else ""
|
|
print(bold(f"{pos}Q{q['id']} | {DOMAINS[q['domain']]} | {q['weight']} pts{' | HARD' if q['difficulty'] == 'hard' else ''}"))
|
|
print(bold(q["title"])); print("-" * 64); print(task_text(q)); print("-" * 64)
|
|
st = load_state()
|
|
if st and not st.get("ended_at"):
|
|
print(status_line(st))
|
|
|
|
def cmd_grade(args):
|
|
st = require_started()
|
|
force = "--force" in args; args = [a for a in args if a != "--force"]
|
|
if not args:
|
|
die("usage: ./exam grade N | all")
|
|
# The web UI already refused this; the CLI handed out per-check feedback mid-exam, which a sim
|
|
# meant to rehearse the real thing (no feedback until you submit) must not do by default.
|
|
if st.get("mode") == "exam" and not st.get("ended_at") and not force:
|
|
die("No feedback during an exam - the real one gives none. Submit with ./exam end (--force to peek anyway).")
|
|
ids = [q["id"] for q in session_questions(st)] if args[0] == "all" else [int(a) for a in args]
|
|
for n in ids:
|
|
q = q_by_id(n); r = grade_one(q)
|
|
st["results"][str(n)] = r
|
|
tag = green("PASS") if r.get("pass") else yellow(f"{r.get('earned')}/{r.get('total')}")
|
|
print(f"Q{n:02d} {tag} {q['title']}"); print(dim(result_detail(r)))
|
|
save_state(st)
|
|
record_study(st, set(ids))
|
|
if untimed(st):
|
|
reached = milestone_lines(update_milestones())
|
|
if reached: # kept with the session, so ./exam end and the results page show them again
|
|
st["milestones"] = st.get("milestones", []) + reached; save_state(st)
|
|
for line in reached:
|
|
print(green(line))
|
|
if args[0] == "all":
|
|
print_score(st)
|
|
|
|
def print_score(st):
|
|
qs = session_questions(st)
|
|
earned, total, pct = score(st)
|
|
print(); print(bold("Per-domain:"))
|
|
for d, name in DOMAINS.items():
|
|
dq = [q for q in qs if q["domain"] == d]
|
|
if not dq:
|
|
continue
|
|
dt = sum(q["weight"] for q in dq); de = sum(st["results"].get(str(q["id"]), {}).get("earned", 0) for q in dq)
|
|
n = int(20 * de / dt)
|
|
print(f" {d:<7} {'█' * n}{'░' * (20 - n)} {de:>3}/{dt:<3} {dim(name)}")
|
|
verdict = green("PASS") if pct >= PASS_PCT else red("FAIL")
|
|
print(f"\n{bold('Score:')} {earned}/{total} = {pct}% -> {verdict} (pass mark {PASS_PCT}%)")
|
|
|
|
def status_line(st):
|
|
if untimed(st):
|
|
return green("study mode - no clock") + dim(f" [{len(session_questions(st))} tasks, kept out of the history]")
|
|
r = remaining(st)
|
|
if st.get("paused_at"):
|
|
return yellow(f"⏸ PAUSED - {fmt(r)} remaining") + dim(" ./exam resume")
|
|
if r <= 0:
|
|
return red("⏰ TIME IS UP - run ./exam end")
|
|
col = green if r > 30 * 60 else (yellow if r > 10 * 60 else red)
|
|
return col(f"⏱ {fmt(r)} remaining") + dim(f" [{st.get('mode', 'exam')}]")
|
|
|
|
def cmd_status(args):
|
|
st = require_started()
|
|
if st.get("ended_at"):
|
|
print("Session ended."); print_score(st); return
|
|
print(status_line(st))
|
|
print(dim(f"{sum(1 for v in st['results'].values() if v['pass'])} passed / {len(st['results'])} graded / {len(session_questions(st))} in session"))
|
|
|
|
def cmd_timer(args):
|
|
st = require_started()
|
|
if st.get("ended_at"):
|
|
die("Session already ended.")
|
|
if untimed(st):
|
|
die("This is a study session - there is no clock.")
|
|
try:
|
|
while True:
|
|
st = load_state(); r = remaining(st)
|
|
if st.get("paused_at"):
|
|
sys.stdout.write("\r" + yellow(bold(f"⏸ {fmt(r)} PAUSED ")) + dim(" ./exam resume ")); sys.stdout.flush(); time.sleep(1); continue
|
|
if r <= 0:
|
|
print("\r" + red(bold("⏰ 00:00:00 TIME IS UP - run ./exam end "))); break
|
|
col = green if r > 30 * 60 else (yellow if r > 10 * 60 else red)
|
|
sys.stdout.write("\r" + col(bold(f"⏱ {fmt(r)} remaining")) + dim(" (Ctrl-C to hide)")); sys.stdout.flush()
|
|
if int(r) in (30 * 60, 15 * 60, 5 * 60) and TTY:
|
|
sys.stdout.write("\a")
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
print()
|
|
|
|
def cmd_pause(args):
|
|
st = require_started()
|
|
if st.get("ended_at"):
|
|
die("Session already ended.")
|
|
set_paused(st, True); print(status_line(st))
|
|
|
|
def cmd_resume(args):
|
|
st = require_started()
|
|
set_paused(st, False); print(status_line(st))
|
|
|
|
def cmd_end(args):
|
|
st = require_started()
|
|
if st.get("ended_at"):
|
|
die("Session already ended and recorded.")
|
|
if not kubectl_ok():
|
|
die(f"Cannot reach the {TRACK_ID.upper()} cluster ({CONTEXT}), so every check would fail and a 0% attempt would be recorded.\n"
|
|
f" Start it first: {setup_hint()} or discard the session: ./exam abort --yes")
|
|
if remaining(st) <= 0:
|
|
print(yellow("Note: time expired before submission."))
|
|
entry = finish_session(st)
|
|
print_score(st)
|
|
def path_note():
|
|
if not OBJECTIVES:
|
|
return
|
|
print()
|
|
for line in st.get("milestones", []):
|
|
print(green(line))
|
|
step = next_step()
|
|
print(bold(f"Next on the path: {step['cmd']}") + dim(f" {step['why']}"))
|
|
if entry is None:
|
|
print(dim("Study session - not added to history or weak-spots.")); path_note(); return
|
|
print(dim(f"Time used: {fmt(st['ended_at'] - st['started_at'] - paused_total(st))}" + (" (over time)" if st.get("over_time") else "")))
|
|
failed = sorted(int(k) for k, v in entry["results"].items() if not v["pass"])
|
|
print(dim(f"Attempt #{len(load_history())} logged to {os.path.basename(HISTORY_FILE)}; {os.path.basename(WEAK_FILE)} updated."))
|
|
if failed:
|
|
print(bold("\nTo drill what you missed: ./exam drill weak") + dim(f" (Q{', Q'.join(map(str, failed))})"))
|
|
path_note()
|
|
|
|
def cmd_abort(args):
|
|
"""End a session without grading it and restore the kubectl context."""
|
|
st = load_state()
|
|
if not st or st.get("ended_at"):
|
|
die("No session in progress.")
|
|
assert st is not None
|
|
if "--yes" not in args:
|
|
die(f"This discards the {st.get('mode', 'exam')} in progress without grading or recording it.\n"
|
|
" Confirm: ./exam abort --yes (to grade and keep it instead: ./exam end)")
|
|
st["ended_at"] = time.time(); st["aborted"] = True
|
|
save_state(st)
|
|
setup_contexts(False, {})
|
|
print(yellow("Session discarded. Nothing was graded or written to history."))
|
|
print(dim(f" kubectl context restored to {CONTEXT}. Seeded objects are still on the cluster;"))
|
|
print(dim(f" './exam drill ...' or './exam start' re-seeds, './teardown.sh {TRACK_ID}' removes the cluster."))
|
|
|
|
def cmd_weak(args):
|
|
agg = stats(); hist = load_history()
|
|
if not hist:
|
|
die("No history yet - finish a session with './exam end'.")
|
|
ids = [i for i in weak_ids(agg) if agg[i]["attempts"] > 0]
|
|
print(bold(f"Weak spots after {len(hist)} attempt(s):"))
|
|
if not ids:
|
|
print(green(" none - everything attempted has passed on its latest try"))
|
|
for i in ids:
|
|
q = q_by_id(i); a = agg[i]
|
|
print(f"\n{red('Q%02d' % i)} {q['domain']:<7} {q['title']} " + dim(f"{a['fails']}/{a['attempts']} failed, last {a['last_earned']} pts, last failed {a['last_fail_at']}"))
|
|
for n in weak_lines(a):
|
|
print(dim(" " + n))
|
|
never = [q["id"] for q in QUESTIONS if agg[q["id"]]["attempts"] == 0]
|
|
if never:
|
|
print(dim(f"\nNever attempted: Q{', Q'.join(map(str, never))}"))
|
|
print(dim(f"\nFull log: {WEAK_FILE} Drill them: ./exam drill weak"))
|
|
|
|
def cmd_history(args):
|
|
hist = load_history(include_synthetic=True)
|
|
if not hist:
|
|
die("No history yet.")
|
|
print(bold(f"{'#':>3} {'when':<17}{'mode':<7}{'score':>7} {'time':<9} tasks"))
|
|
for i, e in enumerate(hist, 1):
|
|
col = green if e["score_pct"] >= PASS_PCT else red
|
|
tasks = f"all {len(QUESTIONS)}" if len(e["ids"]) == len(QUESTIONS) else "Q" + ",".join(map(str, e["ids"]))
|
|
tag = dim(" [scripted verification - not counted]") if e.get("synthetic") else ""
|
|
print(f"{i:>3} {e['at']:<17}{(e['mode'] + ('*' if e.get('hard') else '')):<7}{col('%5.1f%%' % e['score_pct']):>16} {fmt(e['time_used']):<9} {tasks}{tag}")
|
|
exams = [e for e in hist if e["mode"] == "exam" and not e.get("synthetic")]
|
|
print(dim("* = hard"))
|
|
if len(exams) >= 2:
|
|
print(bold("\nDomain trend (full exams, % of that exam's points in the domain):"))
|
|
for d in DOMAINS:
|
|
cells = []
|
|
for e in exams:
|
|
de, dt = exam_domain_scores(e).get(d, (0, 0))
|
|
cells.append(f"{round(100 * de / dt):>3}%" if dt else " -")
|
|
print(f" {d:<7} " + " → ".join(cells))
|
|
|
|
def cmd_reset(args):
|
|
if not args:
|
|
die("usage: ./exam reset N (re-seed one task's scenario)")
|
|
q = q_by_id(int(args[0])); rc, out = run(str(q.get("setup", "")), params=session_params(load_state(), q))
|
|
print(("re-seeded " if rc == 0 else "problem re-seeding ") + f"Q{q['id']:02d}: {q['title']}")
|
|
if out:
|
|
print(dim(out))
|
|
|
|
# --------------------------------------------------------------- web ----
|
|
class Api(http.server.BaseHTTPRequestHandler):
|
|
"""Serves web/index.html plus a small JSON API. State lives in the same files the CLI uses."""
|
|
lock = threading.Lock()
|
|
log_lines = []
|
|
seeding = False
|
|
seed_error = None
|
|
ttyd_port = 7681
|
|
shell = "bash"
|
|
|
|
def log_message(self, format: str, *args: Any) -> None: # quiet
|
|
pass
|
|
|
|
def _json(self, obj, code=200):
|
|
body = json.dumps(obj).encode()
|
|
self.send_response(code); self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body)
|
|
|
|
def _read_json(self):
|
|
n = int(self.headers.get("Content-Length") or 0)
|
|
return json.loads(self.rfile.read(n) or b"{}")
|
|
|
|
def do_GET(self):
|
|
p = urllib.parse.urlparse(self.path)
|
|
if p.path in ("/", "/index.html"):
|
|
with open(os.path.join(ROOT, "web", "index.html"), "rb") as f:
|
|
body = f.read()
|
|
self.send_response(200); self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body); return
|
|
if p.path == "/api/session":
|
|
with Api.lock:
|
|
return self._json(self.session_payload())
|
|
if p.path == "/api/history":
|
|
return self._json({"history": load_history(), "weak": weak_ids(), "stats": {str(k): v.to_dict() for k, v in stats().items()}})
|
|
self._json({"error": "not found"}, 404)
|
|
|
|
def do_POST(self):
|
|
p = urllib.parse.urlparse(self.path); body = self._read_json()
|
|
with Api.lock:
|
|
st = load_state()
|
|
if p.path == "/api/track":
|
|
t = body.get("track")
|
|
if t not in TRACKS:
|
|
return self._json({"error": f"unknown track {t}"}, 400)
|
|
if Api.seeding:
|
|
return self._json({"error": "a session is being seeded"}, 409)
|
|
load_track(t)
|
|
Api.log_lines = []
|
|
return self._json(self.session_payload())
|
|
if p.path == "/api/start":
|
|
if Api.seeding:
|
|
return self._json({"error": "already seeding"}, 409)
|
|
mode = body.get("mode", "exam"); hard = bool(body.get("hard")); rng = random.Random()
|
|
Api.log_lines = []; Api.seed_error = None; Api.seeding = True
|
|
ansi = __import__("re").compile(r"\x1b\[[0-9;]*m")
|
|
ulog = lambda line: Api.log_lines.append(ansi.sub("", line))
|
|
def work():
|
|
try:
|
|
if body.get("sim") in SETS:
|
|
begin_session(set_questions(body["sim"]), "exam", EXAM_SECONDS, True, log=ulog, hard=True, other_ok=False)
|
|
elif mode == "exam":
|
|
begin_session(compose_exam(hard, rng), "exam", EXAM_SECONDS, True, log=ulog, hard=hard, other_ok=False)
|
|
else:
|
|
objective = find_objective(body["objective"]) if body.get("objective") else None
|
|
if objective:
|
|
qs, _ = objective_session(objective, "study" if body.get("study") else "drill", bool(body.get("all")))
|
|
elif body.get("domain") in DOMAINS:
|
|
qs = pick(body["domain"], DRILL_SIZE, hard, rng)
|
|
elif body.get("campaign"):
|
|
_, qs = campaign_batch(hard, rng)
|
|
if not qs:
|
|
qs = no_conflicts(split_solo([q for q in QUESTIONS if q["id"] in weak_ids()])[0], DRILL_SIZE)
|
|
elif body.get("next"):
|
|
_, qs = next_drill(hard, rng)
|
|
else:
|
|
ids = body.get("ids") or weak_ids()
|
|
qs = no_conflicts(split_solo([q for q in QUESTIONS if q["id"] in ids])[0], len(ids))
|
|
per = DRILL_SECONDS_PER_TASK_HARD if hard else DRILL_SECONDS_PER_TASK
|
|
begin_session(qs, "drill", per * len(qs), True, log=ulog, hard=hard,
|
|
untimed_session=bool(body.get("study")), other_ok=False,
|
|
objective=objective and objective["id"])
|
|
except SystemExit as e:
|
|
Api.seed_error = (getattr(e, "msg", None) or "\n".join(l for l in Api.log_lines if l.strip())
|
|
or "start failed - see the terminal running ./exam web")
|
|
except Exception as e: # noqa: BLE001
|
|
Api.seed_error = f"{type(e).__name__}: {e}"
|
|
finally:
|
|
Api.seeding = False
|
|
threading.Thread(target=work, daemon=True).start()
|
|
return self._json({"seeding": True})
|
|
if not st:
|
|
return self._json({"error": "no session"}, 400)
|
|
if p.path == "/api/flag":
|
|
qid = int(body["id"]); flags = set(st.get("flags", []))
|
|
flags.symmetric_difference_update({qid}); st["flags"] = sorted(flags); save_state(st)
|
|
elif p.path == "/api/pause":
|
|
set_paused(st, bool(body.get("paused", True)))
|
|
elif p.path == "/api/notes":
|
|
st["notes"] = body.get("notes", ""); save_state(st)
|
|
elif p.path == "/api/grade":
|
|
if st.get("mode") != "drill" and not st.get("ended_at"):
|
|
return self._json({"error": "No feedback during an exam. Submit to see results."}, 403)
|
|
q = q_by_id(int(body["id"]))
|
|
if "results" not in st or not isinstance(st["results"], dict):
|
|
st["results"] = {}
|
|
st["results"][str(q["id"])] = grade_one(q, session_params(st, q)); save_state(st)
|
|
record_study(st, {q["id"]})
|
|
if untimed(st):
|
|
st["milestones"] = st.get("milestones", []) + milestone_lines(update_milestones()); save_state(st)
|
|
elif p.path == "/api/end":
|
|
if not st.get("ended_at"):
|
|
if not kubectl_ok(): # grading an unreachable cluster would record a 0% attempt
|
|
return self._json({"error": f"Cannot reach the {TRACK_ID.upper()} cluster ({CONTEXT}). Start it with {setup_hint()}, then submit."}, 503)
|
|
finish_session(st, log=lambda *_: None)
|
|
elif p.path == "/api/reset":
|
|
q = q_by_id(int(body["id"])); run(str(q.get("setup", "")), params=session_params(st, q))
|
|
else:
|
|
return self._json({"error": "not found"}, 404)
|
|
return self._json(self.session_payload())
|
|
|
|
def session_payload(self) -> dict[str, Any]:
|
|
st = load_state()
|
|
qs = session_questions(st) if st else QUESTIONS
|
|
payload: dict[str, Any] = {
|
|
"track": TRACK, "track_id": TRACK_ID, "workspace": WS_DISPLAY, "version": VERSION, "ttyd_port": Api.ttyd_port,
|
|
"ttyd": shutil.which("ttyd") is not None, "shell": Api.shell, "setup_log": Api.log_lines,
|
|
"questions": [{"id": q["id"], "domain": q["domain"], "title": q["title"], "weight": q["weight"], "difficulty": q["difficulty"],
|
|
"task": task_text(q, session_params(st, q)), "checks": [{"name": c["name"], "pts": c["pts"]} for c in q["checks"]],
|
|
"solution": render(str(q.get("solution", "")), session_params(st, q)) if solution_allowed(st, int(q["id"])) else None,
|
|
# which objective a task belongs to is a hint, so it follows the solution's rule
|
|
"objectives": [o["id"] for o in objectives_of(q["id"])] if solution_allowed(st, int(q["id"])) else []}
|
|
for q in qs],
|
|
"bank": [{"id": q["id"], "domain": q["domain"], "title": q["title"], "difficulty": q["difficulty"]} for q in QUESTIONS],
|
|
"tracks": TRACKS, "track_k8s": track_k8s(),
|
|
"cluster": CONTEXT, "drilled": drilled_domains(), "sets": {k: {"name": v, "count": len(set_questions(k))} for k, v in SETS.items()},
|
|
"seen": sorted(last_seen()),
|
|
"seeding": Api.seeding, "seed_error": Api.seed_error,
|
|
"session": None,
|
|
}
|
|
payload.update(self.path_payload())
|
|
if st:
|
|
earned, total, pct = score(st)
|
|
payload["session"] = {"mode": st.get("mode"), "hard": st.get("hard", False), "started_at": st["started_at"], "duration": st.get("duration", EXAM_SECONDS),
|
|
"ends_at": st["started_at"] + st.get("duration", EXAM_SECONDS), "ended_at": st.get("ended_at"),
|
|
"over_time": st.get("over_time", False), "flags": st.get("flags", []),
|
|
"paused": bool(st.get("paused_at")), "paused_total": paused_total(st),
|
|
"untimed": untimed(st), "notes": st.get("notes", ""),
|
|
"objective": st.get("objective"), "milestones": st.get("milestones", []),
|
|
"results": st["results"], "score": {"earned": earned, "total": total, "pct": pct, "pass_pct": PASS_PCT},
|
|
"domains": self.domain_scores(st), "server_time": time.time()}
|
|
return payload
|
|
|
|
def path_payload(self) -> dict[str, Any]:
|
|
"""The curriculum as the UI shows it: every objective with its notes, tasks and state, how
|
|
many are studied and done, and the next step on the path. Empty for a track without one."""
|
|
if not OBJECTIVES:
|
|
return {"objectives": [], "path": None}
|
|
agg, studied = stats(), load_study()
|
|
objs = [{"id": o["id"], "key": o["key"], "domain": o["domain"], "text": o["text"], "state": objective_state(o, agg, studied),
|
|
"know": o["know"], "fast": o["fast"].strip("\n"), "traps": o["traps"], "docs": o["docs"],
|
|
"tasks": [{"id": q["id"], "title": q["title"], "set": q.get("set"), "state": task_state(q["id"], agg, studied)}
|
|
for q in objective_tasks(o, True)]} for o in OBJECTIVES]
|
|
step = next_step(agg, studied); full, fixed = mock_scores()
|
|
states = [o["state"] for o in objs]
|
|
return {"objectives": objs,
|
|
"path": {"total": len(objs), "done": states.count("done"), "studied": states.count("done") + states.count("studied"),
|
|
"mock_best": max(full) if full else None, "set_best": {k: (max(v) if v else None) for k, v in fixed.items()},
|
|
"next": {"kind": step["kind"], "cmd": step["cmd"], "why": step["why"], "set": step.get("set"),
|
|
"objective": step["objective"]["id"] if step.get("objective") else None}}}
|
|
|
|
def domain_scores(self, st: dict[str, Any]) -> dict[str, Any]:
|
|
qs = session_questions(st); out = {}
|
|
st_results: dict[str, Any] = st.get("results", {})
|
|
for d in DOMAINS:
|
|
dq = [q for q in qs if q["domain"] == d]
|
|
if dq:
|
|
out[d] = {"earned": sum(st_results.get(str(q["id"]), {}).get("earned", 0) for q in dq),
|
|
"total": sum(q["weight"] for q in dq)}
|
|
return out
|
|
|
|
def cmd_web(args):
|
|
port = 8080
|
|
if "--port" in args:
|
|
port = int(args[args.index("--port") + 1])
|
|
Api.shell = os.environ.get("CKA_SHELL", "bash")
|
|
if "--shell" in args:
|
|
Api.shell = args[args.index("--shell") + 1]
|
|
os.makedirs(WORKSPACE, exist_ok=True)
|
|
import socket
|
|
def port_free(pt):
|
|
with socket.socket() as sk:
|
|
sk.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
try:
|
|
sk.bind(("127.0.0.1", pt)); return True
|
|
except OSError:
|
|
return False
|
|
if not port_free(port):
|
|
die(f"Port {port} is already in use - probably another './exam web' still running.\n"
|
|
f" Find it: ss -ltnp | grep :{port} or start on another port: ./exam web --port {port + 10}")
|
|
if shutil.which("ttyd") and not port_free(Api.ttyd_port):
|
|
print(yellow(f" ttyd port {Api.ttyd_port} already in use - reusing the existing terminal server."))
|
|
ttyd = None
|
|
if shutil.which("ttyd") and port_free(Api.ttyd_port):
|
|
env = dict(os.environ, WORKSPACE=WORKSPACE, KUBECONFIG=os.environ.get("KUBECONFIG", os.path.expanduser("~/.kube/config")))
|
|
rc = os.path.join(ROOT, "examrc.fish" if "fish" in Api.shell else "examrc")
|
|
if "fish" in Api.shell:
|
|
shell_cmd = [Api.shell, "-C", f"source {rc}"]
|
|
else:
|
|
shell_cmd = [Api.shell, "--rcfile", rc, "-i"]
|
|
# -W makes the terminal writable, so it must never be reachable off this machine: it is an
|
|
# unauthenticated root-equivalent shell with the cluster kubeconfig. ttyd binds every
|
|
# interface by default, so pin it to loopback (macOS needs lo0; override if yours differs).
|
|
iface = os.environ.get("CKA_TTYD_INTERFACE", "lo")
|
|
ttyd = subprocess.Popen(["ttyd", "-p", str(Api.ttyd_port), "-i", iface, "-W",
|
|
"-t", "fontSize=14", "-t", "theme={\"background\":\"#1a1b1e\"}",
|
|
"--cwd", ROOT] + shell_cmd, env=env,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
print(dim(f" terminal: ttyd on {iface}:{Api.ttyd_port} running {Api.shell} in {ROOT} with k=kubectl, $do and $now preset"))
|
|
else:
|
|
print(yellow(" ttyd not found - the terminal pane will show install instructions.\n"
|
|
" (dnf/apt/pacman/brew install ttyd, or https://github.com/tsl0922/ttyd/releases)"))
|
|
http.server.ThreadingHTTPServer.allow_reuse_address = True
|
|
srv = http.server.ThreadingHTTPServer(("127.0.0.1", port), Api) # type: ignore[arg-type]
|
|
print(bold(f"\n {TRACK_ID.upper()} exam UI: http://127.0.0.1:{port} (exam {VERSION})\n") + dim(" Ctrl-C stops the UI (not the exam clock)."))
|
|
def stop(*_):
|
|
if ttyd:
|
|
ttyd.terminate()
|
|
srv.shutdown()
|
|
signal.signal(signal.SIGINT, lambda *_: threading.Thread(target=stop).start())
|
|
signal.signal(signal.SIGTERM, lambda *_: threading.Thread(target=stop).start())
|
|
try:
|
|
srv.serve_forever()
|
|
finally:
|
|
if ttyd:
|
|
ttyd.terminate()
|
|
|
|
def cmd_version(args):
|
|
print(f"exam {VERSION} track={TRACK_ID} k8s={K8S} cluster={CONTEXT} bank={len(QUESTIONS)} tasks python={sys.version.split()[0]}")
|
|
|
|
def cmd_cluster_env(args):
|
|
"""Shell assignments for this track's kind cluster - setup.sh, teardown.sh and the tool scripts
|
|
eval this, so the track -> Kubernetes version -> cluster/image mapping lives only here."""
|
|
print(f"TRACK={TRACK_ID}\nK8S={K8S}\nCLUSTER={CLUSTER}\nNODE_IMAGE={KIND_IMAGES.get(K8S, '')}")
|
|
|
|
def cmd_clusters(args):
|
|
"""Which cluster each track runs on, and whether it is up."""
|
|
print(bold(f"{'track':<7}{'exam runs':<11}{'kind cluster':<15}{'context':<19}status"))
|
|
for t in TRACKS:
|
|
load_track(t)
|
|
ver = server_version()
|
|
status = (green(f"up, {ver}") if ver and ver.startswith(K8S + ".") else yellow(f"up, but {ver}") if ver
|
|
else red(f"not reachable ({setup_hint()})"))
|
|
print(f"{t:<7}{K8S:<11}{CLUSTER:<15}{CONTEXT:<19}{status}")
|
|
|
|
def cmd_help(args):
|
|
print(__doc__ + f"\nversion {VERSION}")
|
|
objective_help = "" if not OBJECTIVES else (
|
|
f" objectives the curriculum as a path to the exam: {len(OBJECTIVES)} objectives, their tasks, your progress\n"
|
|
" next start the next step on that path (study an objective, then drill it, then the mocks)\n"
|
|
" notes [OBJ|DOMAIN|all] what to know, fastest commands, traps and docs for an objective (3.2 or probes)\n"
|
|
" study [OBJ|next] an objective's notes and tasks; with no argument, the next one on the path\n"
|
|
" (--all adds its Simulator Set tasks); drill OBJ = the same tasks against the clock\n")
|
|
print(f"""usage: ./exam [--track cka|ckad|cks] <command> (current track: {TRACK_ID}, {len(QUESTIONS)} tasks)
|
|
Each track runs on a kind cluster at its exam's Kubernetes version - {TRACK_ID}: {K8S} on {CONTEXT}.
|
|
|
|
web [--port 8080] [--shell bash] exam-style browser UI: tasks, terminal, clock, submit
|
|
study [DOMAIN|weak|N..] no clock, solutions unlocked, kept out of the history (--count N, default {DRILL_SIZE})
|
|
{objective_help} drill next domain you have drilled least: {DRILL_SIZE} tasks, 6 min each (--hard: 5 min, hard tasks)
|
|
drill all [--hard] campaign: {DRILL_SIZE} never-attempted tasks per session until the bank is done, then weak ones
|
|
drill weak|N..|DOMAIN.. chosen tasks (--time MIN, --force, --hard)
|
|
progress attempted/passed per domain, normal and hard
|
|
solution N reference solution (drills: after grading N; exams: after submit)
|
|
sim {'|'.join(SETS) or '<set>'}{' ' * max(1, 20 - len('|'.join(SETS)))}fixed simulator set, exam rules, {TRACK['minutes']} min
|
|
start [--hard] full exam: {sum(QUOTA.values()) or '?'} tasks sampled by domain ({sum(QUOTA_HARD.values()) or '?'} in hard mode), {TRACK['minutes']} min.
|
|
Requires every domain drilled once first (--force skips). Hard mode adds traps,
|
|
exact-output tasks, compound failures{' and collateral checks' if not TRACK.get('exam_contexts') else ', collateral checks and exam-style contexts'}.
|
|
list tasks in this session, with current + historical results
|
|
show N display task N
|
|
grade N|all grade one task (or all) per check - drills only (--force to peek in an exam)
|
|
status / timer time remaining (timer = live countdown)
|
|
pause / resume stop the clock (not recorded anywhere - but the real exam has no pause)
|
|
end submit: grade per check, score, log attempt, update weak-spots
|
|
weak what you keep getting wrong, with the failing checks
|
|
history every attempt, scores, and per-domain trend
|
|
reset N re-seed the scenario for task N
|
|
clusters each track's Kubernetes version and cluster, and whether it is up
|
|
""")
|
|
|
|
COMMANDS = {"start": cmd_start, "drill": cmd_drill, "list": cmd_list, "show": cmd_show, "grade": cmd_grade,
|
|
"status": cmd_status, "timer": cmd_timer, "pause": cmd_pause, "resume": cmd_resume, "end": cmd_end, "weak": cmd_weak, "history": cmd_history,
|
|
"reset": cmd_reset, "web": cmd_web, "abort": cmd_abort, "progress": cmd_progress, "solution": cmd_solution, "sim": cmd_sim, "study": cmd_study, "version": cmd_version, "help": cmd_help,
|
|
"clusters": cmd_clusters, "cluster-env": cmd_cluster_env,
|
|
"objectives": cmd_objectives, "curriculum": cmd_objectives, "notes": cmd_notes, "next": cmd_next}
|
|
|
|
if __name__ == "__main__":
|
|
cmd = ARGV[0] if ARGV else "help"
|
|
COMMANDS.get(cmd, cmd_help)(ARGV[1:])
|