Multi-Cluster Process Execution — Plan

Goal

Support running process jobs across multiple Kubernetes clusters (e.g. a local minikube plus a GKE cluster), with the cluster for a given job chosen dynamically — per user, per project, or per job — by a plugin hook, while every cluster can read/write the same project's storage regardless of where the job lands.

Depends on

short-lived-storage-credentials-01-storage-backend-model.md and short-lived-storage-credentials-02-hooks-run-first-select-storage.md (Phases 1–2 of short-lived-storage-credentials-00-overview.md): the StorageBackend model, Project.storage_backend_id, and hooks.run_first / sorted-entry-point infrastructure. Multi-cluster execution is only sound once a project's storage is already resolved independently of any cluster — this plan does not re-litigate that, it assumes it. It does not depend on that plan's Phases 3–4 (short-lived credential minting/refresh) — this can be built first with static-key credentials and short-lived credentials layered in later without changes to the cluster-selection design below.

Background — current state

Architecture Summary

Phase 1 — Cluster model + bootstrap migration

1.1 Model

New file: backend/models/cluster.py

from sqlalchemy import Column, String, DateTime, JSON
from datetime import datetime
import uuid

from backend.database import Base


class Cluster(Base):
    __tablename__ = "clusters"

    id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
    name = Column(String(255), nullable=False)
    # NULL kubeconfig = auto-detect (in-cluster config or local kubeconfig), matching
    # today's K8sClient._ensure_initialized() behavior exactly.
    kubeconfig = Column(JSON, nullable=True)
    registry_url = Column(String(255), nullable=True)
    registry_auth = Column(String(255), nullable=True)
    namespace = Column(String(255), nullable=False, default="nagelfluh-jobs")
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)

    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "registry_url": self.registry_url,
            "namespace": self.namespace,
            "created_at": self.created_at.isoformat(),
        }

Add to ProcessVersion (backend/models/process.py), next to the existing k8s_namespace column:

k8s_cluster_id = Column(String(36), ForeignKey("clusters.id"), nullable=True)

Nullable and left NULL on historical rows — unlike Project.storage_backend_id in the dependency plan, there is no need to backfill past process versions; they already ran and are done. Only newly-launched jobs populate it.

1.2 Bootstrap migration

Same pattern as short-lived-storage-credentials-01-storage-backend-model.md §1.2 and the existing 3e9d7f5a8c2d_add_bootstrap_environment.py / e2f3a4b5c6d7_seed_initial_admin.py migrations:

"""seed default cluster from config.env"""
from alembic import op
import sqlalchemy as sa
from datetime import datetime
import os

revision = '<new>'
down_revision = '<add_clusters_table>'

DEFAULT_ID = 'default-cluster-00000000-0000-0000-0000-000000000000'


def upgrade() -> None:
    from backend.config import settings
    conn = op.get_bind()

    exists = conn.execute(
        sa.text("SELECT COUNT(*) FROM clusters WHERE id = :id"), {"id": DEFAULT_ID}
    ).scalar()

    if not exists:
        conn.execute(sa.text("""
            INSERT INTO clusters
                (id, name, kubeconfig, registry_url, registry_auth, namespace, created_at)
            VALUES
                (:id, 'Default Cluster', NULL, :registry_url, :registry_auth, :namespace, :created_at)
        """), {
            "id": DEFAULT_ID,
            "registry_url": settings.registry_url,
            "registry_auth": settings.registry_auth,
            # K8S_NAMESPACE is a raw env var read directly in k8s_client.py, NOT a field
            # on backend.config.Settings — read it the same raw way here.
            "namespace": os.getenv("K8S_NAMESPACE", "nagelfluh-jobs"),
            "created_at": datetime.utcnow().isoformat(),
        })


def downgrade() -> None:
    pass

kubeconfig=NULL preserves today's auto-detect behavior exactly — no existing deployment needs to supply an actual kubeconfig blob to keep working. Unlike the storage-backend bootstrap migration, no backfill of existing rows is needed here (see 1.1) — the default row exists purely as the fallback select_cluster resolves to for jobs launched after this migration runs.

Phase 2 — select_cluster hook

Uses hooks.run_first from the dependency plan — no new hook-runner mechanics needed here.

backend/models/process.py, in run_task(), immediately after the existing job_pre_run call (~line 730):

DEFAULT_CLUSTER_ID = 'default-cluster-00000000-0000-0000-0000-000000000000'

cluster_id = hooks.run_first.select_cluster(DEFAULT_CLUSTER_ID, db, user, process, process_version)
process_version.k8s_cluster_id = cluster_id
await db.commit()

A plugin can key off any combination of user (tier/identity), process.project_id, process.type, or process_version.resource_requests (e.g. route GPU-requesting jobs to a GKE node pool with GPUs, everything else to minikube) — all already present in this call's arguments, no new data threading required. Per hooks.run_first's semantics (established in the dependency plan), the first plugin (by sorted plugin name) to return non-None wins; disagreement between plugins is not an error, just silently resolved by that ordering.

Phase 3 — Per-cluster K8s client registry

backend/services/k8s_client.py changes from one module-level k8s_client = K8sClient() singleton to a registry:

class K8sClientRegistry:
    def __init__(self):
        self._clients = {}  # cluster_id -> K8sClient

    def get(self, cluster: "Cluster") -> K8sClient:
        if cluster.id not in self._clients:
            self._clients[cluster.id] = K8sClient(
                namespace=cluster.namespace,
                kubeconfig=cluster.kubeconfig,
            )
        return self._clients[cluster.id]


k8s_clients = K8sClientRegistry()

K8sClient.__init__ and _ensure_initialized() change to accept an optional kubeconfig dict — when None, fall back to today's exact auto-detect logic (load_incluster_config() then load_kube_config()); when set, load that config explicitly (kubernetes_asyncio.config.load_kube_config_from_dict() or equivalent).

Every call site that references the module-level k8s_client singleton today — job_orchestrator.py, monitor_job, LogManager, the cancel endpoint in routers/processes.py, routers/utilities.py — changes to first resolve the relevant Cluster row (via process_version.k8s_cluster_id, falling back to the default cluster for any pre-existing rows where it's NULL) and then k8s_clients.get(cluster).

Phase 4 — Routing existing operations through the resolved cluster

Each of these currently assumes the single global k8s_client; each needs the same one-line change — resolve Cluster from process_version.k8s_cluster_id, then use k8s_clients.get(cluster) in place of the module-level k8s_client:

Implementation Order

  1. Phase 1 — Cluster model, ProcessVersion.k8s_cluster_id, bootstrap migration. Schema-only; no behavior change (nothing reads k8s_cluster_id yet).
  2. Phase 3 — K8sClientRegistry, K8sClient accepting explicit kubeconfig. Still no behavior change with one Cluster row — the registry always resolves the same client the singleton used to be.
  3. Phase 2 — select_cluster hook call site, populating k8s_cluster_id on every new job. Still resolves to the default cluster with zero plugins installed.
  4. Phase 4 — thread k8s_cluster_id through job creation, monitoring, logs, and cancel. This is the phase that actually makes a second Cluster row usable end-to-end; until it's complete, adding a second Cluster row and a select_cluster plugin would create jobs that can't be monitored or canceled correctly.

Open Questions