Introducing Shard Manager
Shard Manager is a new standalone, general-purpose sharding service. It provides near-optimal load-aware placement. It reduced CPU utilization on the hottest Matching host by 38%. And it provides a centralized view of shard ownership and placement.
For the past year the Cadence team has worked on Shard Manager, a new standalone, general-purpose sharding service (earlier covered in 1 Day, 2 Engineers, 6 Lines of Code). We have now onboarded Cadence's Matching service to Shard Manager across all of Uber. Matching routes workflow and activity tasks to workers. With near-optimal load-aware placement, CPU utilization on the hottest Matching host dropped by 38%.
To use it in your own service, start with the Shard Manager documentation.
Why we built Shard Manager
Cadence distributes workflows across its hosts, grouping them into shards, and distributes the shards evenly using consistent hashing, via the long deprecated ringpop library. So long as each host knows the full set of hosts, it's able to determine exactly what it should be working on and how to route requests to peers.
Both this distribution and coordination became recurring problems:
- Hot sharding. Not all shards are equal, and we constantly saw massive skews in CPU utilization. Evenly distributing the shards doesn't evenly distribute the work.
- Distributed membership. Gossip protocols eventually converge, but discrepancies cause conflicts, slow down the system, and are challenging to debug. Each host is individually responsible for assignment and there's no authoritative answer.
- Little control. The only way to change the distribution is changing the membership, adding or removing hosts. A host that is degraded but still up keeps all its shards.
We saw this when a Matching host ran hot and our auto-scaler added more hosts. Each new host took a set of shards from 100 places in the ring (thanks to virtual nodes). However, it took them based on their hash rather than their load, so it didn't necessarily take the hot shards off the hot host. Adding capacity didn't fix the hotspot.
We realized a central service could solve these problems and built Shard Manager. It owns shard placement, balances shards by their observed load, gives one authoritative answer to who owns a shard, and lets operators drain a host or stop a single shard.
Load balancing and results in Uber's production environments
In Matching, the shards are task lists, the queues that workers poll for tasks, and each one lives on exactly one host. The figure below shows a month of data from one of Uber's production Matching clusters. The period covers the time before and after the onboarding.
The max, average and min CPU across hosts at each moment. The lines turn from orange to blue when load-aware placement is switched on. Hover for values.
We see that the average barely moves, as total traffic stays about the same. However, the spread decreases, leading to much better utilization. Under the hash ring, the hottest host regularly ran at twice the average utilization, while others sat close to idle. Immediately after we switched on load-aware placement, the hottest host dropped from 2.3× to 1.4× the average.
As we size clusters for the hottest host, the same traffic can now run on about 40% fewer hosts.
How close is this to the best possible?
To check how good the placement is, we took a snapshot of the load of every shard in each environment we've rolled out to. We then asked a constraint solver for the best possible placement of exactly those shards on exactly those hosts.
We measure balance as load spread, the standard deviation of host load divided by the average host load. At 0, every host carries the same load. Across environments, Shard Manager's spread was within 5% of the solver's in 23 of 24 snapshots, and within 1% in 14.
The logic underneath the balancing is explained further down.
Controlling shard placement
Because all assignments are now centrally managed, operators can inspect shard
ownership, drain a host, or isolate a shard with a single smctl command.
Who owns this shard? With the hash ring, the answer depended on which host you asked. Now you ask Shard Manager:
$ smctl --namespace cadence-matching shard inspect \
--shard-key orders-tasklist
{ "Owner": "6c1f0e2a-…", … }
A host is misbehaving. Drain it, and its shards move to the other hosts:
$ smctl --namespace cadence-matching executor drain \
--hostname matching-host-3 --reason "debugging high latency"
A shard keeps crashing its host. Moving it just crashes the next host, and then the next, until the whole cluster is down. Now we just drain it and investigate:
$ smctl --namespace cadence-matching shard drain \
--shard-key orders-tasklist
How it works
Architecture
The results above are possible due to the centralized architecture of the Shard Manager. Below we see an architecture diagram of the full system with the Matching integration.
Matching hosts heartbeat to Shard Manager, which assigns shards and pushes the routing map to Frontend.
The system consists of three roles, the Shard Manager, the executors and the spectators. The Shard Manager is the central service, backed by etcd, responsible for deciding placements and communicating these placements to the rest of the system. The Shard Manager service has multiple instances, where one of them is elected as a leader for each namespace. The leader is responsible for load balancing and changing assignments when hosts join and leave.
Executors are responsible for processing (executing) the work. In this case the executors are Cadence's Matching service. They heartbeat to the Shard Manager, reporting liveness and load. The service defines what load means (see the integration guide). For Matching, the load measure is the number of tasks the task list processes.
Spectators are services that interact with executors, so they need to know which executor owns a shard. In the diagram we see Cadence Frontend in the role of spectator, as it routes each request to the Matching host that owns the task list. The spectators establish a gRPC stream to the Shard Manager, which then pushes new shard assignments to the spectators as the assignments change.
The same service can be both spectator and executor. The receiving instance can then forward the requests to the correct owner, allowing the callers of the service to not know about the sharding of the service.
For more information, see Shard Manager architecture.
How the load balancing works
With the executors reporting the load of each shard they own, we can do load-aware balancing. Below is an example with six hosts. It starts from a hash ring's placement, where every host has the same number of shards. To simplify the demo, each step is a single move.
Step through it with Next. A shard turns blue when it moves and fades back to gray as its cooldown runs out.
1: Hash ring placement. Hosts 4 and 6 are too hot.
2: Host 4 → Host 1. Move the shard that evens the two out the most.
3: Host 4 → Host 3. Host 3 is now the coldest, so the move goes there.
4: Host 6 → Host 4.
5: Host 6 → Host 2.
6: Every host is inside the bands, so we stop moving shards.
The Shard Manager smooths the loads so small variations don't trigger moves. Every few seconds it looks for an imbalance:
- It compares each host with the average host using hysteresis bands (upper and lower load thresholds). A host above the upper band is too hot, and a host below the lower band has room to take more load. Anything in between is left alone.
- Each cycle moves a small batch of shards, by default up to 1% of all shards. Each move goes from a hot host to the coldest one, picking the shard that brings the two closest together, excluding shards that would make the cold host hot.
- Shards that have recently moved have to cool down before they can be moved again, which prevents thrashing.
Hosts aren't perfectly balanced when we stop, and that's on purpose. Every move has a cost, and past that point more moves would only churn shards between hosts.
For more details, see Andreas Holt and Theis Mathiassen's master's thesis Extending the Load Balancing for Cadence. It describes the algorithm, compares it with near-optimal placement, and explores extensions.
Future work
- History. We onboarded Matching first. Integrating it is much simpler than History, the Cadence service that manages workflow state and execution, and the cost of moving a task list between hosts is low. History will be onboarded next. We expect a larger effect on CPU utilization there, and with it the ability to handle the same load with fewer instances.
- Make etcd optional. Shard Manager stores its state in etcd today. We plan to remove that dependency, so Shard Manager will be able to run completely in memory.
- Beyond Cadence. Shard Manager is a generic service, not specific to Cadence. Any service that needs to shard its work can use it. The service integration guide shows how, and we'd love to hear from teams who want to use it.