k8s-1m Overview

It can be a challenge to plan IP address space in large clusters. Efficient routing usually means routing a contiguous CIDR to each individual node, which means you’re pre-allocating how many IP addresses (and thus pods) each node can serve. Some nodes that are intended for lots of small workloads end up constrained by the number of available IP addresses before they run out of other hardware resources. A 10.0.0.0/8 has 16 million IPs. For a 1-million node cluster, that leaves just 15 pod IPs per node, which is likely not enough.

The answer is to use IPv6 thoroughly and exclusively. The enormous IPv6 address space means that there’s plenty of room for every pod to have its own globally accessible IP address.

Kubernetes has great support for IPv6, and it requires no code changes to create a fully functioning Kubernetes cluster that uses IPv6 exclusively.

My goal is for each node to have an IPv6 address prefix with a range large enough so each pod on that node can have its own IP out of that range. Plus, of course, at least one for the host itself.

Of course, using IPv6 requires support from your compute vendor. All of the major cloud providers (and even many less-major) support IPv6 to some degree or another. And with public IPv6, it’s trivial to create a single cluster that spans multiple clouds.

I primarily focused on AWS, GCP, and Vultr. (Scoff if you want at Vultr, but they have cheap compute and I am self-bootstrapping this project.) But each one has slightly different twists on its support for IPv6 addressing inside a VM. To give a sense of the range of situations, let me briefly describe each below:

Vultr: Each node gets a /64. The primary IP of the node is the ::1 of that range, and the server automatically receives all traffic for any IP in the full /64 range.

GCP: Each node gets a /96. The primary IP of the node is some random IP within that range. The server must send valid NDP RA packets for the IPs that it wants to receive traffic for.

AWS: Each node gets a /128. You can add a /80 prefix (that comes from a different range) via an API call to an existing NIC or VM. (The ‘Create’ API looks like it can support setting both an IPv6 address and an IPv6 range at creation time, but you’ll get an error). The server must send valid NDP RA packets for the IPs it wants to receive traffic for, and all outgoing packets must use the one MAC address that matches the primary IP.

To satisfy the intersection of these requirements, particularly the requirement about MAC addresses, I create one bridge for all of the pods' interfaces to share. But leave the host interface separate, and enable forwarding to handle traffic between the bridge and the host interface. A host-local IPAM is set to a /96 IPv6 prefix of what I get from the provider. This gives us a full 2^32 IPs per node, plenty of space for pods.

Because these are global public IPv6 addresses, no special routing is necessary. No packet encapsulation or NAT is used. Traffic from each pod is correctly sourced from its true origin pod IP, regardless of destination.

If you only have an IPv6 address, then you can only reach other IPv6 addresses on the internet. Anything that is IPv4-only isn’t directly accessible.

Most services I used in this project worked fine: Ubuntu packages, PyPi packages, the docker.io registry. The main exception was GitHub. Github.com remains stubbornly IPv4-only. Tsk tsk.

Many AWS services have dual-stack endpoints but notably for this project Elastic Container Registry (ECR) does not. Tsk tsk to them as well.

For IPv6 devices to reach IPv4 hosts, most cloud providers offer some sort of NAT64 gateway. You can also roll your own gateway on a Linux VM. I over-engineered this a bit with a custom WireGuard server. All VMs connect via WireGuard to this server and use it as an IPv4 gateway.

High-level, I hand-waved over this problem and did not use network policies between workloads.

1 million nodes would have 1 million separate IPv6 prefixes, which is far too many individual entries for any firewall solution to support. Security-minded folks: clutch your pearls when I say that I do not use extensive firewalling to prevent access into the cluster from the Internet. I do use firewall rules to limit to a select few number of ports that I know need to be reached, but otherwise we must rely on other techniques to safeguard unauthorized inbound access to servers and pods.

Thorough use of TLS covers most use cases for this project. The enormous size of the IPv6 address space also makes scanning impractical. Cilium, kube-proxy, or other network plugins could also limit which pods can reach which pods, but at significant cost of additional watches on the control plane.

If you’re using one single vendor for all of your nodes, it may be plausible that all nodes still get ipv6 ranges out of 1 or a few larger spaces, a count low enough that could be reasonably installed as firewall rules.

Both kube-apiservers and etcd support both HTTP/2 and gRPC. Many individual requests and streams are multiplexed over a single TCP connection. Kubernetes sets a default HTTP/2 limit of 100 concurrent requests (or technically streams) per TCP connection. (HTTP/2 can support far more than that, but as you add more streams you run into performance problems like head-of-line blocking). So each kubelet needs at least 1 connection to the kube-apiserver control plane. And you can expect 1 more connection for kube-proxy, or any similar CNI like Cilium or Calico. With 1M nodes, that means each kube-apiserver is supporting at least 2 million TCP connections. With 8 kube-apiservers, each server would be supporting 250K connections to kubelets.

Linux itself can support this number of connections with some light tuning. And of course make sure you have allowed yourself enough file descriptors. Nevertheless it may be more than your network provider can support. Azure, for example, documents that it can support a maximum of 500k inbound and 500k outbound connections per VM. GCP and AWS do not publish limits, but there are limits in any system to both the total number of concurrent connections as well as the rate of new connections being made.

First a quick overview about the ways you work with state in Kubernetes. Any number of clients interact with kube-apiservers, which then in turn interact with etcd.

kube-apiservers are stateless. etcd is the persistent store for all of Kubernetes resources. All CRUD operations you send to a kube-apiserver are actually persisted by etcd.

kube-apiservers have seven common verbs for state:

etcd has four:

kube-apiserver create, update, patch, and delete all result in an etcd put operation. (A delete is just a put with a null value). etcd doesn’t support any partial updates of values, only putting the entire value. So all operations that involve modifying a resource result in a new etcd put of the entire resource contents.

kube-apiserver watch can, but often doesn’t, result in an etcd watch. More on that below.

Kubelets interact with the kube-apiserver primarily through two resource types:

The Lease is critical: if it isn’t refreshed in time, the NodeController marks the node as NotReady. By default, each kubelet updates its Lease every 10 seconds. At a scale of 1 million nodes, that alone translates to 100K writes per second just to keep the nodes "alive."

Adding in the constant churn of other resources, the system needs to sustain on the order of many hundreds of thousands of writes per second, plus a significant volume of reads.

For kube-apiserver, this is manageable. It’s stateless, so QPS can be scaled out simply by running more replicas. If one instance can’t handle the load, more can be added, and traffic will spread across them.

For etcd, things are different. Etcd is stateful, which makes scaling QPS much harder.

Using the etcd-benchmark tool, I measured about 50K writes/sec out of a single etcd instance backed by NVMe storage. Importantly, adding replicas doesn’t help. Write throughput actually drops with more members since each write must be coordinated across a quorum of replicas to maintain consistency. So with the typical 3-replica setup, effective write QPS is even lower than the benchmarked 50K/s. That’s nowhere near what’s needed to support a 1M-node cluster.

At first glance, 50K QPS seems surprisingly low given modern hardware capabilities. A single NVMe drive can do over 1M 4K writes per second, and a single DDR5 DIMM can push 10x more than that. So why is etcd is far behind raw hardware limits?

The answer lies in etcd’s interface and guarantees. For one thing, etcd is ensuring that all writes are durable to disk. For every put or delete call, etcd ensures the change is written to disk via fsync before acknowledging success. This helps ensure that there is never any data loss if the host crashes or loses power. But that durability drastically reduces the number of IOPS that a modern NVMe drive can support.

Plus, etcd has a pretty broad interface surface area:

Implementing all of those interfaces can make for complex software. Beyond simple puts and deletes, etcd must support transactions, maintain multi-versioned history, and enforce Raft-based consensus across replicas.

These features are what give Kubernetes its consistency and reliability, but they also impose strict constraints on performance. Intuitively, strong consistency means more serialization: operations can’t always be parallelized freely. Writes often need to follow a carefully ordered path through Raft, the WAL, and compaction, ensuring that every replica agrees on state before acknowledging success.

The result is raw hardware capable of millions of writes per second, but etcd delivering orders of magnitude less due to the interfaces and guarantees it must uphold.

But do we need all of these things?

I built a new program called mem_etcd that implements the etcd interface but with the simplifications described above. Written in Rust, it provides fully correct semantics for the etcd APIs that Kubernetes depends on.

mem_etcd maintains two main data structures:

Each value also stores the non-compacted revision history for that key. This design makes writes to existing keys O(1), while writes to new keys and range queries are O(log n) (where n is the number of resources of that Kind). Range queries also require additional linear work up to the query’s limit.

Despite its name, mem_etcd can provide durability by writing a write-ahead log (WAL) to disk. Each prefix of /registry/[$APIGROUP/]$APIKIND/[$NAMESPACE/] is written to its own separate file. By default, files are written in buffered mode, so put calls can complete before the data is durably written to disk. This behavior can be changed with a CLI flag that enables fsync, forcing all writes to be flushed to disk before the put completes. You can also configure some prefixes to not be written to disk at all.

These tests were run on a pair of c4d-standard-192-lssd instances, with one VM running mem_etcd and the other running the client benchmark. In these results, you can easily observe how badly enabling fsync negatively impacts throughput and latency. Note that the baseline comparison of etcd is a single replica of etcd v3.5.16 running on a tmpfs (ram-based) disk. This should be an optimal environment for etcd as there is no actual disk involved and fsync, while still being a syscall, is otherwise a no-op. mem_etcd is storing its WAL on a local NVMe, what GCE calls Titanium SSD. Though the instance type has 16 local disks, only 1 is used for this test.

etcd-lease-flood is a custom benchmark designed to simulate the dominant type of load in a large Kubernetes cluster. Each client creates 100 Lease objects directly in etcd, using the same protobuf encoding as Kubernetes. For each Lease, the client repeatedly issues put updates in a tight loop, attempting to update the Lease as quickly as possible.

There are several different types of watches and each has different performance characteristics. Let’s unpack them.

https://kubernetes.io/docs/reference/using-api/api-concepts/#semantics-for-watch has some useful details about how the kube-apiserver handles the parameters of your watch:

Let’s re-format the important points into a table:

So a watch is often preceded by a list. The list provides a snapshot-in-time of a set of resources, marked with a revision number. Then you start a watch from that revision number, which will then stream to you all of the changes that have occurred since that revision.

When resourceVersion is set, a watch against the kube-apiserver does not create a new watch against etcd. At startup time, a kube-apiserver creates watch streams against etcd for each of the well-known standard resources. Any time a client creates a watch, the kube-apiserver handles that stream itself based on the one etcd watch stream it maintains. So while client watches can be expensive for kube-apiservers, it adds no additional load to etcd. You can horizontally scale more kube-apiservers.

Furthermore, watches are not really that bad for etcd. A watch has a beginning and an end range, and those ranges fit within the same prefixes as Range queries. With each Put we need to do a log(n) lookup in the list of watches to find watches that could match that key. But there are far far fewer watches than objects. The n is small and is done asynchronously after the write is committed anyway, so it does not affect the request time to complete a write.

Watches do create network amplification. For each write into etcd, there may be N corresponding watches for that object. That results in a lot of outbound network traffic from etcd. The kube-apiservers are on the receiving end of these watches. kube-apiservers are consolidating their own watches, but etcd is still sending a copy of the data to each kube-apiserver. While adding more kube-apiserver replicas does help with many Kubernetes scalability problems, each replica does put additional pressure on the etcd NIC. The network throughput of etcd is the most immediate hardware bottleneck of large-scale Kubernetes clusters. However, these demands are limited to just between etcd and the kube-apiservers. In a single datacenter with modern hardware there’s still plenty of potential interconnect that could be established amongst these servers.

Let’s revisit our 1M kubelet Lease requirement. Kubelet is issuing an Update (aka Replace, aka PUT) of its Lease resource every 10 seconds.

This is an old Lease:

This is the body of calling Update() when renewing that Lease:

Note the renewTime has been updated to something 10 seconds later. (renewTime is in fact always set to 40 seconds in the future, so we can tolerate some amount of failed or slow lease updates).

The other key field is the resourceVersion. When a client sends an Update() to a kube-apiserver, it includes the same resourceVersion from the previous version of the resource it’s updating. This is for safety to ensure that no other client has updated the resource in-between. Every time a resource is updated on the server, the server assigns the new resource a monotonically-increasing new resourceVersion. An Update operation must include a resourceVersion that indicates what the old version of the resource it thinks it’s replacing. That way we’re not accidentally overwriting some other change that has happened in-between.

You’d think that kube-apiserver could simply convert this Update operation into a Txn-Put operation in etcd, passing through this command in a straightforward and stateless way. Unfortunately kube-apiserver’s Update implementation also always needs to obtain the entire Old version of this resource. There’s a few reasons for this:

So to keep Update calls performant, kube-apiserver will maintain Watch caches of most commonly-used resources. When an Update occurs, it’ll pull the old version from its local watch cache. If for some reason the old version is not in the watch cache, then kube-apiserver will first issue a Range to etcd to get the old resource before calling Txn-Put.

Having to do two synchronous calls to etcd for each Update would double our QPS needs and latency, so it’s much better if we can rely on an up-to-date watch cache.

However this introduces a new requirement and constraint for 1M nodes: kube-apiservers must be able to “Watch” at a rate of at least 100K events/sec.

In my testing this is where things get a little tight.

It’s akin to the classic scatter/gather design of a distributed search system. Think of each node as a document in the corpus, and each pod as a search query. The query is fanned out to many shards, each responsible for a fraction of the documents. Each shard selects its top candidate(s) and sends them back to a central gatherer to ultimately identify the overall top result.

Generic scatter-gather design pattern

The key difference is that in search, documents are read-only and thus queries can be evaluated in parallel without conflicts. In scheduling, however, executing a decision actually modifies the documents (i.e. allocates node resources). If two pods are scheduled in parallel to the same node, one may succeed while the other must fail due to insufficient resources.

Nevertheless, for a large cluster it’s reasonable to design for “optimistic concurrency”. That is, presume that multiple pods can be scheduled at the same time without conflicts. We still check to see if a conflict arises before committing. And if a conflict occurs, we still do the correct thing by “rolling back” the other pod, e.g. it has to be re-scheduled. But the chance and resulting impact of this happening is low - low enough that you get much higher throughput by running in parallel and absorbing the wasted effort if it does happen.

So my initial architecture idea of the distributed scheduler:

Scatter-gather as a Kubernetes pod scheduler pattern

The Relay starts a watch on unscheduled pods against the kube-apiserver. As streams of pods come in, the Relay forwards them to different schedulers.
Each Scheduler is responsible for doing filtering and scoring against its own subset of the overall nodes, then sending back to Relay the top-winning node and score.

The Relay then aggregates these scores, picks the overall winner, sends a true/false back to the Scheduler, and that true/false dictates whether the scheduler should actually bind the pod to that node.

The scheduler here is thus a slightly modified version of the upstream kube-scheduler. It has a custom gRPC server endpoint for receiving the new pod. It has custom code to know which of the overall nodes it is responsible for. And it has a custom Permit extension point for sending the proposed node back to the Relay. The Permit extension point runs after the nodes are filtered and scored, but before the pod is bound. Permit extensions return ‘true’ or ‘false’ to approve whether or not the pod should be scheduled on the specified node.

This is the basic design and it works pretty well. It doesn’t quite work up to 1M node scale - we’ll talk about that next - but it delivers a much more scalable scheduler solution than what exists today, while preserving all of the nuanced complex battle-tested logic of the current system.

Today’s scheduler is effectively O(n x p), where n is the number of nodes and p the number of pods. That complexity becomes untenable as n grows. The sharded approach helps counteract the scaling problem: if you have n nodes, then you can shard the work across r replicas where r is some factor of n, turning that large factor back into something more tractable.

I should mention there’s one fairly large exception, and that’s pod evictions. Pod eviction can occur when there’s a new pod to schedule, but there’s not enough available resources currently on the cluster to schedule that pod. When this occurs, the scheduler does a scan across all pods currently running in the cluster, trying to identify a set of lower-priority pods that, if they were to be killed, would leave enough space for this new pod. To be fair, I didn’t implement this. You could squint at the current approach and imagine how we could also distribute the work of eviction calculation, but I didn’t do it.

On my hardware, a single scheduler was able to filter and score a pod against 1K nodes in about 1ms. So we could do 1K pods on 1K nodes in 1s. Remember the goal was 1M pods on 1M nodes in 60s. Recall that the overall work is O(n x p) (each pod has to be evaluated against each node), so going from 1K pods and nodes to 1M pods & 1M nodes is not a factor of 1K more work, but 1K*1K, or 1 million times more work. Even allowing ourselves 60s instead of 1s, we’re going to need a lot more schedulers.

Created a 100K node cluster and then timed how long it takes to schedule 100K pods. The pods have no nodeSelector or affinity.

Each scheduler ran on a dedicated c4d-standard-32; that is 32 AMD Turin cores and 128GB of DDR5 RAM. Experiments where dist-schedulers that had more than 1 replica also had 1 dedicated dist-scheduler-relay VM.

Each dist-scheduler replica is configured to run 30 separate internal schedulers, each with a parallelism setting of ‘2’.

The default-scheduler is kube-scheduler 1.32.3 with no modifications.

One puzzling result is how much better a 1x dist-scheduler performed than the default-scheduler. Adjusting the parallelism setting had no impact on the performance nor the CPU-seconds, which seemed to peak at about 20 (leaving 12 cores free).

Note that adding replicas to dist-scheduler did result in a more-or-less linear time improvement. In other words, doubling the number of dist-scheduler replicas results in a halving of the time to completion. This trend continues to the 256x replica/1M pod scale as we’ll see in the next section.

In the below graphs, the green line annotation indicates when the first pod was created and the red line indicates when the millionth pod was scheduled.