Approach: use consistent hashing with virtual nodes (vnodes) proportional to physical node capacity. Compute target vnode counts per physical node by normalizing capacities, then assign/adjust vnodes to match targets while preferring local reassignments that minimize hash-range splits. Migrate in controlled batches using least-loaded windows and rate-limiting to preserve client latency.Pseudocode: compute target vnode countspython
def compute_targets(capacities, total_vnodes):
# capacities: {node: capacity_units}
total = sum(capacities.values())
targets = {n: max(1, int(round(total_vnodes * c / total))) for n, c in capacities.items()}
# adjust for rounding to sum exactly total_vnodes
diff = total_vnodes - sum(targets.values())
if diff > 0:
for n in sorted(capacities, key=capacities.get, reverse=True)[:diff]:
targets[n] += 1
elif diff < 0:
for n in sorted(targets, key=targets.get)[: -diff]:
targets[n] -= 1
return targets
python
def plan_migrations(current_assign, targets, vnode_ring):
# current_assign: {vnode_id: node}
# vnode_ring: ordered list of vnode_ids (hash ring)
needs_add = {n: targets[n] - sum(1 for v,n2 in current_assign.items() if n2==n) for n in targets}
# prefer moving vnodes from nodes with surplus, choosing contiguous vnode ranges to minimize split
migrations = []
surplus_nodes = [n for n,v in needs_add.items() if v<0]
deficit_nodes = [n for n,v in needs_add.items() if v>0]
for d in deficit_nodes:
k = needs_add[d]
for s in surplus_nodes:
while needs_add[s] < 0 and k>0:
# pick vnode on ring assigned to s that causes minimal boundary change:
vnode = pick_best_vnode_to_move(s, current_assign, vnode_ring)
migrations.append((vnode, s, d))
current_assign[vnode] = d
needs_add[s] += 1
k -= 1
if needs_add[s]==0:
surplus_nodes.remove(s)
break
return migrations
python
def execute_batches(migrations, batch_size, max_inflight_bytes, max_ops_per_sec):
batches = chunk(migrations, batch_size)
for batch in batches:
bytes_est = sum(estimate_size(v) for v,_,_ in batch)
if bytes_est > max_inflight_bytes:
shrink = shrink_batch_to_bytes(batch, max_inflight_bytes)
send_batch(shrink)
wait_for_quiesce()
remaining = set(batch) - set(shrink)
migrations.insert(0, list(remaining)) # requeue
continue
rate_limit_send(batch, max_ops_per_sec)
wait_for_replication(batch) # ensure consistent handoff
