The Monotonic Shard Key and the Frankfurt Requirement
A logistics platform shards a 2.1 TB events collection across four shards. The shard key is {createdAt: 1}. Inserts arrive at 12,000/sec and almost always carry the current minute. Shard0's data disk sits at 96% utilization while the other three hover near 15%. Portal queries filter on {tenantId, createdAt: {$gte, $lt}} and take 900 ms at p99 even though tenantId is indexed. Overnight the balancer moved 380 chunks, and the on-call engineer saw three jumbo chunks flagged in sh.status(). A new EU data-residency rule requires that all events belonging to EU tenants be stored on shards physically located in Frankfurt. The cluster runs MongoDB 5.0 with three mongos routers.
1. Diagnose the hot shard and the jumbo chunks. Which measurements would you collect before proposing a fix?
Read the worked answer
The write hotspot is explained by the shard key itself. createdAt increases monotonically, so every new document lands in the highest chunk range, and that range lives on one shard. Splits occur, but splitting only creates new chunks at the top of the range; the balancer then moves them hours later in bulk. The result is a sawtooth: one shard absorbing all live insert traffic, then sudden migration storms. That also explains the 380 chunks moved overnight - the balancer was chasing a permanently moving hotspot, not balancing a stable distribution.
Collect, concretely:
- sh.status() and db.events.getShardDistribution() for chunk counts and sizes per shard, plus per-shard disk utilization and iostat to confirm the write pressure is on shard0.
- config.chunks for the chunk ranges and which are flagged jumbo; a jumbo chunk means the balancer cannot move it, so it is stranded where it was created.
- config.changelog for migration volume and timings to correlate the overnight spike with application latency.
- Per-shard serverStatus().wiredTiger and opcounters to show insert distribution directly.
Jumbo chunks in an events collection usually come from low cardinality within a chunk: if many documents share the same createdAt value at the chunk's granularity, no split point exists inside the 64 MB range, so it can never split or move. A monotonic key with second- or millisecond-level duplication produces exactly that.
Also measure the query side: run explain("executionStats") for the portal query and check totalKeysExamined versus nReturned and how many shards were targeted. A 900 ms p99 with tenantId indexed on a scatter-gather query is consistent with fan-out across four shards, not with a missing index.
2. Recommend a shard key for the observed workload, and explain how you would migrate to it on MongoDB 5.0.
Read the worked answer
Requirements: equality on tenantId plus a range on createdAt for the portal query, and 12,000 inserts/sec distributed across shards.
A compound key of {tenantId: 1, createdAt: 1} satisfies the query shape: tenantId equality means the router targets a single shard, and createdAt range within that tenant is served by an index on the shard key prefix, so the query becomes a targeted range scan. Write distribution is much better than now because inserts spread across the many tenants' current-time ranges instead of a single global maximum.
The risk to name explicitly is a whale tenant. If one tenant generates most of the traffic, its current-time chunks will all land on one shard and you have reintroduced the hotspot at tenant scope. Options: add a third, high-cardinality field to break the whale, accepting that range queries within a tenant then fan out; or isolate whales into their own zone.
Migration on 5.0: use sh.reshardCollection("db.events", {tenantId: 1, createdAt: 1}). It rewrites the collection in place from the application's perspective, but it is I/O- and network-heavy on 2.1 TB - schedule it, watch disk headroom on every shard, and expect the cluster to be busier than usual. Do not mistake refineCollectionShardKey for a fix: it can only append a suffix field to the existing key, and it does not redistribute existing chunks, so the hotspot survives. Pre-5.0 the only path would be a new sharded collection plus dual writes and backfill.
3. How would you implement the Frankfurt residency requirement, and what does it constrain about the shard key?
Read the worked answer
Zone sharding is the mechanism, and it imposes one hard constraint: the field that determines residency must be a prefix of the shard key. Fortunately the proposed key {tenantId: 1, createdAt: 1} starts with tenantId, so tenants can be mapped to zones. If the key had been {createdAt: 1} or hashed on _id, no zone assignment could guarantee that all of a tenant's documents stay in Frankfurt, and the residency requirement would be unimplementable without a reshard.
Implementation steps:
- Tag the Frankfurt shards: sh.addShardToZone("shardFRA01", "EU") for each, and similarly tag non-EU shards if you want EU tenants strictly excluded from them.
- Define ranges on the sharded collection that cover exactly the EU tenants. If tenant IDs are UUIDs or otherwise not range-friendly, allocate an explicit prefix or numeric block per tenant and define one range per block; do not leave gaps that let a chunk land outside the zone.
- Call sh.updateZoneKeyRange("logistics.events", {tenantId: minKeyValueForEU}, {tenantId: maxKeyValueForEU}, "EU"). Ranges must not overlap across zones.
- Ensure the balancer is running and let it migrate matching chunks to the tagged shards, then verify with sh.status() that every EU range has only EU-tagged shards as owners. Use moveChunk deliberately if a range must move immediately.
Operational cautions: the balancer moving terabytes into Frankfurt will saturate the WAN link and both ends' disks - throttle it or schedule it. Add monitoring that alerts if any chunk in an EU range is owned by a non-EU shard, because a mis-typed zone range fails silently otherwise. Finally, verify that the config server metadata and any backups of EU data also stay within the permitted jurisdiction.