Design a Container Orchestrator
Place 300,000 containers on 10,000 machines from declared desired state, keep them running through failures, and give every service a stable address.
Last updated 2026-10-27. Difficulty: hard. Patterns: scheduling, reconciliation, consensus, bin-packing, failure-detection. Reported at Google, Microsoft, Amazon, Databricks, Uber, Nvidia.
The interviewer asks, the candidate answers and draws, and you press Next. Skip to any stage or key moment, and ask the AI Mentor anything along the way.
Or sit it in the style of Google · Microsoft · Amazon · Databricks, run as that company’s round is reported to go.
The short answer
How to design Container Orchestrator in a system design interview, in five steps:
- Scope it. Declare a workload, place pods on machines, keep the declared count running, roll out a new version, stable service addresses, priorities, quotas and preemption, resource isolation and overcommit, planned maintenance, status and events.
- Set the targets. Cluster scale: 10,000 machines · 300,000 pods; Scheduling throughput and latency: 2,000 placements/s · prod p99 < 5 s; Recovery from machine failure: detect ≤ 40 s · replaced ≤ 3 min; Static stability: control plane down → 0 pods stop.
- Do the numbers. State in the store: ~1.5 GB; Placements per second: ~170/s avg · 2,000/s peak; Scheduler CPU at peak: ~20 cores full scan · ~1 core sampled.
- Draw the design. API servers, Watch cache, State store, Lease store, Scheduler, Scheduling queue, Node agent, Image registry, Usage metrics.
- Go deep where it is hard.
- Desired state in a small consistent store → Raft-replicated store of desired and actual state; every component a level-triggered controller watching it.
- Placing 2,000 pods a second → Filter, score a sample, bind optimistically: parallel workers on one snapshot, a conditional bind, requeue on conflict.
- Priorities, preemption and overcommit → Priority bands with preemption, plus batch on reclaimed capacity (requests guaranteed, the gap revocable).
- Machine failure, partitions and fencing → Leases renewed by agents, a wait before eviction, a zone-silence brake, self-fencing and epochs for at-most-one pods.
- Stable addresses and endpoint fan-out → Name plus virtual IP; endpoint slices from readiness; versioned deltas to per-machine proxies and load balancers.
- Scaling the control plane, and cells → Watch caches in the API servers, sharded controllers, a capped size, many clusters per region as cells.
Functional requirements
- Declare a workload. A service (keep N replicas running) or a batch job (run N tasks to completion): image digest, replicas, CPU and memory requests and limits, placement constraints (zone, machine class, GPUs, spread), health checks and a priority. Teams say what should run; the system works out where and keeps it true.
- Place pods on machines. A pod is one or more containers placed together, usually one. Each lands on a machine that meets its constraints and has room for its requests. Production pods are placed within seconds; a batch job of 100,000 tasks may queue, but it must never delay production.
- Keep the declared count running. Restart a crashed container in place, and replace the pods of a failed or drained machine elsewhere. Never run two copies of a pod that declares itself at-most-one, such as a shard leader.
- Roll out a new version. A new template replaces pods gradually inside a surge and an unavailability budget, gated on readiness, and can be paused or reversed. Staged rollouts across clusters and canary analysis belong to the deployment system, which drives this API.
- Stable service addresses. A service is a name and a virtual IP backed by its ready pods. Endpoint changes reach every per-machine proxy and load balancer within seconds, and a pod that is unready or draining stops getting new traffic before it goes away.
- Priorities, quotas and preemption. Production outranks batch: when the cluster is full, a production pod evicts batch pods to fit. Quotas per team per priority stop everyone from declaring themselves production.
- Resource isolation and overcommit. Limits are enforced on the machine with cgroups, so a noisy pod hurts only itself. Batch may run on capacity that production reserved but is not using, and gives it back the moment production needs it.
- Planned maintenance. Drain a machine for a kernel upgrade or repair by evicting its pods only as fast as each service’s disruption budget allows, so 500 machines can be upgraded in a night without anyone noticing.
- Status and events. Desired against actual for every workload, and why a pod is pending (“no machine in use1b has 8 free GPUs”). An orchestrator nobody can debug is one nobody trusts.
- Out of scope. Building images and staged deploys across clusters (the deployment system), the container runtime itself (containerd, runc), the packet data plane between pods, persistent volume provisioning, and buying or retiring machines (the capacity planner).
Non-functional requirements
- Cluster scale (10,000 machines · 300,000 pods). A fixed, tested maximum per cluster. A region runs many clusters of this size rather than one bigger one, and every number below is sized against these two.
- Scheduling throughput and latency (2,000 placements/s · prod p99 < 5 s). Batch bursts set the throughput; production pods must never wait behind them. Measured from pod created to pod bound, when capacity exists.
- Recovery from machine failure (detect ≤ 40 s · replaced ≤ 3 min). About 5 machines fail a day, each taking ~30 pods. Faster detection means more false alarms from a stalled agent or a rebooting switch, and the wait before rescheduling is set by what a duplicate would cost.
- Static stability (control plane down → 0 pods stop). Running pods, restarts on the same machine and the traffic between pods must not depend on the control plane. An outage of the store or the API servers stops changes, never serving.
- Consistent desired state (linearizable · every write conditional). Two writers (two scheduler workers, a scheduler and a controller) must never both succeed from stale reads. Every object carries a version and every update is conditional on it.
- Utilisation against safety (≥ 60 % of CPU used). The requirement that drives the hardest trade-off. Packing tightly saves hundreds of machines (1 % of the fleet is 100 machines), while spreading replicas across failure domains and keeping room to reschedule costs capacity. Batch on unused production reservations closes most of the gap.
- Endpoint freshness (readiness change → every proxy < 5 s). Traffic sent to a pod that is draining or dead is user-visible errors. The change must reach 10,000 proxies in seconds, which at rollout peaks is a million updates a second.
- Fairness in the control plane (per-team shares of API and scheduler). A runaway controller or a 100,000-task submission must not slow other teams’ writes, watches or placements. API servers queue requests per client with fair shares; the scheduling queue orders by priority, then by team.
Back-of-envelope estimates
- State in the store: ~1.5 GB. 300,000 pods × ~4 KB (spec and status) = 1.2 GB; 10,000 machines × ~10 KB = 100 MB; ~20,000 services and jobs × ~5 KB = 100 MB; endpoint slices, quotas and config about 100 MB. Roughly 1.5 GB: it fits in memory on every replica and in a Raft store’s comfortable few-GB range only because it holds state. One day of these containers’ logs would be tens of terabytes.
- Placements per second: ~170/s avg · 2,000/s peak. ~100,000 batch pods run at any moment with a typical life of ~10 minutes: 100,000 ÷ 600 s ≈ 170 starts/s. 200,000 service pods replaced about once a day by deploys add only ~2/s. Peaks come from submissions: a 100,000-task job placed at 2,000/s takes 50 s, about 12× the average.
- Scheduler CPU at peak: ~20 cores full scan · ~1 core sampled. Checking and scoring one machine costs ~1 µs, so a pass over 10,000 machines is 10 ms a pod; at 2,000 pods/s that is 20 cores kept busy and a 10 ms floor on every decision. Stopping after ~500 feasible machines and scoring only those is ~0.5 ms, about 1 core. Sampling, not hardware, is what makes 2,000/s cheap.
- Store writes at peak: ~8 k/s. Each pod start is ~4 writes: created by its controller, bound by the scheduler, running and then ready from the node agent. 2,000 starts/s × 4 = 8,000 writes/s of ~4 KB objects. One five-member Raft group on NVMe with batched commits can take this, but it is the busiest the store gets: heartbeats, events and metrics must go elsewhere, and this is a reason to cap the cluster.
- Heartbeat traffic: 1,000/s · ~0.3 MB/s. 10,000 node agents renew a lease every 10 s = 1,000 writes/s of ~300 B, about 0.3 MB/s. Trivial in bytes, but as Raft writes it is an eighth of the main store’s peak load forever, so leases live in their own small Raft group, and full machine status is written only when it changes or every 5 minutes.
- Machine failure to replacements serving: ~135 s. Suspect after 40 s without a renewal (four missed), then wait 60 s more in case it is a partition rather than a death. Placing ~30 pods takes under 5 s; pulling and starting takes ~30 s with most layers cached. 135 s. Traffic is protected sooner: the machine’s endpoints are removed at 40 s, and proxies eject failing endpoints on their own within seconds.
- Endpoint updates fanned out: ~1 M/s at rollout peaks. 200,000 service pods in ~5,000 services; rollouts and failures change up to ~100 endpoints a second at the busiest. A proxy on each of 10,000 machines needs every service: 100 × 10,000 = 1 M updates/s. As ~200 B deltas that is 200 MB/s; resending a 5,000-endpoint service’s whole list (~500 KB) per change would be hundreds of GB a second. This is the biggest stream in the system.
- Capacity reclaimed by overcommit: ~1,500 machines. Production requests reserve ~70 % of the fleet’s CPU but use ~40 % on average, because teams size for peak. Letting batch use half of that 30-point gap, at a priority that can be throttled or evicted, gives back 15 % of 10,000 machines = 1,500 machines, more than any packing tweak.
Components
- Teams / deploy system (CLI · CI · deploy plans): Teams declare workloads, and the deployment system writes new versions into each cluster as one stage of its plan. Both only ever write desired state through the API and read status back; neither talks to a machine. Every change is conditional on the version it read, so two writers cannot silently overwrite each other.
- API servers (stateless · admission · fair queues): The only path to the store, run as ~10 stateless replicas. Authenticates, validates, applies defaults and enforces quotas before anything is written, and turns every update into a write conditional on the object’s version. Queues requests per client with fair shares, so a runaway controller or a 100,000-task submission cannot starve everyone else.
- Watch cache (in API servers · recent history): Each API server holds every object in memory, fed by one watch on the store, plus a few minutes of recent changes. Lists and watches are served from here, so 10,000 agents, the discovery tier and the controllers cost the store ten streams, not thousands. A client that reconnects resumes from its resource version instead of relisting 1.2 GB of pods.
- State store (Raft × 5 · MVCC · ~1.5 GB): An etcd-style key-value store: a Raft log across five replicas, every write stamped with a cluster-wide increasing revision, conditional transactions, and watches that start from a given revision. It holds only state (desired and actual for every pod, machine, workload and endpoint), about 1.5 GB, compacted to a few minutes of history. Its write rate, 8 k/s at peak, is the ceiling for the whole cluster.
- Lease store (separate Raft group · 1 k/s): A small, separate consensus group for node leases (and leader-election leases), so 1,000 heartbeat writes a second never compete with binds in the main store. Each lease is ~300 B: holder, renew time, a 40-second duration and an epoch. Losing it pauses failure detection, not serving.
- Scheduler (filter · score · optimistic bind): Places unbound pods. Keeps an in-memory snapshot of all 10,000 machines; four workers each walk machines from a random offset, filter on constraints and free requests, stop after 500 feasible machines and score those (spread first for production, packing for batch). Binds are writes conditional on the machine’s allocation version, so a conflict requeues the pod instead of overcommitting a machine.
- Scheduling queue (priority · fair share · backoff): Pending pods ordered by priority, then the team’s fair share, then age, so a production pod created mid-burst jumps ahead of 100,000 batch tasks. Pods that fit nowhere move to an unschedulable set and are retried only when the cluster changes in a way that could help, not in a hot loop.
- Node agent (one per machine · cgroups): Runs on every machine and watches only the ~30 pods bound to it. Pulls images by digest, starts containers with cgroup CPU weights and memory limits, runs liveness and readiness probes, restarts crashed containers in place and reports status. Keeps running its pods when the control plane is unreachable, and kills any pod the store no longer lists for it.
- Image registry (by digest · layer cache): Where images come from, run by another team. Agents pull by digest, and most layers are already cached on a busy machine, so a start usually fetches tens of MB. Large rollouts can be prestaged by the deployment system so pods start in seconds.
- Usage metrics (time series · not in the store): Per-pod and per-machine CPU and memory usage every 10 s, from every agent, in a time-series store built for volume. Kept out of the state store on purpose: it is millions of points a second. Feeds autoscaling and the reclaim controller that works out how much reserved but unused capacity batch may borrow.
- Node lifecycle (leases · taints · eviction): Decides when a silent machine is gone. Measures lease silence on its own monotonic clock: at 40 s it marks the machine unreachable and its pods not ready, 60 s later it evicts ordinary pods. If more than 10 % of a zone goes silent at once it assumes a partition and slows eviction to a trickle, because mass rescheduling would turn a network fault into an outage.
- Controllers (level-triggered · sharded): One reconcile loop per kind: services and their rollouts, jobs, endpoint slices, disruption budgets, reclaimed capacity. Each lists, watches from that revision, and repeatedly compares desired with actual for the objects it owns, acting only through the API. Leader-elected, and the heavy ones are sharded by hash; none keeps state the store does not have.
- Discovery service (versioned deltas · ~20 replicas): A stateless fan-out tier between endpoint slices and their 10,000 consumers. Watches the slices once and streams each subscriber versioned deltas of only what changed (~200 B per endpoint), so a busy rollout costs 200 MB/s instead of hundreds of GB. A reconnecting proxy sends its last version and gets only what it missed.
- Node proxy (per machine · VIP → endpoints): Runs on every machine and programs local forwarding (eBPF or IPVS) from each service’s virtual IP to its ready endpoints, preferring the caller’s zone. Holds every service’s endpoints, about 20 MB, and keeps forwarding on its last copy if discovery or the control plane is down. When every endpoint of a service is unready it fails open and uses them anyway.
- Load balancers (L4 / L7 · north-south): The edge balancers from the load balancer design, owned by the networking team. For the few hundred services exposed outside the cluster they subscribe to the same endpoint deltas and still health-check endpoints themselves, with fail-open thresholds, so a readiness bug here cannot empty a pool there.
User flows
- Declare a service, get running pods. The main write path. A team writes desired state once, and four independent loops (controller, scheduler, node agent, endpoints) each watch for their part and write their result back. None of them calls another, so any of them can crash and resume.
- A team applies a service spec: 30 replicas of payments-api, 2 CPUs and 4 GB each, spread over three zones, at production priority. Admission runs before anything is stored: authentication, schema validation, defaults, and the team’s production quota (30 × 2 = 60 cores must fit in what it has left). The write is a create conditional on the name not existing. The only thing that has happened when it returns is that a record exists; nothing is placed by this call.
- The service controller sees the new workload on its watch and creates 30 pod objects, none of them bound to a machine. The controller compares desired (30) with actual (pods owned by this workload: 0) and creates the difference; it never asks what event just happened. Its watch is served from the API servers’ watch cache, so twenty controllers and 10,000 agents cost the store one stream per API server. Each pod records its owner, so a restarted controller recounts instead of creating 30 more.
- The scheduler takes the pods from its queue, filters the machines and scores the ones that pass. Filter: enough unrequested CPU and memory, the right machine class, no taint the pod does not tolerate, and the zone spread still satisfiable. Score: spread across zones and racks first, then prefer fuller machines so whole machines stay free for large pods. It all runs against the scheduler’s in-memory snapshot, about 0.5 ms a pod when it stops after 500 feasible machines.
- The scheduler binds each pod with a write conditional on the machine’s allocation version it decided from. Binding is optimistic: the scheduler assumes its snapshot is right, and the transaction checks that the pod is still unbound and that nothing else has changed the machine’s allocations since. If another worker took that capacity first, the bind fails and the pod goes back to the queue, a few milliseconds lost. Workers start their walks at different machines, so collisions are rare.
- The node agent on that machine sees a pod bound to it, pulls the image and starts the container inside its limits. Each agent watches only the ~30 pods bound to its machine, so 10,000 agents are 10,000 small filtered watches, not 10,000 copies of the cluster. It re-checks fit as a last line of defence, pulls by digest (most layers are cached), and starts the container with a CPU weight from its 2-core request and a hard 4 GiB memory limit, so a leak kills only this container.
- The agent runs the readiness probe and reports the pod running, then ready. Running means the process started; ready means it passed its readiness probe and may get traffic, and everything downstream keys on ready. The agent writes only the status part of the object and coalesces updates, so a pod start costs two status writes, not one per probe.
- The endpoints controller adds the ready pod to the service’s endpoint slice, and discovery pushes the change to every proxy and the load balancers. A service’s endpoints are split into slices of at most 100, so adding one pod rewrites one ~10 KB slice rather than a 5,000-entry list. Discovery watches slices once and sends each subscriber a one-endpoint delta. About 2 s after ready the pod takes traffic; apply to serving is ~30 s, mostly image pull and start.
- A call, and a pod that stops being ready. The read path: a stable name resolves to healthy endpoints on the caller’s own machine, with no control-plane call per request. And when a pod fails its readiness probe, it leaves every proxy’s rotation in seconds without being restarted.
- A pod calls payments-api by name, and the name resolves to the service’s virtual IP. The name and the virtual IP are fixed for the life of the service; only the endpoints behind them change. Because the IP never moves, clients that cache DNS far past its TTL do no harm, which is the classic DNS failure this avoids. The node proxy answers these lookups locally from the data discovery gave it.
- The node proxy on the caller’s machine picks a ready endpoint, preferring its own zone, and forwards the connection. Client-side balancing: no hop through a central balancer, and zone preference keeps traffic off cross-zone links. The proxy holds every service’s endpoints, about 200,000 entries at ~100 B, ~20 MB per machine, and keeps forwarding on that copy if discovery or the whole control plane is unreachable.
- Traffic from outside reaches the same pods through the load balancers, which subscribe to the endpoints of exposed services. Only the few hundred services exposed outside the cluster are sent to the load balancers, not all 5,000. They also health-check endpoints themselves, with fail-open thresholds, as a second opinion: a bug in readiness reporting must not empty a public pool.
- One payments pod starts failing its readiness probe because its database connection pool is exhausted. Three failed probes at 2 s intervals, about 6 s. The agent flips ready to false and does not restart the container, because its liveness probe still passes. Restarting a pod that is merely overloaded would turn a slow dependency into a restart storm.
- The endpoints controller marks that endpoint not ready in its slice. Marked not ready rather than removed, so it can come back without churning the slice, and so a proxy that sees every endpoint of a service unready can fall back to using them all (fail open) instead of failing every request. The controller writes each slice at most once a second, batching changes during a busy rollout.
- Discovery streams a one-endpoint delta to all 10,000 node proxies and to the load balancers. One ~200 B delta × 10,000 proxies is 2 MB, sent from a horizontally scaled tier rather than from 10,000 watches on the API servers. Subscribers acknowledge each version, so a proxy that has fallen behind is visible. Within ~2 s of the slice write the pod gets no new connections, and existing ones finish.
- When the probe passes again the same path puts the pod back; if liveness fails instead, the agent restarts the container in place. A restart in place needs neither the scheduler nor the rest of the control plane. The agent backs off (10 s, 20 s, 40 s, up to 5 minutes) so a crash-looping container does not burn its machine. A pod moves to another machine only if its machine fails or it is evicted.
- A 100,000-task batch job meets production. The scale case: a batch job submits 100,000 tasks at once while production keeps deploying. The scheduler must place 2,000 pods a second, keep production at the front, let batch borrow idle reservations, and when the cluster is full, evict batch for production and never the reverse.
- A team submits a job of 100,000 tasks at batch priority, 1 CPU and 2 GB each. Admission checks the team’s batch quota, which is larger and cheaper than production quota because batch can be evicted. One job object is written; its tasks do not exist yet.
- The job controller creates task pods in batches, paced to what the scheduler can place. 100,000 pod objects written in one second would be 400 MB landing on a store that holds 1.5 GB. Paced at ~2,000 a second, creation matches placement, the pending set stays at a few thousand, and the API servers’ per-client fair share keeps this controller from crowding out everyone else’s writes.
- The scheduling queue orders by priority, so production pods created during the burst jump ahead of the batch tasks. A heap keyed by (priority, team share, age). A production pod created mid-burst is placed in milliseconds instead of waiting 50 s behind 100,000 tasks. Tasks that fit nowhere move to an unschedulable set and are retried when a machine is added or a pod finishes, not every few milliseconds.
- Four scheduler workers decide in parallel against the same snapshot, each sampling machines from a different starting point. Each worker walks machines from a random offset, stops at 500 feasible ones and scores those: ~0.5 ms a pod, so four workers have 8,000 a second of capacity for a 2,000 a second burst. The tasks of one job share an equivalence class, so the filter result for task 51,234 is reused for 51,235. Batch scoring favours packing: the fullest machine that fits.
- Batch tasks may land on capacity that production reserved but is not using, as estimated from recent usage. Agents report usage every 10 s. A reclaim controller writes each machine a batch allocatable figure: unreserved capacity plus production’s requests minus its p95 usage over the last hour, minus a 10 % margin. Batch is filtered against that figure and production against hard requests, so borrowing never weakens a production reservation.
- Binds commit optimistically; the few that lose a race for a machine’s last free slot are requeued. Each bind is a transaction conditional on the machine’s allocation version. With 2,000 binds a second spread over 10,000 machines, two workers rarely pick the same machine within the same few milliseconds, and losing costs one requeue. The Omega paper argued for this shared-state, optimistic model; a lock per machine would serialise exactly the machines everyone wants.
- The cluster is full; a production pod preempts two batch tasks on one machine to fit. When nothing fits, the scheduler finds the machine where evicting the fewest, lowest-priority, most recently started pods frees enough room, and never preempts inside the production band, which would cascade. It deletes the victims with a 10 s grace and nominates the machine, so the freed room is held for the production pod rather than refilled by the next batch task.
- The job controller sees the evicted tasks and recreates them; they wait at batch priority for capacity. Batch is built to be evicted: tasks are short or checkpoint, and an eviction does not count against the job’s retry limit. The Borg paper reports eviction rates for non-production work far above production’s, which is the deal batch accepts for cheap capacity.
- A machine stops heartbeating. The failure path, and the dangerous one: a machine that stops answering might be dead, or cut off and still running its pods. The control plane must replace capacity quickly without ever running two copies of a pod that must be unique, and without mistaking a partitioned zone for 3,000 dead machines.
- Every 10 s each node agent renews its machine’s lease; the renewals from m-use1b-04417 stop. A lease is a ~300 B object (holder, renew time, 40 s duration, epoch) in its own Raft group, so 1,000 renewals a second never compete with binds. Full machine status is written only when it changes or every 5 minutes. Silence alone does not say whether the machine is dead or just unreachable.
- After 40 s without a renewal, the node lifecycle controller marks the machine unreachable and its pods not ready. It measures silence on its own monotonic clock from the last renewal it saw, so a skewed clock on the machine cannot fake liveness. Marking the pods not ready takes them out of every proxy within seconds: traffic stops at 40 s although nothing has been rescheduled. A NoSchedule taint stops new pods landing there.
- Before evicting, it checks how many machines in the zone went silent; a zone-sized silence pauses eviction. One machine failing is normal, about 5 a day. A thousand machines in one zone going silent in the same minute is almost certainly a network fault between them and the control plane, and evicting 30,000 pods would turn a partition into an outage and a rescheduling storm. Above 10 % of a zone silent, eviction slows to a few machines a minute and a human is paged.
- After a further 60 s it evicts the machine’s ordinary pods; at-most-one pods are released only once their lease has surely expired on the machine too. The 60 s wait trades capacity for safety, since most shorter silences are reboots or blips that come back. A pod that declares itself at-most-one (a shard leader, a singleton consumer) is safe to replace only because its agent kills it on its own after 30 s without a successful renewal; its replacement also gets a higher epoch to present to whatever it writes to.
- Controllers see fewer pods than declared and create replacements, which the scheduler places elsewhere. Nothing here is special to failures: the controller sees 29 of 30 and creates one, exactly as on the first day, which is the payoff of level-triggered loops. Thirty pods are placed in well under a second and are serving about 135 s after the machine died. Spread scoring already kept replicas on different racks, so one machine rarely holds two replicas of one service.
- The partition heals; the returning agent finds its pods gone from the store and stops them. During the partition its pods kept running (static stability), so ordinary replicas briefly existed twice, which is harmless for stateless pods behind a balancer. On reconnect the agent relists its own pods, sees that the evicted ones are gone, and stops them. The Borg paper describes the same rule: a returning machine is told to kill tasks that were rescheduled.
- The machine renews its lease, its taint is lifted and it rejoins empty. It comes back empty rather than reclaiming its old pods: they were already replaced, and moving them back is churn for nothing. The lease epoch increments, fencing anything that still holds 18. A machine that goes silent three times in an hour stays tainted and goes to repair, because it costs more in reschedules than it adds in capacity.
- Roll a service to a new version. A rollout is just another controller: a new template is new desired state, and the service controller replaces pods inside a surge and an unavailability budget, gated on readiness. The deployment system drives this cluster by cluster and judges the result; it does not re-implement it.
- The deployment system writes payments-api’s new image digest into this cluster as one stage of its plan. The write is conditional on the resourceVersion it read, so a concurrent edit by the owning team makes it fail with 409 and re-read rather than silently win. The generation goes to 2: comparing it with the controller’s observed generation is how anyone tells whether the new spec has been acted on yet.
- The service controller starts 3 surge pods from the new template. maxSurge 3 and maxUnavailable 0 mean at least 30 ready pods and at most 33 at any time: 10 % extra capacity for the duration in exchange for never dropping below full strength. Pods carry their template hash, so the controller can count old and new from the store at any moment and remembers nothing.
- The scheduler places them and agents start them with the new image. The same path as any new pod. Spread scoring counts old and new replicas together, so the new version does not pile into one zone. If the deployment system prestaged the image, a start takes seconds.
- Each new pod that stays ready lets the controller retire one old pod, which leaves rotation and drains before it stops. A new pod counts only after 10 s of continuous readiness, so one that crashes right after its first probe does not advance the rollout. The old pod is marked terminating, removed from endpoints, and only then sent SIGTERM with up to 30 s to finish in-flight requests; stopping it first would cut requests mid-flight.
- The controller repeats until all 30 pods run the new template, and reports the generation it has observed. Three at a time at ~30 s per batch, 30 pods take about 5 minutes. The status reports observed generation 2 with 30 updated and 30 ready, which is what the deployment system waits for before judging this cluster and moving to the next one.
- If new pods never become ready, the rollout stops after its surge and reports itself stalled; the old pods keep serving. With maxUnavailable 0 a broken release costs 3 surge pods and nothing else. After a 10-minute progress deadline the controller sets progressing to false. Rolling back is writing the old template again, which the same controller converges to; whether to do that automatically is the deployment system’s decision.
- Draining a machine for a kernel upgrade uses budgets too: each pod is evicted only if its service can spare it. The eviction API refuses an eviction that would take payments-api past its disruption budget (at most 1 unavailable), so a drain that meets a service already one short waits instead of making it two. Draining 500 machines a night is just many evictions that obey the same rule, which is why planned maintenance stays invisible.
Deep dives
Desired state in a small consistent store
Why keep the cluster’s state in a Raft-replicated store and make everything else a controller that watches it? Why not have a master tell machines what to run?
An orchestrator is many decisions made by different parts at different speeds: what should run (teams), where (the scheduler), whether it is running (agents), which endpoints are healthy (the endpoints controller). If those parts call each other, every call is a place where a crash, a timeout or a lost message leaves the cluster wrong, with no record of what was intended. At 10,000 machines, about 5 failures a day and 2,000 decisions a second, some message is always being lost.
The insight is to make intent into data: one record per object saying what should be true, in a store that is strongly consistent, versioned and watchable. Each component reads the difference between desired and actual and acts on it, again and again, so a missed event is repaired on the next pass. That only works if the store stays small (state, about 1.5 GB, not logs or metrics), so every replica holds it in memory and every write commits in milliseconds.
- Raft-replicated store of desired and actual state; every component a level-triggered controller watching it chosen
- A master that keeps intent in memory and pushes commands to machines rejected
- Event-driven components reacting to messages on a log (edge-triggered) rejected
- A sharded SQL or wide-column database as the store situational: an archive of history, or a federated control plane far beyond one cluster; the per-cluster store stays one small consensus group
The answer: Five replicas of an etcd-style store: a Raft log, an MVCC key-value map where every write gets a cluster-wide increasing revision, conditional transactions (compare a version, then write), and watches that start from a given revision. It holds only state (300,000 pods, 10,000 machines, 20,000 workloads and their endpoint slices, about 1.5 GB), compacted every few minutes. Every other component is a controller: list, then watch from the list’s revision, keep a local cache, and on every change and every few minutes compare desired with actual for the objects it owns, acting only through conditional writes. Controllers are level-triggered: they never assume they saw every event, and a periodic resync re-checks everything they own, so a lost event costs at most one resync interval. Objects split spec (written by teams) from status (written by controllers), and a generation counter against an observed generation tells anyone whether a change has been acted on. Kubernetes is the best-known public example of this design, with etcd as the store; the Borg paper (EuroSys 2015) describes an earlier shape, a master holding state in memory backed by a Paxos-replicated store.
A controller’s watch connection drops and it misses events. What happens?
It reconnects with the last resource version it processed, and the API server replays every change since then from its window of recent history, so usually nothing is missed. If that version has been compacted away (older than a few minutes), the server answers 410 Gone and the controller relists, gets a snapshot at a new revision and recomputes. Either way it ends up comparing desired with actual on current data. Missing an event is a delay, never a wrong state, and the periodic resync covers events that were received but mishandled.
Why five replicas, not three or seven?
Five tolerates two failures, so one replica can be down for an upgrade while another fails and writes still commit. Three tolerates one, which is thin for the component every write depends on. Seven puts a fourth acknowledgement in every commit for failure tolerance we would rarely use. Place them two, two and one across three zones, so losing any zone still leaves a majority.
It is already replicated. Why not store logs, events and metrics there too?
Every byte in the store is replicated by consensus, held in memory on every replica and sent to watchers. Logs from 300,000 containers are tens of terabytes a day and metrics are millions of points a second; either would multiply the write rate a thousandfold and make every snapshot slow. The store is for the small set of facts that need agreement. Events go to a separate store with a short TTL, logs and metrics to systems built for volume.
Two controllers update the same object at once. How does one not overwrite the other?
Every update is conditional on the version the writer read. The second one fails with a conflict, re-reads and reapplies its change on top. That is optimistic concurrency: cheap because real conflicts are rare, and it means no component needs a lock to edit shared state. Fields with different owners are split (spec from the team, status from controllers) so they do not conflict at all.
How do you know the reconcile loops are keeping up?
Per controller: work-queue depth, time from an object changing to the controller handling it, and objects whose generation is ahead of their observed generation for more than a minute. Cluster-wide: pods pending over a minute, workloads with fewer ready replicas than declared, and watch lag on the API servers. A controller falling behind shows up as a growing gap long before anyone notices missing pods.
Placing 2,000 pods a second
How does the scheduler choose a machine for every pod, at 2,000 pods a second, without two decisions taking the same capacity?
Each decision must find, among 10,000 machines, one that meets the pod’s constraints (CPU, memory, GPUs, zone, machine class, anti-affinity) and is good for the cluster’s future, not just for this pod. Checking and scoring a machine costs about a microsecond, so a full pass is 10 ms; at 2,000 pods a second that is 20 cores and a 10 ms floor on every decision. And decisions depend on each other: two made in parallel from the same view can both pick a machine’s last free slot.
Two observations shape the answer. A good machine is nearly as good as the best one: once a few hundred random feasible machines are scored, the best of them is very close to the best of all 10,000. And collisions are rare when decisions spread over many machines, so it is cheaper to decide in parallel and retry the few that collide than to serialise every decision through a lock.
- Filter, score a sample, bind optimistically: parallel workers on one snapshot, a conditional bind, requeue on conflict chosen
- Score every machine for every pod situational: scarce pools such as a few hundred GPU machines, where a full scan is cheap and placement quality matters most
- One pod at a time under a global lock rejected
- Two-level scheduling: a resource manager offers machines to framework schedulers (Mesos style) situational: many independent frameworks with very different placement needs; the Omega paper compares this model with shared state
The answer: The scheduler keeps an in-memory snapshot of every machine (allocatable, requested, labels, taints) and every pod’s placement, fed by watches. Four workers pop pods from the queue; each walks machines from a random offset, applies the filters, stops after 500 feasible machines (5 % of the cluster) and scores those. The score is a weighted sum: for production, spread across zones and racks first, then prefer fuller machines so whole machines stay free for big pods; for batch, mostly packing. The tasks of one job share an equivalence class so the filter work is reused. The worker marks the capacity as assumed in the snapshot and issues the bind asynchronously: a transaction that checks the pod is unbound and the machine’s allocation version is unchanged, and on conflict releases the assumption and requeues the pod. The Omega paper (EuroSys 2013) argued for this shared-state, optimistic model over a single monolithic scheduler or two-level offers, and Kubernetes’ scheduler scores only a percentage of nodes in large clusters for the same throughput reason.
Stopping at 500 feasible machines can miss the best one. Does that matter?
Rarely for ordinary pods: the best of 500 random feasible machines is within a fraction of a percent of the best overall on a smooth score. It never makes a placement impossible, because the walk continues until it finds 500 or runs out, so a pod that only 40 GPU machines can take still sees all 40; scarce filters cost a full walk, not a wrong answer, and an index of machines by free GPUs makes that walk a lookup. To know the real cost, replay a day of placements offline with full scans and compare stranded capacity.
Bin packing or spreading: which do you score for?
Both, in order. Spreading a service’s replicas across zones, racks and machines is about failure: one rack switch should not take three of four replicas. Packing is about cost: full machines leave whole machines free for large pods and let the capacity planner retire empty ones. So spread is a hard or heavily weighted rule per service, and among machines that satisfy it, prefer the fuller ones. Pure least-loaded placement looks safe but strands capacity: every machine ends up with 6 free cores and a 16-core pod fits nowhere.
The scheduler crashes. What is lost?
Nothing durable. Pods it had assumed but not bound are still unbound in the store, so the standby, elected through a lease, rebuilds its snapshot from the watch cache in a few seconds and carries on. Binds in flight either committed or did not, and the conditional write makes a double bind impossible. The cost is a few seconds without new placements; running pods are untouched.
Pods arrive at 20,000 a second instead of 2,000. What changes?
Decision CPU is easy: more workers. The limit moves to the store and the watch path, because 80,000 writes a second is past what one consensus group should take. So coalesce writes (bind and running in fewer updates), cap the cluster and send the extra work to other clusters, and push very fine-grained batch work into frameworks that lease long-lived worker pods and schedule tasks inside them. At that rate, one pod per task is the wrong unit.
How do you stop a large pod waiting forever while small pods fill every gap?
Greedy packing starves big pods. The queue ages pods so a long wait raises their order within their band; when one reaches the front and fits nowhere, the scheduler nominates a machine and holds it, placing no new lower-priority pods there while its pods finish or are preempted. For gang jobs (all 64 workers or none), a coscheduling step reserves the whole set before binding any, so half a job never sits idle holding capacity.
Priorities, preemption and overcommit
The cluster is full and a production pod needs room. Meanwhile production reserves far more than it uses. How do you handle both?
Teams request capacity for their peak plus a margin, so production reserves about 70 % of the cluster’s CPU and uses about 40 % on average. Leaving the gap idle wastes the equivalent of thousands of machines; filling it with more production pods risks every service hitting its peak together. Batch work is the opposite shape: elastic in time and place, and able to survive being killed.
The key is the difference between a request and a limit. The request is what the scheduler reserves and what the pod is guaranteed under contention; the limit is the most it may use. If production’s requests stay guaranteed and batch’s use of the gap is revocable, idle reservations become capacity without weakening any promise, provided the machine can take the capacity back fast: CPU by throttling, which is instant, and memory by eviction, which is not.
- Priority bands with preemption, plus batch on reclaimed capacity (requests guaranteed, the gap revocable) chosen
- No overcommit: schedule strictly by requests situational: small clusters or dedicated pools for latency-critical services, where predictability is worth the cost
- Overcommit production as well: schedule everything by measured usage rejected
- Let production preempt production by priority rejected
The answer: Three bands, production, batch and best-effort, each with a quota per team per cluster enforced at admission, so the bands mean something. Production is placed by requests against hard allocatable capacity; when nothing fits, the scheduler preempts the fewest, lowest-band, most recently started pods on one machine, never inside the production band, with a 10 s grace, and nominates that machine for the preemptor. Batch is placed against a second figure per machine, batch allocatable: unreserved capacity plus production’s requests minus its p95 usage over the last hour, minus a 10 % margin, recomputed every minute by a reclaim controller from usage metrics. On the machine, production gets high cgroup CPU weights and batch low ones, so CPU contention throttles batch instantly; when free memory falls below a threshold, the agent evicts batch pods itself, before the kernel’s OOM killer picks a victim. For production memory, set the limit equal to the request. The Borg paper (EuroSys 2015) describes the same split: non-production tasks on reclaimed resources, and production tasks not allowed to preempt one another to avoid cascades.
What happens on the machine when production’s usage spikes into what batch borrowed?
CPU: the cgroup weights decide, so production gets most of the contended cores within a scheduler tick and batch slows down; nothing is killed. Memory cannot be squeezed: when free memory crosses a threshold, the agent evicts batch pods, lowest band and largest overuse first, until it is back above it. The threshold leaves enough room that the agent acts before the kernel does, because the kernel’s OOM killer might pick a production process.
Why both a request and a limit? Why not one number?
They answer different questions. The request is reserved and guaranteed, and it is what the scheduler adds up; the limit is a ceiling that protects neighbours. A CPU limit above the request lets a pod use idle cycles on its machine, which is free performance. For memory I set limit equal to request in production, because memory above the request is exactly what gets evicted under pressure, and a service relying on it is relying on luck.
A team marks its batch jobs production so they are never preempted. What stops it?
Quota. Each team has a production quota per cluster sized from what it pays for, and admission rejects pods beyond it; batch quota is larger and cheaper because it is revocable. Charging teams for production requests rather than usage makes over-requesting cost money, which is the only pressure that keeps requests honest over time.
How do you know the reclaim estimate is not too aggressive?
Measure what it risks: batch evictions for memory, and production CPU throttling on machines that host batch, per cluster per day. Rising production throttling means the margin is too thin; rising memory evictions mean the memory estimate is too optimistic. The Borg paper describes comparing more and less aggressive reclamation settings over several weeks and watching out-of-memory events, which is the right method: change it as an experiment, not a guess.
Preemption kills batch tasks that were 90 % done. How do you limit the waste?
Pick victims to lose the least work: prefer the most recently started, and respect each job’s own disruption budget so one job does not lose every task at once. Give a grace period so tasks can checkpoint. And make preemption rarer: keep a small buffer of free production capacity per zone so most production arrivals fit without evicting anything.
Machine failure, partitions and fencing
A machine stops sending heartbeats. When do you reschedule its pods, and how do you make sure a pod that must be unique never runs twice?
From the control plane, a dead machine and a machine behind a network partition look the same: silence. About 5 machines a day really die, and their ~30 pods each should come back quickly. But a partitioned machine is still running its pods, and if they are replaced the cluster briefly runs two copies. For a stateless web replica that is harmless; for a shard leader or a singleton that writes to a database it is split brain, and it corrupts data.
So the question is not only how fast to detect, but what a duplicate would cost. Detection is a lease with a timeout tuned against false alarms; rescheduling waits a further period chosen for what a duplicate would cost; and for pods that must be unique, safety comes from the machine fencing itself when it cannot renew, plus an epoch checked by whatever the pod writes to, not from timing alone.
- Leases renewed by agents, a wait before eviction, a zone-silence brake, self-fencing and epochs for at-most-one pods chosen
- Reschedule as soon as heartbeats are missed rejected
- Gossip-based failure detection among machines (SWIM) situational: very large clusters, as extra evidence for the lifecycle controller; the lease stays the authority
- Never reschedule automatically; page a human situational: pods tied to local data that cannot be recreated elsewhere, such as a database replica on local disks, where that database’s own operator decides
The answer: Each agent renews a 40 s lease every 10 s in a separate small Raft group. The node lifecycle controller measures silence with its own monotonic clock: at 40 s it marks the machine unreachable, taints it and marks its pods not ready, so traffic stops; 60 s later it evicts ordinary pods and controllers replace them, about 135 s from failure to replacements serving. If more than 10 % of a zone goes silent within a minute, it assumes a partition, slows eviction to a few machines a minute and pages. Pods that declare at-most-one get two more protections: the agent kills them itself after 30 s without a successful renewal, counted from when it sent the last one so it always errs early, and the controller releases them only after the lease has expired plus a clock-drift margin. Each replacement gets a higher epoch, which it presents to the database or lock it writes to, so a stale copy that somehow survived is refused at the resource. When a partition heals, the agent stops every pod the store no longer lists for it; the Borg paper describes the same rule for returning machines.
Why 40 seconds? Detecting in 5 would be better.
Faster detection buys capacity back sooner and costs false alarms. A machine under heavy load, a long pause in the agent, or a top-of-rack switch rebooting can silence a machine for 10 to 30 seconds while every pod on it is fine; at 10,000 machines a 5-second timeout would reschedule healthy machines many times a day. Users are protected sooner anyway: proxies and load balancers eject endpoints that fail requests within seconds, so the 40 s is about capacity, not errors.
A whole zone loses the control plane for ten minutes. What happens to its pods?
They keep running and serving: agents do not stop ordinary pods when they lose the control plane, and proxies keep forwarding on their last endpoint sets. Only at-most-one pods stop themselves, because they can no longer prove they are the owner. The lifecycle controller sees a third of the cluster go silent, recognises a partition and does not evict. When the zone reconnects, leases resume and nothing moved; mass eviction would have been the real outage.
Why does the agent count its deadline from when it sent the renewal, not when the reply came?
The controller’s clock starts from the last renewal it received, which can be no earlier than when the agent sent it. Counting from the send time makes the agent’s deadline earlier than the controller’s even if the reply took seconds. Both sides use monotonic clocks for elapsed time because wall clocks jump. The gap between the agent’s 30 s and the controller’s 40 s plus margin covers clock-rate drift and a slow kill.
A singleton’s old copy was paused, not dead, and wakes after the replacement started. What stops it?
Timing cannot, which is why the epoch exists. The replacement started with epoch 19 and wrote it to the database’s fencing column or lock; the old copy still holds 18, so its next conditional write is rejected and it exits. It is the fencing-token pattern from the lock service applied to pods: the lease decides who should run, the epoch at the resource decides whose writes count.
How do you tell a dead machine from a dead agent?
Look from more than one place. If the lease stops but load balancers still see the machine’s pods answering health checks, the agent is broken and the machine is fine, and evicting would replace healthy pods. The lifecycle controller can ask a few peers to probe the machine before deciding, the indirect probe SWIM uses, and the agent runs under a watchdog that restarts it without touching its pods.
Stable addresses and endpoint fan-out
Every service needs an address that never changes while its pods come and go. How do 10,000 machines learn which endpoints are healthy within seconds?
Pods are replaced on every deploy, failure and preemption, each time on a new machine with a new IP. Callers need a name that never changes and an endpoint set that is current. DNS alone is too slow and too coarse: clients cache answers well past their TTL, and DNS cannot say “this endpoint is draining”. And the set has to reach every caller: with a proxy on each of 10,000 machines, every change is sent 10,000 times, up to a million updates a second during busy rollouts.
The trick is to split the stable part from the changing part. The name and a virtual IP are allocated once per service and never change, so DNS caching is harmless. The endpoints behind them are data derived from pod readiness, sharded into small slices so one change rewrites one slice, and distributed as versioned deltas by a tier built for fan-out rather than by the store.
- Name plus virtual IP; endpoint slices from readiness; versioned deltas to per-machine proxies and load balancers chosen
- DNS records per pod with short TTLs rejected
- Central load balancers for all traffic (server-side discovery) situational: traffic from outside the cluster and callers that cannot run a proxy; inside, per-machine proxies are cheaper
- Proxies watch one whole endpoint list per service from the API servers rejected
The answer: A service gets a name and a virtual IP at creation, both fixed for its life. The endpoints controller watches the service’s pods and maintains endpoint slices of at most 100 entries, each with IP, port, zone and a ready flag, writing a slice at most once a second. A discovery service of about 20 stateless replicas watches the slices from the API servers and streams versioned deltas to every node proxy and to the load balancers; a subscriber that reconnects sends its last version and receives only what it missed. Each node proxy programs local forwarding for every virtual IP (eBPF or IPVS), picks ready endpoints and prefers its own zone. When every endpoint of a service is unready, the proxy fails open and uses them anyway, because a readiness bug is likelier than every pod being down. Kubernetes replaced its single per-service Endpoints object with EndpointSlices of up to 100 endpoints for this fan-out cost, and the delta stream is the xDS pattern from the load balancer design.
Why not let every pod resolve DNS and connect directly?
DNS answers are cached by language runtimes and resolvers well past their TTL, so a pod that moved keeps getting traffic for minutes, and DNS cannot mark an endpoint as draining. It also gives no zone or weight information. DNS is still used, but only for the stable part: the name resolves to a virtual IP that never changes, and the proxy behind that IP knows the live endpoints.
The discovery service is down. What happens to traffic?
Nothing at first: every proxy keeps its last endpoint sets and keeps forwarding. New pods get no traffic and removed pods are not removed, so the picture slowly drifts while the outage lasts, and proxies’ own outlier ejection drops endpoints that start failing. When discovery returns, proxies reconnect with their last versions and catch up on deltas. Discovery holds no state of its own, so any replica can serve any proxy.
Mid-rollout, one proxy has the new endpoints and another still has the old ones. Is that a problem?
For a second or two, and it is fine as long as old pods drain before they stop. The rollout removes an old pod from endpoints first, waits for that to propagate, and only then sends SIGTERM with a 30 s grace, so a proxy that is late still reaches a pod that is draining but alive. Discovery tracks which version each subscriber has acknowledged, so one stuck far behind is visible and can be restarted.
How much does each machine hold, and does it grow with the cluster?
Every service’s endpoints: about 200,000 entries at ~100 B, ~20 MB per machine, and it grows with the cluster, one more reason to cap cluster size. In a very large mesh, proxies can subscribe only to the services their local pods are allowed to call, which cuts memory and fan-out by an order of magnitude at the cost of keeping those dependency lists accurate.
How do you measure endpoint freshness?
Stamp each change with the time the pod’s readiness changed and have proxies report the version they applied; the difference is propagation delay, tracked as a distribution with an alert at p99 over 5 s. Also count requests proxies sent to endpoints already marked not ready. Those two numbers are the user-visible half of the whole design.
Scaling the control plane, and cells
Why stop at 10,000 machines per cluster? What breaks first as a cluster grows, and how do you run many of them?
Everything in the control plane grows with the cluster: the store’s size (1.5 GB at 300,000 pods), its write rate (8 k/s at peak), the number of watchers (an agent and a proxy per machine), the scheduler’s snapshot and the blast radius. A bug in one controller or a bad store compaction touches every pod in the cluster. And recovery is worst exactly when it matters: after the API servers restart, thousands of watchers reconnect at once, and if they all relist, the store falls over as it comes back.
The control plane scales its readers by caching and sharding, but one consensus group puts a hard ceiling on writes. Past a size, you do not grow a cluster, you add clusters. A cluster becomes a cell: a fixed, tested maximum size, independent of the others, with every important service spread across several cells in a region.
- Watch caches in the API servers, sharded controllers, a capped size, many clusters per region as cells chosen
- One huge cluster per region on a sharded store situational: a team ready to build and own a sharded store, and workloads that truly need one enormous pool
- A small cluster per team situational: hard isolation requirements such as regulated workloads, accepting lower utilisation
The answer: Readers never touch the store directly. Each of ~10 API server replicas keeps a watch cache: one watch on the store, every object in memory and a few minutes of recent changes, so lists and watches are served from memory and a reconnecting client resumes from its resource version instead of relisting. Lists are paginated, and API servers queue requests per client with fair shares, so a misbehaving controller cannot starve others. Controllers run as leader-elected replicas and the heavy ones are sharded (endpoints by service hash, jobs by namespace hash), each shard owning its objects through a lease. Leases and events live in their own small stores. And the cluster is capped at 10,000 machines and 300,000 pods, the size it is tested at; a region runs 10 to 20 such clusters as cells, the deployment system spreads each service across at least three, and a thin global layer only decides which cells a service runs in. The Borg paper reports a median cell of about 10,000 machines, and Kubernetes publishes lower tested limits (5,000 nodes and 150,000 pods); reaching 10,000 here depends on keeping heartbeats, events and endpoint fan-out off the main store and API path.
All the API servers restart for an upgrade. What protects the store?
They restart one at a time, never together. Each one lists once from the store, starts its single watch and fills its cache; its clients reconnect to the others, or back to it, with their resource versions and resume rather than relist. A client whose version is too old relists from the API server’s memory, paginated, never from the store. The failure to avoid is twenty controllers each listing 1.2 GB of pods from the store at once, about 25 GB of reads in seconds.
Can you shard the scheduler too?
Two ways. By workload: a batch scheduler and a service scheduler both see every machine and commit binds optimistically, the Omega model, with rare conflicts resolved by the conditional bind. Or by machine pool: each shard owns a set of machines, which removes conflicts but means a pod sees only part of the cluster. I would start with one scheduler with several workers, which covers 2,000 a second, and split batch out only when its placement logic diverges.
The store loses its leader. What does the cluster look like for the next ten seconds?
Raft elects a new leader within a couple of seconds of the election timeout. Meanwhile writes fail and clients retry: nothing is placed and no status is recorded, but every pod keeps running, proxies keep forwarding and agents keep restarting crashed containers. API servers reconnect their watches from their last revisions. Users see nothing; deploys see a pause.
How do you upgrade the orchestrator itself across 15 clusters?
One cluster at a time, as a staged rollout like any other: control plane first, then agents machine by machine under the same disruption budgets as a kernel upgrade, with each component compatible with the previous version of the others. A canary cluster running internal workloads goes first and bakes for days. Cells are what make this safe: a bad orchestrator release breaks one cluster, and every important service runs in at least three.
When is a cluster too big even below the limits?
When its blast radius is larger than the business accepts. A cluster holding every replica of a critical service is a single point of failure whatever its size, so spreading services across cells matters more than the exact cap. I would also put workloads with very different needs, such as GPU training and latency-critical serving, in separate clusters, because their scheduling policies and upgrade cadences differ.
The theory behind it
- Kubernetes for system design interviews: The Kubernetes concepts that come up in system design: pods, deployments, services and ingress, autoscaling of pods and nodes, health probes, resource requests and limits, stateful sets, jobs, rolling updates, multi-zone scheduling, and what Kubernetes does not solve for you.
- Consensus, leases and coordination: Leader election, Raft and Paxos at interview depth, ZooKeeper and etcd, distributed locks and why they are fencing tokens, plus how to avoid needing coordination at all.
- Gossip protocols and failure detection: How large clusters know who is alive without a central coordinator: heartbeats and timeouts, phi accrual failure detectors, gossip dissemination and SWIM, membership changes, split brain, and how these show up in databases, caches and service discovery.