Design a Container Orchestrator

• 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).

• 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.

• 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.

• 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.

1. 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.
   1. A team applies a service spec: 30 replicas of payments-api, 2 CPUs and 4 GB each, spread over three zones, at production priority.
   2. The service controller sees the new workload on its watch and creates 30 pod objects, none of them bound to a machine.
   3. The scheduler takes the pods from its queue, filters the machines and scores the ones that pass.
   4. The scheduler binds each pod with a write conditional on the machine’s allocation version it decided from.
   5. The node agent on that machine sees a pod bound to it, pulls the image and starts the container inside its limits.
   6. The agent runs the readiness probe and reports the pod running, then ready.
   7. 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.
2. 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.
   1. A pod calls payments-api by name, and the name resolves to the service’s virtual IP.
   2. The node proxy on the caller’s machine picks a ready endpoint, preferring its own zone, and forwards the connection.
   3. Traffic from outside reaches the same pods through the load balancers, which subscribe to the endpoints of exposed services.
   4. One payments pod starts failing its readiness probe because its database connection pool is exhausted.
   5. The endpoints controller marks that endpoint not ready in its slice.
   6. Discovery streams a one-endpoint delta to all 10,000 node proxies and to the load balancers.
   7. When the probe passes again the same path puts the pod back; if liveness fails instead, the agent restarts the container in place.
3. 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.
   1. A team submits a job of 100,000 tasks at batch priority, 1 CPU and 2 GB each.
   2. The job controller creates task pods in batches, paced to what the scheduler can place.
   3. The scheduling queue orders by priority, so production pods created during the burst jump ahead of the batch tasks.
   4. Four scheduler workers decide in parallel against the same snapshot, each sampling machines from a different starting point.
   5. Batch tasks may land on capacity that production reserved but is not using, as estimated from recent usage.
   6. Binds commit optimistically; the few that lose a race for a machine’s last free slot are requeued.
   7. The cluster is full; a production pod preempts two batch tasks on one machine to fit.
   8. The job controller sees the evicted tasks and recreates them; they wait at batch priority for capacity.
4. 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.
   1. Every 10 s each node agent renews its machine’s lease; the renewals from m-use1b-04417 stop.
   2. After 40 s without a renewal, the node lifecycle controller marks the machine unreachable and its pods not ready.
   3. Before evicting, it checks how many machines in the zone went silent; a zone-sized silence pauses eviction.
   4. 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.
   5. Controllers see fewer pods than declared and create replacements, which the scheduler places elsewhere.
   6. The partition heals; the returning agent finds its pods gone from the store and stops them.
   7. The machine renews its lease, its taint is lifted and it rejoins empty.
5. 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.
   1. The deployment system writes payments-api’s new image digest into this cluster as one stage of its plan.
   2. The service controller starts 3 surge pods from the new template.
   3. The scheduler places them and agents start them with the new image.
   4. Each new pod that stays ready lets the controller retire one old pod, which leaves rotation and drains before it stops.
   5. The controller repeats until all 30 pods run the new template, and reports the generation it has observed.
   6. If new pods never become ready, the rollout stops after its surge and reports itself stalled; the old pods keep serving.
   7. Draining a machine for a kernel upgrade uses budgets too: each pod is evicted only if its service can spare it.

Ask: 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?

Good answers name: Raft-replicated store of desired and actual state; every component a level-triggered controller watching it, A master that keeps intent in memory and pushes commands to machines, Event-driven components reacting to messages on a log (edge-triggered), A sharded SQL or wide-column database as the store.

Our pick: 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.

1. 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.
2. 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.
3. 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.
4. 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.
5. 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.

Ask: How does the scheduler choose a machine for every pod, at 2,000 pods a second, without two decisions taking the same capacity?

Good answers name: Filter, score a sample, bind optimistically: parallel workers on one snapshot, a conditional bind, requeue on conflict, Score every machine for every pod, One pod at a time under a global lock, Two-level scheduling: a resource manager offers machines to framework schedulers (Mesos style).

Our pick: 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.

1. 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.
2. 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.
3. 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.
4. 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.
5. 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.

Ask: The cluster is full and a production pod needs room. Meanwhile production reserves far more than it uses. How do you handle both?

Good answers name: Priority bands with preemption, plus batch on reclaimed capacity (requests guaranteed, the gap revocable), No overcommit: schedule strictly by requests, Overcommit production as well: schedule everything by measured usage, Let production preempt production by priority.

Our pick: 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.

1. 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.
2. 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.
3. 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.
4. 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.
5. 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.

Ask: 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?

Good answers name: Leases renewed by agents, a wait before eviction, a zone-silence brake, self-fencing and epochs for at-most-one pods, Reschedule as soon as heartbeats are missed, Gossip-based failure detection among machines (SWIM), Never reschedule automatically; page a human.

Our pick: 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.

1. 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.
2. 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.
3. 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.
4. 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.
5. 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.

Ask: 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?

Good answers name: Name plus virtual IP; endpoint slices from readiness; versioned deltas to per-machine proxies and load balancers, DNS records per pod with short TTLs, Central load balancers for all traffic (server-side discovery), Proxies watch one whole endpoint list per service from the API servers.

Our pick: 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.

1. 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.
2. 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.
3. 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.
4. 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.
5. 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.

Ask: Why stop at 10,000 machines per cluster? What breaks first as a cluster grows, and how do you run many of them?

Good answers name: Watch caches in the API servers, sharded controllers, a capped size, many clusters per region as cells, One huge cluster per region on a sharded store, A small cluster per team.

Our pick: 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.

1. 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.
2. 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.
3. 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.
4. 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.
5. 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.