"""Submit authorized Factory workflows as portable Kubernetes Jobs.""" from __future__ import annotations import argparse import base64 import json import os import re import secrets import sqlite3 import stat import subprocess import tempfile import threading import time from pathlib import Path from typing import Any, BinaryIO, Sequence from adw_modules import repository_registry, result_mailbox, tracer APPROVED_IMAGE = "ghcr.io/alexcodeplace/overdeck-agent-sandbox@sha256:1db62fb686815054b86a658bed943d81faa71fc23908a325a952ae50df36c9db" APPROVED_RESULT_HOST = "100.126.128.50" APPROVED_PROXY_URL = "http://10.43.200.8:8888" RESULT_PROCESSING_TIMEOUT_SECONDS = 120 MAX_RESULT_OBJECTS = 100_000 MAX_RESULT_OBJECT_BYTES = 256 * 1024 * 1024 MAX_RESULT_EXPANDED_BYTES = 1024 * 1024 * 1024 MAX_RESULT_PATCH_BYTES = 256 * 1024 * 1024 MAX_OBJECT_LIST_BYTES = 16 * 1024 * 1024 MAX_JOB_LOG_BYTES = 64 * 1024 * 1024 MAX_JOB_TRACE_RECORDS = 100_000 MAX_RECOVERY_TRACE_ERROR_BYTES = 32 * 1024 PROCESS_ADDRESS_SPACE_BYTES = 2 * 1024 * 1024 * 1024 LOG_TRUNCATION_MARKER = b"\n[Factory controller: log truncated at configured limit]\n" DNS = re.compile(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$") RESULT_BUNDLE_PATH = "/result/result.bundle" RESULT_MARKER_PATH = "/result/complete" def run(argv: Sequence[str], *, cwd: Path | None = None, env=None, input_data: str | None = None, check=True, timeout: float | None = None): result = subprocess.run(argv, cwd=cwd, env=env, input=input_data, text=True, capture_output=True, check=False, timeout=timeout) if check and result.returncode: raise RuntimeError(f"{argv[0]} exited {result.returncode}: {(result.stderr or result.stdout).strip()[-1200:]}") return result def run_bounded_output( argv: Sequence[str], path: Path, *, limit: int, timeout: float, ) -> tuple[bool, str | None]: if limit <= 0: raise ValueError("bounded output limit must be positive") errors = tempfile.TemporaryFile() try: with path.open("wb") as output: completed = subprocess.run( ["/usr/bin/prlimit", f"--fsize={limit + 1}", "--", *argv], stdout=output, stderr=errors, check=False, timeout=timeout, ) truncated = path.stat().st_size > limit if truncated: marker = LOG_TRUNCATION_MARKER[:limit] with path.open("r+b") as output: output.truncate(limit - len(marker)) output.seek(0, os.SEEK_END) output.write(marker) return True, None if completed.returncode: errors.seek(0, os.SEEK_END) size = errors.tell() errors.seek(max(0, size - 1200)) error = errors.read().decode(errors="replace").strip() return False, error or f"{argv[0]} exited {completed.returncode}" return False, None finally: errors.close() class JobLogFollower: def __init__( self, argv: Sequence[str], path: Path, mirror: tracer.KubernetesTraceMirror, *, limit: int = MAX_JOB_LOG_BYTES, record_limit: int = MAX_JOB_TRACE_RECORDS, popen=subprocess.Popen, ): if limit <= 0 or record_limit <= 0: raise ValueError("bounded output limits must be positive") self.path = path self.mirror = mirror self.limit = limit self.record_limit = record_limit self.records_consumed = 0 self.errors: list[str] = [] self.bytes_written = 0 self.truncated = False self._stderr: BinaryIO = tempfile.TemporaryFile() self._output = path.open("wb") try: self.process = popen( list(argv), stdout=subprocess.PIPE, stderr=self._stderr, bufsize=0, ) except BaseException: self._output.close() self._stderr.close() raise if self.process.stdout is None: self._output.close() self._stderr.close() raise RuntimeError("Kubernetes log follower has no output stream") self.thread = threading.Thread(target=self._drain, name="factory-k3s-logs", daemon=True) self.thread.start() def _write(self, data: bytes) -> None: remaining = self.limit + 1 - self.bytes_written if remaining > 0: chunk = data[:remaining] self._output.write(chunk) self.bytes_written += len(chunk) if self.bytes_written > self.limit or len(data) > remaining: self.truncated = True def _drain(self) -> None: assert self.process.stdout is not None continuation = False try: while True: piece = self.process.stdout.readline(tracer.MAX_KUBERNETES_TRACE_BYTES + 1) if not piece: break self._write(piece) if self.truncated and not self.errors: self.errors.append("Kubernetes Job log exceeded its configured limit") if self.errors: continue complete = piece.endswith(b"\n") if continuation: continuation = not complete continue if not complete and len(piece) > tracer.MAX_KUBERNETES_TRACE_BYTES: if piece.startswith(tracer.KUBERNETES_TRACE_PREFIX.encode()): self.errors.append("Kubernetes trace record exceeds its size limit") continuation = True continue try: line = piece.rstrip(b"\r\n") if line.startswith(tracer.KUBERNETES_TRACE_PREFIX.encode()): if self.records_consumed >= self.record_limit: self.errors.append("Kubernetes trace record budget exceeded") continue consumed = self.mirror.consume(line) if consumed: self.records_consumed += 1 except (OSError, RuntimeError, ValueError, sqlite3.Error) as exc: self.errors.append(str(exc)) except BaseException as exc: self.errors.append(f"Kubernetes log follower read failed: {type(exc).__name__}") finally: self.process.stdout.close() def finish(self, timeout: float = 10) -> tuple[bool, str | None]: forced = False try: self.process.wait(timeout=timeout) except subprocess.TimeoutExpired: forced = True self.process.terminate() try: self.process.wait(timeout=2) except subprocess.TimeoutExpired: self.process.kill() self.process.wait(timeout=2) self.thread.join(timeout=timeout) if self.thread.is_alive(): self.errors.append("Kubernetes log follower did not stop") if forced: self.errors.append("Kubernetes log follower did not exit after Pod termination") if self.process.returncode: self._stderr.seek(0, os.SEEK_END) size = self._stderr.tell() self._stderr.seek(max(0, size - 1200)) detail = self._stderr.read().decode(errors="replace").strip() self.errors.append(detail or f"kubectl logs exited {self.process.returncode}") self._output.flush() self._output.close() self._stderr.close() if self.truncated: marker = LOG_TRUNCATION_MARKER[:self.limit] with self.path.open("r+b") as output: output.truncate(self.limit - len(marker)) output.seek(0, os.SEEK_END) output.write(marker) return self.truncated, self.errors[0] if self.errors else None def abort(self) -> None: if self.process.poll() is None: self.process.terminate() try: self.process.wait(timeout=2) except subprocess.TimeoutExpired: self.process.kill() self.process.wait(timeout=2) self.thread.join(timeout=2) if not self._output.closed: self._output.close() if not self._stderr.closed: self._stderr.close() def canonical_trace_paths() -> tuple[Path, Path]: state = Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local/state")) db = Path(os.environ.get("FACTORY_DB_PATH", state / "overdeck/factory/sssf.db")) return db, db.parent / "kubernetes-events.jsonl" def git(repo: Path, *args: str, env=None) -> str: return run(["git", *args], cwd=repo, env=env).stdout.strip() def workspace_snapshot(repo: Path) -> tuple[str, str, str]: base = git(repo, "rev-parse", "--verify", "HEAD^{commit}") with tempfile.NamedTemporaryFile(prefix="factory-k8s-index-") as index: env = os.environ.copy() env["GIT_INDEX_FILE"] = index.name index.truncate(0) git(repo, "read-tree", base, env=env) git(repo, "add", "-A", env=env) tree = git(repo, "write-tree", env=env) commit_env = os.environ.copy() commit_env.update({"GIT_AUTHOR_NAME": "Overdeck Factory", "GIT_AUTHOR_EMAIL": "factory@overdeck.invalid", "GIT_COMMITTER_NAME": "Overdeck Factory", "GIT_COMMITTER_EMAIL": "factory@overdeck.invalid"}) commit = git(repo, "commit-tree", tree, "-p", base, env=commit_env) return base, tree, commit def origin(repo: Path) -> str: return git(repo, "remote", "get-url", "origin") def label(value: str, name: str) -> str: if len(value) > 63 or not DNS.fullmatch(value): raise ValueError(f"{name} must be a Kubernetes DNS label") return value def signing_key(value: bytes | None = None) -> bytes: if value is not None: return value configured = os.getenv("FACTORY_ATTEMPT_SIGNING_PRIVATE_KEY") path = Path(configured) if configured else Path(os.getenv("XDG_STATE_HOME", "~/.local/state")).expanduser() / "overdeck/factory/attempt-signing-private-key.pem" try: descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) with os.fdopen(descriptor, "rb") as source: status = os.fstat(source.fileno()) key = source.read(65537) except OSError as exc: raise ValueError("Factory attempt signing private key is unreadable") from exc if ( not stat.S_ISREG(status.st_mode) or status.st_uid != os.getuid() or status.st_mode & 0o077 or len(key) > 65536 or not key.startswith(b"-----BEGIN PRIVATE KEY-----") ): raise ValueError("FACTORY_ATTEMPT_SIGNING_PRIVATE_KEY is not a private owner-only key") return key def manifest(*, job_name: str, namespace: str, image: str, secret: str, upload_secret: str, service_account: str, envelope: repository_registry.AttemptEnvelope, execution_ref: str, result_ref: str, invocation: Sequence[str], upload: result_mailbox.UploadTarget, timeout: int, pull_secret: str = "overdeck-ghcr-pull", runtime_secret: str = "overdeck-factory-runtime") -> dict[str, Any]: for value, name in ((job_name, "job name"), (namespace, "namespace"), (secret, "secret"), (upload_secret, "upload secret"), (service_account, "service account"), (pull_secret, "pull secret"), (runtime_secret, "runtime secret")): label(value, name) if image != APPROVED_IMAGE: raise ValueError("FACTORY_K8S_IMAGE must use the approved immutable GHCR digest") if envelope.workflow != repository_registry.WORKFLOW: raise ValueError("Factory Kubernetes transport requires an authorized workflow") if secret != envelope.credential_handle: raise ValueError("Factory Kubernetes transport requires the repository-bound credential Secret") if not re.fullmatch(r"refs/heads/factory-exec/[0-9a-f]{24}", execution_ref): raise ValueError("execution ref is invalid") if not re.fullmatch(r"refs/heads/factory-result/[0-9a-f]{24}", result_ref): raise ValueError("result ref is invalid") attempt_id = execution_ref.removeprefix("refs/heads/factory-exec/") if job_name != f"factory-{attempt_id[:16]}" or upload_secret != f"factory-upload-{attempt_id[:16]}": raise ValueError("Factory Job resources do not match the execution identity") parsed_upload = result_mailbox.validate_target(upload, attempt_id) if parsed_upload.hostname != APPROVED_RESULT_HOST: raise ValueError("Factory result mailbox host is not approved") upload_json = result_mailbox.serialize_target(upload) if len(upload_json.encode()) > 131072: raise ValueError("Factory result upload target exceeds the Kubernetes control-plane limit") if not invocation or any(not isinstance(value, str) for value in invocation): raise ValueError("invocation must contain strings") if ( envelope.attempt_id != execution_ref.removeprefix("refs/heads/factory-exec/") or envelope.execution_ref != execution_ref or envelope.result_ref != result_ref or envelope.invocation_sha256 != repository_registry.invocation_sha256(invocation) or envelope.upload_sha256 != result_mailbox.target_sha256(upload) ): raise ValueError("Factory attempt envelope does not match the Job transport") invocation_json = json.dumps(list(invocation), separators=(",", ":")) if len(invocation_json.encode()) > 131072: raise ValueError("Factory invocation exceeds the Kubernetes control-plane limit") if not 60 <= timeout <= 86400: raise ValueError("timeout must be between 60 and 86400 seconds") envelope_json = envelope.serialize() if len(envelope_json.encode()) > 131072: raise ValueError("Factory attempt envelope exceeds the Kubernetes control-plane limit") labels = {"app.kubernetes.io/name": "overdeck-factory", "app.kubernetes.io/component": "worker", "overdeck.dev/factory-job": job_name} locked = {"allowPrivilegeEscalation": False, "readOnlyRootFilesystem": True, "capabilities": {"drop": ["ALL"]}} mounts = [ {"name": "workspace", "mountPath": "/workspace"}, {"name": "home", "mountPath": "/home/factory"}, {"name": "tmp", "mountPath": "/tmp"}, {"name": "result", "mountPath": "/result"}, {"name": "repository-key", "mountPath": "/home/factory/.ssh/id_ed25519", "subPath": "id_ed25519", "readOnly": True}, {"name": "repository-key", "mountPath": "/home/factory/.ssh/known_hosts", "subPath": "known_hosts", "readOnly": True}, {"name": "runtime-credentials", "mountPath": "/home/factory/.pi/agent/auth.json", "subPath": "pi-auth.json", "readOnly": True}, {"name": "runtime-credentials", "mountPath": "/home/factory/.pi/agent/models.json", "subPath": "pi-models.json", "readOnly": True}, {"name": "runtime-credentials", "mountPath": "/home/factory/attempt-signing-public-key.pem", "subPath": "attempt-signing-public-key.pem", "readOnly": True}, ] pod = { "serviceAccountName": service_account, "automountServiceAccountToken": False, "restartPolicy": "Never", "imagePullSecrets": [{"name": pull_secret}], "hostAliases": [{"ip": "10.43.200.8", "hostnames": ["overdeck-factory-egress"]}], "securityContext": {"runAsNonRoot": True, "runAsUser": 1000, "runAsGroup": 1000, "fsGroup": 1000, "seccompProfile": {"type": "RuntimeDefault"}}, "initContainers": [{"name": "prepare-home", "image": image, "imagePullPolicy": "Always", "command": ["/bin/mkdir"], "args": ["-p", "/home/factory/.pi/agent", "/home/factory/.ssh"], "securityContext": locked, "volumeMounts": [{"name": "home", "mountPath": "/home/factory"}]}], "containers": [{ "name": "factory", "image": image, "imagePullPolicy": "Always", "command": ["/usr/local/bin/factory-kubernetes-worker"], "env": [ {"name": "FACTORY_REPOSITORY_ID", "value": envelope.repository_id}, {"name": "FACTORY_CREDENTIAL_HANDLE", "value": envelope.credential_handle}, {"name": "FACTORY_ATTEMPT_ENVELOPE", "value": envelope_json}, {"name": "FACTORY_EXECUTION_REF", "value": execution_ref}, {"name": "FACTORY_RESULT_REF", "value": result_ref}, {"name": "FACTORY_INVOCATION_JSON", "value": invocation_json}, {"name": "FACTORY_RESULT_UPLOAD_JSON", "valueFrom": {"secretKeyRef": {"name": upload_secret, "key": "upload-json"}}}, {"name": "FACTORY_DATA_DIR", "value": "/home/factory/state/data"}, {"name": "FACTORY_DB_PATH", "value": "/home/factory/state/sssf.db"}, {"name": "PI_AUTH_PATH", "value": "/home/factory/.pi/agent/auth.json"}, {"name": "PI_MODELS_PATH", "value": "/home/factory/.pi/agent/models.json"}, {"name": "HTTPS_PROXY", "value": APPROVED_PROXY_URL}, {"name": "HTTP_PROXY", "value": APPROVED_PROXY_URL}, {"name": "NODE_USE_ENV_PROXY", "value": "1"}, {"name": "UV_NO_SYNC", "value": "1"}, ], "resources": {"requests": {"cpu": envelope.resource_policy["cpu_request"], "memory": envelope.resource_policy["memory_request"]}, "limits": {"cpu": envelope.resource_policy["cpu_limit"], "memory": envelope.resource_policy["memory_limit"]}}, "securityContext": locked, "volumeMounts": mounts, }, { "name": "result-reader", "image": image, "imagePullPolicy": "Always", "command": ["/bin/sh", "-c", "while :; do sleep 30; done"], "resources": {"requests": {"cpu": "1m", "memory": "8Mi"}, "limits": {"cpu": "10m", "memory": "16Mi"}}, "securityContext": locked, "volumeMounts": [{"name": "result", "mountPath": "/result", "readOnly": True}], }], "volumes": [ {"name": "workspace", "emptyDir": {"sizeLimit": "8Gi"}}, {"name": "home", "emptyDir": {"sizeLimit": "2Gi"}}, {"name": "tmp", "emptyDir": {"sizeLimit": "256Mi"}}, {"name": "result", "emptyDir": {"sizeLimit": "256Mi"}}, {"name": "repository-key", "secret": {"secretName": secret, "defaultMode": 256}}, {"name": "runtime-credentials", "secret": {"secretName": runtime_secret, "defaultMode": 288}}, ], } return {"apiVersion": "batch/v1", "kind": "Job", "metadata": {"name": job_name, "namespace": namespace, "labels": labels}, "spec": {"suspend": True, "backoffLimit": 0, "activeDeadlineSeconds": timeout, "ttlSecondsAfterFinished": 3600, "template": {"metadata": {"labels": labels}, "spec": pod}}} def write_receipt(path: Path, value: dict[str, Any]) -> None: temporary = path.with_suffix(".json.writing") with temporary.open("w") as output: json.dump(value, output, sort_keys=True, indent=2) output.write("\n") output.flush() os.fsync(output.fileno()) os.replace(temporary, path) descriptor = os.open(path.parent, os.O_RDONLY) try: os.fsync(descriptor) finally: os.close(descriptor) def upload_secret_manifest(namespace: str, name: str, job_name: str, job_uid: str, upload_json: str) -> dict[str, Any]: label(namespace, "namespace") label(name, "upload secret") label(job_name, "job name") if not re.fullmatch(r"[0-9a-f-]{36}", job_uid): raise ValueError("Factory Job UID is invalid") if len(upload_json.encode()) > 131072: raise ValueError("Factory result upload target exceeds the Kubernetes control-plane limit") return { "apiVersion": "v1", "kind": "Secret", "metadata": { "name": name, "namespace": namespace, "labels": { "app.kubernetes.io/name": "overdeck-factory", "overdeck.dev/factory-job": job_name, }, "ownerReferences": [{ "apiVersion": "batch/v1", "kind": "Job", "name": job_name, "uid": job_uid, "controller": True, "blockOwnerDeletion": True, }], }, "immutable": True, "type": "Opaque", "data": {"upload-json": base64.b64encode(upload_json.encode()).decode()}, } def mailbox_network_policy(namespace: str, name: str, job_name: str, job_uid: str, upload: result_mailbox.UploadTarget) -> dict[str, Any]: label(namespace, "namespace") label(name, "mailbox policy") label(job_name, "job name") if not re.fullmatch(r"[0-9a-f-]{36}", job_uid): raise ValueError("Factory Job UID is invalid") parsed = result_mailbox.validate_target(upload) if parsed.hostname != APPROVED_RESULT_HOST or parsed.port is None: raise ValueError("Factory result mailbox host is not approved") return { "apiVersion": "networking.k8s.io/v1", "kind": "NetworkPolicy", "metadata": { "name": name, "namespace": namespace, "labels": { "app.kubernetes.io/name": "overdeck-factory", "overdeck.dev/factory-job": job_name, }, "ownerReferences": [{ "apiVersion": "batch/v1", "kind": "Job", "name": job_name, "uid": job_uid, "controller": True, "blockOwnerDeletion": True, }], }, "spec": { "podSelector": {"matchLabels": {"overdeck.dev/factory-job": job_name}}, "policyTypes": ["Egress"], "egress": [{ "to": [{"ipBlock": {"cidr": f"{APPROVED_RESULT_HOST}/32"}}], "ports": [{"protocol": "TCP", "port": parsed.port}], }], }, } def kubectl(namespace: str, kubeconfig: str | None, *args: str) -> list[str]: out = ["kubectl"] if kubeconfig: out += ["--kubeconfig", kubeconfig] return out + ["--namespace", namespace, *args] def delete_remote_ref(repo: Path, ref: str) -> str | None: probe = run(["git", "ls-remote", "--refs", "origin", ref], cwd=repo, check=False) if probe.returncode: return f"could not inspect {ref}: {(probe.stderr or probe.stdout).strip()}" fields = probe.stdout.split() if not fields: return None if len(fields) != 2 or fields[1] != ref or not re.fullmatch(r"[0-9a-f]{40,64}", fields[0]): return f"could not establish the expected object for {ref}" expected = fields[0] deleted = run( ["git", "push", f"--force-with-lease={ref}:{expected}", "origin", f":{ref}"], cwd=repo, check=False, ) if deleted.returncode: return f"could not delete {ref}: {(deleted.stderr or deleted.stdout).strip()}" return None def reconcile_attempts(repo: Path, output_root: Path, kubeconfig: str | None, *, now: int | None = None) -> None: current = int(time.time()) if now is None else now errors: list[str] = [] if not output_root.exists(): return for directory in output_root.iterdir(): receipt_path = directory / "attempt.json" if not directory.is_dir() or not re.fullmatch(r"factory-[0-9a-f]{16}", directory.name) or not receipt_path.is_file(): continue try: receipt = json.loads(receipt_path.read_text()) required = { "version", "attempt_id", "job_name", "upload_secret", "mailbox_policy", "namespace", "execution_ref", "execution_commit", "result_ref", "expires_at", "state", } allowed = required | {"result_retained", "trace_error"} if set(receipt) - allowed or not required.issubset(receipt) or receipt["version"] != 1 or receipt["state"] == "cleaned" or receipt["expires_at"] > current: continue result_retained = receipt.get("result_retained", False) has_trace_error = "trace_error" in receipt trace_error = receipt.get("trace_error") if ( not isinstance(result_retained, bool) or ( has_trace_error and ( not isinstance(trace_error, str) or not trace_error or len(trace_error.encode("utf-8")) > MAX_RECOVERY_TRACE_ERROR_BYTES ) ) or receipt["job_name"] != directory.name or receipt["job_name"] != f"factory-{receipt['attempt_id'][:16]}" or receipt["upload_secret"] != f"factory-upload-{receipt['attempt_id'][:16]}" or receipt["mailbox_policy"] != f"factory-mailbox-{receipt['attempt_id'][:16]}" or not repository_registry.ATTEMPT_ID.fullmatch(receipt["attempt_id"]) or not repository_registry.EXECUTION_REF.fullmatch(receipt["execution_ref"]) or not repository_registry.RESULT_REF.fullmatch(receipt["result_ref"]) or not repository_registry.COMMIT_ID.fullmatch(receipt["execution_commit"]) or not isinstance(receipt["expires_at"], int) ): raise ValueError("invalid recovery receipt") label(receipt["namespace"], "namespace") name = receipt["job_name"] attempt_errors: list[str] = [] deleted = run(kubectl(receipt["namespace"], kubeconfig, "delete", "job", name, "--ignore-not-found", "--cascade=foreground", "--wait=true"), check=False, timeout=30) if deleted.returncode: attempt_errors.append(f"could not reconcile job {name}: {(deleted.stderr or deleted.stdout).strip()}") refs = [receipt["execution_ref"]] if not result_retained: refs.insert(0, receipt["result_ref"]) for ref in refs: if error := delete_remote_ref(repo, ref): attempt_errors.append(error) if not attempt_errors: receipt["state"] = "cleaned" write_receipt(receipt_path, receipt) errors.extend(attempt_errors) except (OSError, ValueError, TypeError, json.JSONDecodeError) as exc: errors.append(f"could not reconcile {receipt_path}: {exc}") if errors: raise RuntimeError("; ".join(errors)) def worker_termination(pod: dict[str, Any]) -> dict[str, Any] | None: statuses = pod.get("status", {}).get("containerStatuses", []) factory = next((item for item in statuses if item.get("name") == "factory"), None) return factory.get("state", {}).get("terminated") if factory else None def result_ready(namespace: str, kubeconfig: str | None, pod_name: str) -> bool: probe = run( kubectl(namespace, kubeconfig, "exec", pod_name, "-c", "result-reader", "--", "/usr/bin/test", "-f", RESULT_MARKER_PATH), check=False, timeout=15, ) return probe.returncode == 0 def copy_result_bundle(namespace: str, kubeconfig: str | None, pod_name: str, output: Path) -> Path: """Copy the bounded result volume without ever routing artifact bytes through logs.""" path = output / "result.bundle" command = kubectl(namespace, kubeconfig, "exec", pod_name, "-c", "result-reader", "--", "/bin/cat", RESULT_BUNDLE_PATH) try: with path.open("wb") as destination: copied = subprocess.run(command, stdout=destination, stderr=subprocess.PIPE, check=False, timeout=30) except subprocess.TimeoutExpired as exc: path.unlink(missing_ok=True) raise RuntimeError("timed out copying Kubernetes result bundle") from exc if copied.returncode: path.unlink(missing_ok=True) raise RuntimeError( f"could not copy Kubernetes result bundle: {copied.stderr.decode(errors='replace').strip()[-1200:]}" ) if not path.is_file() or not 0 < path.stat().st_size <= result_mailbox.MAX_BUNDLE_BYTES: path.unlink(missing_ok=True) raise RuntimeError("Kubernetes result bundle size is not approved") return path def worker_log_ready(pod: dict[str, Any]) -> bool: statuses = pod.get("status", {}).get("containerStatuses", []) factory = next((item for item in statuses if item.get("name") == "factory"), None) state = factory.get("state", {}) if factory else {} return "running" in state or "terminated" in state def run_limited_git( repo: Path, *args: str, env: dict[str, str] | None = None, input_path: Path | None = None, output_path: Path | None = None, file_limit: int = result_mailbox.MAX_BUNDLE_BYTES * 2, check: bool = True, ): argv = [ "/usr/bin/prlimit", f"--as={PROCESS_ADDRESS_SPACE_BYTES}", f"--fsize={file_limit}", f"--cpu={RESULT_PROCESSING_TIMEOUT_SECONDS}", "--", "git", *args, ] source = input_path.open("rb") if input_path is not None else None destination = output_path.open("wb") if output_path is not None else subprocess.PIPE errors = tempfile.TemporaryFile() try: completed = subprocess.run( argv, cwd=repo, env=env, stdin=source, stdout=destination, stderr=errors, check=False, timeout=RESULT_PROCESSING_TIMEOUT_SECONDS + 5, ) errors.seek(0, os.SEEK_END) error_size = errors.tell() errors.seek(max(0, error_size - 1200)) error = errors.read().decode(errors="replace").strip() finally: errors.close() if source is not None: source.close() if output_path is not None: destination.close() if check and completed.returncode: raise RuntimeError(f"git exited {completed.returncode}: {error}") return completed def validate_quarantined_objects( repo: Path, quarantine: Path, result_commit: str, input_commit: str, env: dict[str, str], directory: Path, ) -> None: objects = directory / "objects" run_limited_git( repo, "--git-dir", str(quarantine), "rev-list", "--objects", "--no-object-names", result_commit, f"^{input_commit}", env=env, output_path=objects, file_limit=MAX_OBJECT_LIST_BYTES, ) object_count = 0 with objects.open() as source: for line in source: object_count += 1 if object_count > MAX_RESULT_OBJECTS or not repository_registry.COMMIT_ID.fullmatch(line.strip()): raise RuntimeError("result bundle object count or identity is not approved") sizes = directory / "object-sizes" run_limited_git( repo, "--git-dir", str(quarantine), "cat-file", "--batch-check=%(objectname) %(objecttype) %(objectsize)", env=env, input_path=objects, output_path=sizes, file_limit=MAX_OBJECT_LIST_BYTES, ) total = 0 measured = 0 with sizes.open() as source: for line in source: fields = line.split() if len(fields) != 3 or not repository_registry.COMMIT_ID.fullmatch(fields[0]) or fields[1] not in {"blob", "commit", "tag", "tree"}: raise RuntimeError("result bundle object metadata is invalid") try: size = int(fields[2]) except ValueError as exc: raise RuntimeError("result bundle object size is invalid") from exc if size < 0 or size > MAX_RESULT_OBJECT_BYTES: raise RuntimeError("result bundle contains an oversized object") total += size measured += 1 if total > MAX_RESULT_EXPANDED_BYTES: raise RuntimeError("result bundle expanded size is not approved") if measured != object_count: raise RuntimeError("result bundle object inventory is incomplete") def import_result(repo: Path, bundle: Path, input_commit: str) -> str: if not bundle.is_file() or not 0 < bundle.stat().st_size <= result_mailbox.MAX_BUNDLE_BYTES: raise RuntimeError("result bundle size is not approved") common_directory = Path(git(repo, "rev-parse", "--path-format=absolute", "--git-common-dir")) object_directory = common_directory / "objects" if not object_directory.is_dir(): raise RuntimeError("controller Git object directory is unavailable") with tempfile.TemporaryDirectory(prefix="factory-result-quarantine-", dir=bundle.parent) as temporary: directory = Path(temporary) run_limited_git( repo, "bundle", "verify", str(bundle), output_path=directory / "bundle-verification", file_limit=MAX_OBJECT_LIST_BYTES, ) quarantine = directory / "repository.git" run_limited_git(repo, "init", "--bare", str(quarantine)) alternate = str(object_directory) if "\n" in alternate: raise RuntimeError("controller Git object directory is invalid") (quarantine / "objects/info/alternates").write_text(alternate + "\n") env = os.environ.copy() env["GIT_ALTERNATE_OBJECT_DIRECTORIES"] = str(object_directory) run_limited_git( repo, "--git-dir", str(quarantine), "-c", "fetch.unpackLimit=0", "-c", "fetch.fsckObjects=true", "fetch", "--no-tags", str(bundle), "HEAD", env=env, ) result_commit = run_limited_git( repo, "--git-dir", str(quarantine), "rev-parse", "FETCH_HEAD^{commit}", env=env, ).stdout.decode().strip() if run_limited_git( repo, "--git-dir", str(quarantine), "merge-base", "--is-ancestor", input_commit, result_commit, env=env, check=False, ).returncode: raise RuntimeError("result bundle does not descend from the submitted workspace") validate_quarantined_objects(repo, quarantine, result_commit, input_commit, env, directory) run_limited_git( repo, "--git-dir", str(quarantine), "fsck", "--strict", "--connectivity-only", "--no-reflogs", "--no-dangling", result_commit, env=env, ) quarantine_ref = "refs/factory/result" run_limited_git( repo, "--git-dir", str(quarantine), "update-ref", quarantine_ref, result_commit, env=env, ) run_limited_git( repo, "-c", "fetch.unpackLimit=0", "fetch", "--no-tags", str(quarantine), quarantine_ref, env=env, ) imported = run_limited_git(repo, "rev-parse", "FETCH_HEAD^{commit}").stdout.decode().strip() if imported != result_commit: raise RuntimeError("validated result commit changed during import") return result_commit def write_result_patch(repo: Path, input_commit: str, result_commit: str, path: Path) -> None: run_limited_git( repo, "diff", "--binary", input_commit, result_commit, "--", output_path=path, file_limit=MAX_RESULT_PATCH_BYTES, ) def publish_result(repo: Path, result_commit: str, result_ref: str) -> None: run_limited_git(repo, "push", "origin", f"{result_commit}:{result_ref}") def submit(repo: Path, invocation: Sequence[str], repository_id: str, *, key: bytes | None = None, registry_path: Path = repository_registry.REGISTRY_PATH) -> Path: registry = repository_registry.load(registry_path) repository_registry.resolve(registry, repository_id, repository_registry.WORKFLOW) repository = repository_registry.authorize_origin( registry, repository_id, repository_registry.WORKFLOW, origin(repo) ) image = os.getenv("FACTORY_K8S_IMAGE", "") secret = repository.credential.handle namespace = os.getenv("FACTORY_K8S_NAMESPACE", "overdeck-factory") account = os.getenv("FACTORY_K8S_SERVICE_ACCOUNT", "overdeck-factory-runner") kubeconfig = os.getenv("FACTORY_K8S_KUBECONFIG") or None timeout = int(os.getenv("FACTORY_K8S_TIMEOUT_SECONDS", "7200")) _, input_tree, input_commit = workspace_snapshot(repo) nonce = secrets.token_hex(12) execution_ref = f"refs/heads/factory-exec/{nonce}" result_ref = f"refs/heads/factory-result/{nonce}" name = f"factory-{nonce[:16]}" upload_secret = f"factory-upload-{nonce[:16]}" mailbox_policy = f"factory-mailbox-{nonce[:16]}" output_root = Path(git(repo, "rev-parse", "--path-format=absolute", "--git-path", "factory-kubernetes")) reconcile_attempts(repo, output_root, kubeconfig) output = output_root / name output.mkdir(parents=True, exist_ok=True) mailbox = result_mailbox.ResultMailbox( os.getenv("FACTORY_K8S_RESULT_HOST", ""), nonce, output ) primary: BaseException | None = None cleanup_errors: list[str] = [] creation_attempted = False retained_result = False result_path: Path | None = None patch_path = output / "result.patch" receipt: dict[str, Any] | None = None trace_mirror: tracer.KubernetesTraceMirror | None = None log_follower: JobLogFollower | None = None log_receipt_written = False trace_complete = False try: upload = mailbox.start() upload_json = result_mailbox.serialize_target(upload) envelope = repository_registry.issue_envelope( registry, repository_id, repository_registry.WORKFLOW, signing_key(key), attempt_id=nonce, execution_ref=execution_ref, result_ref=result_ref, input_commit=input_commit, invocation=invocation, upload_sha256=result_mailbox.target_sha256(upload), ttl_seconds=timeout, ) job = manifest( job_name=name, namespace=namespace, image=image, secret=secret, upload_secret=upload_secret, service_account=account, envelope=envelope, execution_ref=execution_ref, result_ref=result_ref, invocation=invocation, upload=upload, timeout=timeout, pull_secret=os.getenv("FACTORY_K8S_IMAGE_PULL_SECRET", "overdeck-ghcr-pull"), runtime_secret=os.getenv("FACTORY_K8S_RUNTIME_SECRET", "overdeck-factory-runtime"), ) receipt = { "version": 1, "attempt_id": nonce, "job_name": name, "upload_secret": upload_secret, "mailbox_policy": mailbox_policy, "namespace": namespace, "execution_ref": execution_ref, "execution_commit": input_commit, "result_ref": result_ref, "expires_at": envelope.expires_at, "state": "prepared", } write_receipt(output / "attempt.json", receipt) git(repo, "push", "origin", f"{input_commit}:{execution_ref}") creation_attempted = True created_job = run( kubectl(namespace, kubeconfig, "create", "-f", "-", "-o", "json"), input_data=json.dumps(job), timeout=30, ) job_uid = json.loads(created_job.stdout).get("metadata", {}).get("uid", "") network_document = mailbox_network_policy(namespace, mailbox_policy, name, job_uid, upload) run(kubectl(namespace, kubeconfig, "create", "-f", "-"), input_data=json.dumps(network_document), timeout=30) secret_document = upload_secret_manifest(namespace, upload_secret, name, job_uid, upload_json) run(kubectl(namespace, kubeconfig, "create", "-f", "-"), input_data=json.dumps(secret_document), timeout=30) run( kubectl(namespace, kubeconfig, "patch", "job", name, "--type=merge", "-p", '{"spec":{"suspend":false}}'), timeout=30, ) deadline = time.monotonic() + timeout + 30 pod: dict[str, Any] | None = None terminated: dict[str, Any] | None = None while time.monotonic() < deadline: pods = json.loads(run(kubectl(namespace, kubeconfig, "get", "pods", "-l", f"job-name={name}", "-o", "json"), timeout=30).stdout) items = pods.get("items", []) if len(items) > 1: raise RuntimeError(f"Kubernetes Job {name} has multiple worker Pods") if len(items) == 1: pod = items[0] pod_name = pod.get("metadata", {}).get("name", "") if log_follower is None and pod_name and worker_log_ready(pod): metadata = pod.get("metadata", {}) spec = pod.get("spec", {}) placement = { key: value for key, value in { "cluster": "k3s", "namespace": namespace, "job": name, "job_uid": job_uid, "pod": pod_name, "pod_uid": metadata.get("uid"), "node": spec.get("nodeName"), "container": "factory", }.items() if isinstance(value, str) and value } trace_db, trace_events = canonical_trace_paths() trace_mirror = tracer.KubernetesTraceMirror( trace_db, trace_events, nonce, placement, ) log_follower = JobLogFollower( kubectl(namespace, kubeconfig, "logs", "-f", pod_name, "-c", "factory"), output / "job.log", trace_mirror, ) terminated = worker_termination(pod) if terminated is not None and pod_name and result_ready(namespace, kubeconfig, pod_name): break time.sleep(2) else: raise TimeoutError(f"Kubernetes Job {name} did not produce a result before its deadline") assert pod is not None and terminated is not None if log_follower is None or trace_mirror is None: raise RuntimeError(f"Kubernetes Job {name} ended without a trace follower") log_truncated, log_error = log_follower.finish() log_receipt = { "bytes": (output / "job.log").stat().st_size, "error": log_error, "truncated": log_truncated, } (output / "job-log.json").write_text(json.dumps(log_receipt, sort_keys=True) + "\n") log_receipt_written = True worker_failed = terminated.get("exitCode") != 0 retain_on_failure = envelope.result_publication.get("retain_on_failure") is True trace_failure: str | None = None if log_error is not None or log_truncated: trace_failure = f"Kubernetes Job {name} trace failed: {log_error or 'log truncated'}" else: try: trace_mirror.complete() except RuntimeError as exc: trace_failure = f"Kubernetes Job {name} trace failed: {exc}" else: trace_complete = True if trace_failure is not None: trace_mirror.fail() if receipt is not None: receipt["trace_error"] = trace_failure write_receipt(output / "attempt.json", receipt) if not (worker_failed and retain_on_failure): raise RuntimeError(trace_failure) (output / "status.json").write_text(json.dumps(pod, indent=2) + "\n") if worker_failed and not retain_on_failure: raise RuntimeError(f"Kubernetes Job {name} failed; API output saved under {output}") bundle_path = copy_result_bundle(namespace, kubeconfig, pod_name, output) try: result_commit = import_result(repo, bundle_path, input_commit) if worker_failed: publish_result(repo, result_commit, result_ref) except Exception as exc: if worker_failed and trace_failure is not None: raise RuntimeError( f"{trace_failure}; failed result bundle was not retained: {exc}" ) from exc raise if worker_failed: retained_result = True if receipt is not None: receipt["result_retained"] = True write_receipt(output / "attempt.json", receipt) failure = ( f"Kubernetes Job {name} failed; validated result retained at {result_ref}; " f"API output saved under {output}" ) if trace_failure is not None: failure = f"{trace_failure}; {failure}" raise RuntimeError(failure) if workspace_snapshot(repo)[1] != input_tree: raise RuntimeError("workspace changed while Kubernetes Job ran; result not applied") write_result_patch(repo, input_commit, result_commit, patch_path) if patch_path.stat().st_size: run_limited_git(repo, "apply", "--check", "--binary", "-", input_path=patch_path) publish_result(repo, result_commit, result_ref) if workspace_snapshot(repo)[1] != input_tree: raise RuntimeError("workspace changed before validated result application") if patch_path.stat().st_size: run_limited_git(repo, "apply", "--binary", "-", input_path=patch_path) result_path = output except BaseException as exc: primary = exc (output / "recovery.json").write_text(json.dumps({"execution_ref": execution_ref, "result_ref": result_ref, "error": str(exc)}, indent=2) + "\n") finally: if log_follower is not None and not log_receipt_written: try: log_follower.abort() log_path = output / "job.log" (output / "job-log.json").write_text(json.dumps({ "bytes": log_path.stat().st_size if log_path.exists() else 0, "error": "controller stopped before the log follower completed", "truncated": log_follower.truncated, }, sort_keys=True) + "\n") log_receipt_written = True except Exception as exc: cleanup_errors.append(f"could not stop Kubernetes log follower: {exc}") if trace_mirror is not None: try: if not trace_complete: trace_mirror.fail() trace_mirror.close() except Exception as exc: cleanup_errors.append(f"could not close Kubernetes trace mirror: {exc}") try: mailbox.close() except Exception as exc: cleanup_errors.append(f"could not close result mailbox: {exc}") try: patch_path.unlink(missing_ok=True) except OSError as exc: cleanup_errors.append(f"could not remove result patch: {exc}") if creation_attempted: try: deleted = run(kubectl(namespace, kubeconfig, "delete", "job", name, "--ignore-not-found", "--cascade=foreground", "--wait=true"), check=False, timeout=30) except subprocess.TimeoutExpired: try: remaining = run(kubectl(namespace, kubeconfig, "get", "job", name, "--ignore-not-found", "-o", "name"), check=False, timeout=30) except subprocess.TimeoutExpired: cleanup_errors.append(f"could not confirm deletion of Job {name}: request timed out") else: if remaining.returncode: cleanup_errors.append(f"could not confirm deletion of Job {name}: {(remaining.stderr or remaining.stdout).strip()}") elif remaining.stdout.strip(): cleanup_errors.append(f"could not delete Job {name}: foreground deletion timed out") else: if deleted.returncode: cleanup_errors.append(f"could not delete Job {name}: {(deleted.stderr or deleted.stdout).strip()}") refs = [execution_ref] if not retained_result: refs.insert(0, result_ref) for ref in refs: if error := delete_remote_ref(repo, ref): cleanup_errors.append(error) if not cleanup_errors and receipt is not None: receipt["state"] = "cleaned" write_receipt(output / "attempt.json", receipt) if primary: for error in cleanup_errors: primary.add_note(error) raise primary if cleanup_errors: raise RuntimeError("; ".join(cleanup_errors)) assert result_path is not None return result_path def main(argv=None) -> int: parser = argparse.ArgumentParser(prog="factory kubernetes") parser.add_argument("--repo", default=".") parser.add_argument("--repository-id", required=True) parser.add_argument("invocation", nargs=argparse.REMAINDER) args = parser.parse_args(argv) repo = Path(git(Path(args.repo).resolve(), "rev-parse", "--show-toplevel")) if not args.invocation: parser.error("an ADW invocation is required") print(f"factory kubernetes: completed; API output: {submit(repo, args.invocation, args.repository_id)}") return 0 if __name__ == "__main__": raise SystemExit(main())