Phase 7: Horizontal Sharding & Distributed Cluster Topology¶
Audience: Principal Distributed Systems Architect & Infrastructure Specialist
Source References:src/mongo/s/,src/mongo/db/s/,src/mongo/db/transaction/
1. Distributed Topology Components¶
A production MongoDB Sharded Cluster consists of three decoupled layers:
┌────────────────────────────────────────────────────────────────────────┐
│ mongos (Stateless Query Routers) │
│ - Client connections terminate here │
│ - Analyzes query predicates against local routing cache │
│ - Dispatches targeted or scatter-gather RPCs to specific shards │
└───────────────────────────────────┬────────────────────────────────────┘
│
┌──────────────────────────┴──────────────────────────┐
▼ ▼
┌────────────────────────────────────┐ ┌───────────────────────────────────┐
│ Config Server Replica Set (CSRS) │ │ Shard Replica Sets (Data Nodes) │
├────────────────────────────────────┤ ├───────────────────────────────────┤
│ - Dedicated 3-node replica set │ │ Shard 1 (Primary + Replicas) │
│ - Stores routing metadata in │ │ - Holds chunk ranges [MinKey, 100)│
│ config.databases & config.chunks │ │ Shard 2 (Primary + Replicas) │
│ - Balancer process runs here │ │ - Holds chunk ranges [100, MaxKey)│
└────────────────────────────────────┘ └───────────────────────────────────┘
2. Shard Key Selection & Mathematical Distribution¶
The shard key determines how documents are partitioned into Chunks (contiguous ranges of data).
2.1 Range-Based Sharding vs. Hashed Sharding¶
Range-Based Sharding (e.g., shardKey: { region: 1, created_at: 1 }):
[Shard 1: ["APAC", 0) -> ["APAC", 5000)] ──► [Shard 2: ["EU", 0) -> ["EU", 5000)]
- Strength: Range queries (e.g. created_at >= 1000) route to a single target shard.
- Fatal Pathology: Monotonically increasing keys (e.g. ObjectId or auto-increment ID)
route 100% of incoming writes to the MAX shard, creating an extreme write hotspot!
Hashed Sharding (e.g., shardKey: { user_id: "hashed" }):
Calculates MD5 hash of user_id (64-bit integer range).
- Strength: Uniform mathematical distribution across all shards.
- Pathology: Range queries become Scatter-Gather (mongos must query EVERY shard in cluster).
2.2 Refined Shard Keys (MongoDB 4.4+)¶
In legacy MongoDB, choosing the wrong shard key required a complete cluster teardown and restore.
MongoDB 4.4+ introduced refineCollectionShardKey, allowing administrators to add a high-cardinality suffix to an existing shard key on a live production cluster:
// Add user_id suffix to eliminate monotonically increasing timestamp hotspot
sh.refineCollectionShardKey("ecommerce.orders", { date: 1, user_id: 1 });
3. Chunk Sizing, Splitting & Zero-Downtime Migration¶
3.1 Chunk Splitting Mechanics¶
- Default chunk size: 64 Megabytes (
config.settings.chunksize). - As documents are inserted into a chunk,
mongodmonitors write metrics. - When an active chunk reaches 64 MB, the primary shard evaluates the midpoint and inserts a split point into
config.chunkson the CSRS. - Crucial: Splitting a chunk is purely a metadata operation on the Config Server; zero physical documents are moved during a split!
3.2 The Balancer & Live Chunk Migration Protocol¶
When the chunk distribution across shards becomes unequal (crossing migration thresholds), the Balancer initiates an online chunk migration from the Donor Shard to the Receiver Shard:
Zero-Downtime Chunk Migration Lifecycle:
1. Initialization:
- Donor Shard opens an internal cursor for documents in range [MinKey, MaxKey).
2. Bulk Document Copy:
- Receiver Shard pulls documents in batches and inserts them into its local WiredTiger tables.
3. Catch-Up Phase (Buffered Writes):
- Any client writes occurring on the Donor Shard during the transfer are logged to an
in-memory Migration Delta Buffer.
- The Donor streams these deltas to the Receiver until lag drops to < 1 second.
4. Critical Section Handshake (Brief Latch):
- Donor Shard pauses client writes to that specific chunk range for < 50 milliseconds.
- Applies the final delta batch to Receiver.
- Updates chunk ownership on the Config Server Replica Set (CSRS).
5. Metadata Epoch Increment:
- CSRS increments the Config Version Epoch.
- mongos routers detect the epoch change and update their routing tables.
6. Asynchronous Range Deletion:
- The Donor Shard asynchronously deletes the migrated documents from its local tables.
3.3 The Jumbo Chunk Crisis¶
A chunk is designated Jumbo when:
- Its physical size exceeds 64 MB (or even tens of gigabytes).
- All documents inside the chunk share the exact same shard key value (e.g., millions of documents with status: "PENDING").
- Because a chunk cannot be split if all documents have identical keys, the Balancer refuses to migrate the chunk.
- Fix: Refine the shard key with a high-cardinality suffix to enable interior split points.