58. Walkthrough: Design a Cluster Scheduler / Container Orchestrator (Kubernetes, Borg)

Step 0 — Clarify the business goal

🎤 Candidate: The business outcomes are utilization: machines are the biggest infrastructure cost, and better bin packing saves real money. Also developer velocity (self-serve deploys) and reliability (automatic recovery). Do we mix latency-sensitive services and batch jobs on the same machines? That's where most of the utilization gain comes from, and most of the complexity.

🧑‍💼 Interviewer: Yes, both, on ~10,000-machine clusters.

Step 1–2 — Requirements

Step 5 — High-level architecture

 Users / CI ─▶ API server (authN/Z, validation, admission policies) ─▶ state store (etcd: Raft)
                        ▲  watch streams (list + watch; informers cache state in every component)
     ┌──────────────────┼────────────────────────┬────────────────────────────┐
     │                  │                        │                            │
 Controllers        Scheduler                 Node agents (kubelet)        Autoscalers
 (deployments,      watch unscheduled pods →   watch pods bound to node →   (replicas, nodes)
  replicasets,      FILTER feasible nodes →    start/stop containers via
  jobs, endpoints)  SCORE (bin-pack, spread,   runtime; report status;
 reconcile desired  affinity) → BIND (write    enforce cgroups limits;
 vs actual state    pod.nodeName, optimistic   evict on resource pressure
                    concurrency)

The core idea: level-triggered reconciliation. Every controller loops "observe actual state → compare with desired → act to converge". It doesn't react to individual events ("pod died"). It reacts to the current difference, so missed events, restarts, and duplicate notifications are harmless. That's why the system self-heals.

Step 6 — Deep dives

1. Scheduling algorithm: filter (resources, taints, affinity, ports, volumes zone) → score (least-allocated vs most-allocated bin packing, topology spread, image locality) → bind. To scale: sample a subset of feasible nodes on large clusters, use optimistic concurrency on binding (conflicts are retried), and batch. Google's Omega paper (2013) compares monolithic, two-level (Mesos), and shared-state optimistic schedulers; the shared-state idea influenced Kubernetes.
2. Priorities, preemption, and quotas: production workloads can evict lower-priority batch work. Quotas per team stop one org from taking the whole cluster. Pod disruption budgets limit voluntary evictions.
3. Utilization (the business case): Borg (EuroSys 2015) reported large savings from mixing production and batch on shared cells and reclaiming resources that production jobs reserved but didn't use. The catch is noisy neighbors: CPU throttling (§22.6), memory pressure, and I/O contention need isolation (cgroups, CPU pinning for latency-critical work) and eviction policies.
4. Control-plane scalability: list/watch with informer caches so components don't poll the API server; watch bookmarks and resource versions; etcd sized carefully (it's a consensus store with size limits, §19). The upstream Kubernetes scalability thresholds are about 5,000 nodes and 150,000 pods per cluster, so larger fleets use many clusters (cells, §20.4) with multi-cluster tooling. Meta's Twine (OSDI 2020) takes a different approach: one control plane managing very large shared regions.
5. Failure handling: node heartbeats stop → mark the node unhealthy after a grace period → evict and reschedule pods. Stateful workloads need fencing (§19.3): a partitioned node may still be running the old pod, so volume attachment and leases must prevent two writers.

Step 7 — Failure modes & corner cases