Large-scale cluster management at Google with Borg
paper link
https://drive.google.com/file/d/1k0ALs9WIxvln4yTXSEmh1ejRl-H3CzBf/view?usp=sharing
thoughts
this is such a dense read for me. took me roughly 8-9 hours just to wrap my head around different pieces in this paper. mindboggling is an understatement, maybe its mind boggling xd. this paper touches different parts of a distributed system stack: networking, replication, consensus (paxos), sharding, polling, scheduling, virtualization, allocation, static linked binaries, and goes on and on. this paper is still relevant despite there being an updated 2020 borg paper (which we’ll cover in the future), because borg still runs a lot of google’s legendary, high-value workloads. a better way to put it: this paper is a precursor to kubernetes, even though the development timeline is a bit messy, kubernetes was being built in parallel while google kept improving borg. but a lot of the core design choices google made for kubernetes came from the lessons of building and maintaining borg.
borg is a cluster manager. basically, within the same data center (site) there are machines, and google employees have jobs to run on them. all these jobs are managed efficiently by borg. think of borg as the same idea as kubernetes, just slightly older, with a slightly different feature set, and internal to google. borg admits, schedules, starts, restarts, and monitors a fleet of applications at google. it hides the complexity of resource management and failure handling from the users, staying highly efficient and focused on high reliability and availability.
they have a concept called Cell (one cluster of machines) which contains many jobs (from many users, both prod and non-prod) and each job contains many tasks.
One interesting learning from this paper is how unreliable SIGTERM is in practice. at google’s scale, sending a sigterm doesn’t guarantee a graceful shutdown. The key idea is that SIGTERM doesn’t do anything by itself, it’s just a message to the program saying “please shut down.” What happens next is up to the program. There are two cases. If the program doesn’t handle SIGTERM, the operating system applies the default behaviour, which is to terminate the program. In this case, SIGTERM ends up acting almost the same as SIGKILL: the program just dies. If the program has written code to handle SIGTERM, then when the signal arrives, instead of dying, the program runs that code. The program decides what to do, for example save its data, finish current work, and then exit on its own. Sometimes there isn’t even time to deliver the sigterm before a task gets killed outright, so you have to handle that case too, the paper reports sigterm doesn’t get delivered about 20% of the time. graceful shutdown via sigterm alone isn’t something you can fully rely on.
priority: resource allocation follows quota and priority. every job gets a priority number, and higher priority takes over lower priority when there aren’t enough resources to go around (the lower priority job gets scheduled for some other time). eg: A cell is full. Most machines are busy with a large batch job that is reprocessing yesterday’s logs. Suddenly Search traffic spikes, and the Search team wants to add 200 more tasks at production priority. Borg finds machines where it can make room by killing batch tasks, sends them SIGTERM and then SIGKILL, and starts the Search tasks there. The killed batch tasks go back into the pending queue and get rescheduled somewhere else later. Search wins because it has higher priority, which also prevents preemption cascades.
each task gets a unique identifier. even if the machine the task runs on changes, the task gets moved, or the machine gets nuked, we can still get updated info on it through that identifier. the unique identifier also doubles as the dns identifier, so we can easily reach it within the network.
borg is written in c++. it follows a typical master slave setup where the master is called the borgmaster and the slaves are called borglets. the borgmaster uses paxos for consensus to leader-elect among its 5 replicas. borg also keeps creating checkpoints, and using these checkpoints we can recreate the master or get a lagging replica back in sync, basically it’s used to recreate borgmaster’s state. the most interesting part is fauxmaster, which is the first time I’m hearing about this, but it made absolute sense. fauxmaster is an exact clone of the original borgmaster, created with the help of checkpoints at a particular state, but with the borglets stubbed out instead of the real per-machine agents. the master part of fauxmaster is the real production code, loaded with real state from a checkpoint. that’s what makes its test answers trustworthy. beyond testing, it’s also used for replaying past events to debug borgmaster failures, and for capacity planning (how many of these jobs would fit? you can just dry run it in fauxmaster to check).
scheduling breaks into two parts: figuring out which machines are free, then deciding where to place each job on them. the first part is the easy one, jobs already declare what resources they need in borg’s config language, so borg just filters down to machines with enough spare capacity. placement is the hard part, and there are three approaches: two extremes and the hybrid borg actually uses. in every case it comes down to scoring each candidate machine and picking one based on that score. the first extreme is E-PVM. for each candidate machine, borg adds the machine’s current resource usage to the job’s resource needs to get a score, then places the job on whichever machine scores lowest, basically the least loaded one. this spreads jobs thin and leaves headroom everywhere, but it fragments the cluster: instead of some machines empty and some full, you end up with a lot of machines half filled, which wastes capacity overall. the second extreme does the opposite: pack jobs onto the highest-scoring, most-loaded machines, so you keep as many machines as possible empty. the tradeoff is you lose headroom. if a running job suddenly needs more resources, borg has to scramble, killing lower-priority tasks, expanding into the freed space, or moving the job to a different machine or cell. it also creates a bad incentive: an engineer might understate a job’s resource needs to get it scheduled faster, only for the job to need more once it’s actually running. borg’s real approach is a hybrid of the two: it scores machines using both signals, so it can pack tightly enough to be efficient while still keeping enough headroom across the cluster to absorb sudden demand.
borglets are small agents that run on each individual machine and are responsible for communicating with the master. the master sends a heartbeat poll, and the borglet is the one that responds back. if a task crashes, the borglet just restarts it, and if a task needs to be replaced, it kills the old one and reschedules it. so the local, per machine control of a task is handled by the borglet. the problem now is that the master would get overloaded, since every machine’s borglet has to reach out to it with its current state, and that’s the state the master uses to build checkpoints and track heartbeats. what happens instead is the five replicas split up the borglets between themselves and periodically collect their state and heartbeats, taking that work off the leader. so the replicas aren’t just sitting idle for the sake of fault tolerance, they’re doing real work too.
kubernetes lets you attach any key/value labels to a pod, and since a pod can carry several at once, you can slice the same set of pods many ways: by service, tier, release track, version, or any combination. borg’s only built-in grouping was the job, so anything else had to be faked through naming conventions and custom tools.
kubernetes fixed a similar limitation on the networking side too. in borg, all tasks on a machine share one IP (like 10.0.0.7), so the machine tells them apart by port. that makes the machine’s ~65,000 ports one shared pool that borg has to hand out and track, and every program has to be told which port it got. kubernetes instead gives each pod its own IP, so a packet for 10.244.7.2 goes straight to the Maps pod and one for 10.244.7.3 goes to the Search pod, no port lookup needed on the machine’s end. because of network namespaces, each pod also gets its own full set of ~65,000 ports, so Maps and Search can both use 8080 with no sharing, no allocation, and no conflicts. there’s still a limit, but it moves from ports to IPs per machine: a machine given the range 10.244.7.0 to 255 can host roughly 250 pods, which in practice is already more than its cpu and memory can support anyway, and with IPv6 even that ceiling mostly disappears.
final thoughts
overall a super dense paper with a lot of learnings. makes me wonder whether people still put in this kind of effort to build such great software anymore. understanding this paper helped me wrap my head around the reasoning behind a lot of kubernetes’s design choices. the most interesting parts for me in this paper are fauxmaster, their task identifier and how they uniquely identify a task despite the flawed port based design (also the dns mapping), the sigterm failure rates at scale, and recreating state using checkpoints, which is basically what people now talk about with the WAL+S3 trend.