Sharding at Scale — routing, rebalancing, hotspot

"""shard_router.py — vnode ring + movement report + hot-key guard (~100 line)"""
import bisect
import hashlib
import random
from collections import Counter


def h(s):
    return int(hashlib.md5(str(s).encode()).hexdigest(), 16)


class ShardRouter:
    """consistent-hash ring, প্রতি node vnodesটা virtual point"""

    def __init__(self, nodes, vnodes=64):
        self.vnodes = vnodes
        self.ring = sorted((h(f'{nd}:{v}'), nd)
                           for nd in nodes for v in range(vnodes))
        self.points = [p for p, _ in self.ring]

    def route(self, key):
        i = bisect.bisect(self.points, h(key)) % len(self.ring)
        return self.ring[i][1]

    def map_keys(self, keys):
        return {k: self.route(k) for k in keys}


def movement_report(before, after, label):
    moved = [k for k in before if before[k] != after[k]]
    print(f'\n[{label}] total={len(before)}, moved={len(moved)} '
          f'({100 * len(moved) / len(before):.1f}%)')
    dest = Counter(after[k] for k in moved)
    src = Counter(before[k] for k in moved)
    print('  যারা হারাল :', dict(src.most_common()))
    print('  যারা পেলো  :', dict(dest.most_common()))


def hot_keys(counts, factor=100):
    """mean-এর factor গুণের বেশি QPS নিলে flag"""
    mean = sum(counts.values()) / len(counts)
    return {k: round(n / mean) for k, n in counts.items() if n > factor * mean}


def salt(key, buckets):
    return [f'{key}#{b}' for b in range(buckets)]


# ── 1. routing + topology-change ──
KEYS = [f'user-{i}' for i in range(5_000)]
r4 = ShardRouter(['n1', 'n2', 'n3', 'n4'])
m4 = r4.map_keys(KEYS)
load = Counter(m4.values())
print('4-node load:', dict(sorted(load.items())))

r5 = ShardRouter(['n1', 'n2', 'n3', 'n4', 'n5'])      # n5 join!
m5 = r5.map_keys(KEYS)
movement_report(m4, m5, 'join n5')

r3 = ShardRouter(['n1', 'n3', 'n5'])                  # n2,n4 die
m3 = r3.map_keys(KEYS)
movement_report(m5, m3, 'lose n2+n4')

# ── 2. hot-key detection ──
random.seed(3)
traffic = Counter()
for _ in range(50_000):
    if random.random() < 0.35:
        traffic['celeb-user'] += 1
    else:
        traffic[f'user-{random.randrange(4_000)}'] += 1
print(f'\nhot-keys (>100× mean): {hot_keys(traffic)}')

# ── 3. salting — এক key-কে ছড়িয়ে অনেক key ──
subkeys = salt('celeb-user', 32)
spread = Counter(r5.route(k) for k in subkeys)
print('salted subkeys-এর node-বণ্টন:', dict(sorted(spread.items())))
salted_traffic = Counter()
for k, n in traffic.items():
    if k == 'celeb-user':
        for b, sk in enumerate(subkeys):
            salted_traffic[sk] += max(0, n - b) // len(subkeys)
    else:
        salted_traffic[k] += n
print('after-salt hot-keys:', hot_keys(salted_traffic))

নমুনা output:

4-node load: {'n1': 1014, 'n2': 1354, 'n3': 1379, 'n4': 1253}

[join n5] total=5000, moved=1127 (22.5%)
  যারা হারাল : {'n4': 373, 'n3': 325, 'n2': 243, 'n1': 186}
  যারা পেলো  : {'n5': 1127}

[lose n2+n4] total=5000, moved=1991 (39.8%)
  যারা হারাল : {'n2': 1111, 'n4': 880}
  যারা পেলো  : {'n5': 723, 'n3': 668, 'n1': 600}

hot-keys (>100× mean): {'celeb-user': 1406}
salted subkeys-এর node-বণ্টন: {'n1': 5, 'n2': 7, 'n3': 6, 'n4': 7, 'n5': 7}
after-salt hot-keys: {}

তিনটা পাঠ:

• Join-এ moved 22.5% — theory 20%, বাড়তিটা vnode-quantization (64/ node); vnodes=512-এ চালালে 20%-এর কাছাকাছি নামব।
• Initial-load-ই অসমান (1014–1379) — vnode কম হলে statistical luck; production-এর বড় default-vnode-এর কারণ এটাই।
• Salted subkeys ৫ node-এ ছড়াল — এক key-র আগুন পাঁচ চুলায়; detector নিঃশব্দ। Fan-out-cost read-side-এ — next optimization সেখানেই।

নিজে বাড়ান:

1. Bounded-load variant — overload-node পেলে next-clockwise-node-এ overflow; max-load-ratio track করুন
2. Pre-split simulation — fixed 1024-chunk model-এর movement-cost dynamic split-এর সাথে তুলনা করুন
3. Tail-sim — S shard-এ per-shard latency থেকে aggregate p99; S-বাড়ানো curve-টা আঁকুন