Replication, Partitioning, and Sharding Questions

Requirements & constraints:

• High ingest (millions/sec), petabyte storage, mostly append-only, queries: recent-range scans, aggregated rollups, rare deep-historical retrievals, multi-tenant isolation, cost limits.

Partitioning & granularity:

• Time-first partitioning: partition by fixed time windows (hour/day) combined with secondary shard key (tenant, metric type, or hashed device id). Example: daily partitions for general metrics, hourly for very high-cardinality sources.
• Choose window size by ingest and query locality: smaller windows → faster deletions and compactions but more partitions to manage; larger windows → fewer files but slower cold-query performance. At PB scale, prefer daily partitions with hour sub-shards for hot paths.
• Avoid one-file-per-event: use batching and segment files (WAL/TSDB blocks) sized for efficient IO (e.g., 100MB–1GB).

Retention policies:

• Tiered retention: retain raw high-cardinality recent data (days→weeks) on hot storage; downsample (minute/5m/1h aggregates) for medium term (months); only aggregates or indices for long term (years).
• Automated lifecycle jobs to drop entire time partitions (cheap) rather than deleting rows.
• Policy per-tenant/metric: SLAs drive retention and storage tier.

Compaction:

• Time-ordered compaction: compact small segment files into larger columnar blocks per partition to reduce read amplification and improve compression.
• Multi-tier compaction: frequent compaction for hot partitions, background bulk compaction for cold partitions.
• Use delta-merge or LSM tuning to balance write amplification vs read latency; keep compaction windows aligned with time partitions.

Indexing & storage layout:

• Columnar layout for metrics (timestamp, value, tags) with inverted or tokenized index on high-cardinality tags; maintain time index per block for fast range scans.
• Use bloom filters and per-block min/max for tag/value pruning.
• Avoid global indexes for high-cardinality tags; favor local block indexes plus a light-weight metadata catalog mapping partitions → tag cardinatlity stats.

Query patterns to optimize:

• Recent-range scans: optimize for single-partition reads, cache hot blocks, pre-aggregate rolling windows.
• Rollups/aggregations: maintain materialized aggregates per retention tier and serve via time-bucketed indices.
• Tag-filter heavy queries: push predicates down, use bitmap indexes for moderately-high cardinality, and fallback to streaming scan when cardinality explodes.

Operational considerations & trade-offs:

• Monitor partition count, compaction lag, and index sizes; autoscale compaction workers.
• Trade-offs: smaller partitions simplify retention and compaction but increase metadata overhead; heavier indexing speeds selective reads but raises storage and write cost.
• Test with representative ingest/query replay; model cost vs latency for retention tiers.

This approach balances fast recent queries, cost-effective long-term storage, and manageable operational complexity at petabyte scale.

Hash-based sharding

• What: Use a hash function on the shard key to map records uniformly across shards.
• Typical use-cases: High-throughput OLTP systems needing uniform data distribution (user IDs, session stores, caches).
• Primary trade-offs: Excellent load balancing and simple routing (hash -> shard). Hard to do range queries, re-sharding requires data movement or consistent-hashing techniques; hotspots if key distribution is skewed.
• Poor fit example: Time-series analytics that require scanning contiguous time ranges (e.g., "fetch last 30 days across many users") — hashing scatters contiguous keys across shards.

Range-based sharding

• What: Partition data by contiguous key ranges (e.g., lexicographic or numeric ranges).
• Typical use-cases: Workloads with natural ordering and frequent range queries (time-series, leaderboards, geospatial tiling).
• Primary trade-offs: Efficient range scans and locality, but risk of hotspots as traffic concentrates in popular ranges; requires dynamic split/merge logic to rebalance.
• Poor fit example: Uniform random-access workloads with no locality (e.g., uniformly-distributed random user IDs) — ranges create uneven shard utilization.

Directory-based sharding

• What: Central mapping service (directory) holds explicit mapping from keys (or key ranges) to shards.
• Typical use-cases: Complex routing policies, multi-tenant systems, heterogeneous shards, or when flexible placement decisions needed.
• Primary trade-offs: Extremely flexible and supports custom placement/rehoming, but adds single-point-of-knowledge (needs high-availability), extra lookup latency, and operational complexity.
• Poor fit example: Ultra-low-latency, massively parallel caches where an extra directory lookup per request would be unacceptable (e.g., sub-millisecond CDN edge caches).

Requirements & constraints:

• Minimize load on OLTP (no heavy full-table scans, low latency impact)
• Provide consistent snapshots for analytical correctness
• Keep warehouse near-real-time (seconds–minutes lag)
• Support schema evolution, retries, monitoring

Proposed ETL strategy (high-level):

1. Use change data capture (CDC) + event streaming as primary pipeline:

   • Capture WAL/binlog-based CDC (Debezium for Postgres/MySQL) to produce ordered change events to Kafka.
   • This avoids reading tables and offloads work to the DB’s replication stream, minimizing OLTP impact.

2. Consistent snapshots for initial and periodic full-state:

   • For initial load or periodic full snapshots, use a transactionally consistent export: either logical backup at a consistent WAL position (pg_basebackup + export at LSN) or use the DB’s snapshot isolation (EXPORT with REPEATABLE READ) to get point-in-time dump.
   • Record the WAL/binlog offset of that snapshot and start CDC from the same offset to avoid gaps/duplicates.

3. Streaming apply to warehouse for near-real-time:

   • Consume CDC events from Kafka, transform in a idempotent manner (upsert keys, soft deletes), and write to warehouse using micro-batches (e.g., Kafka Connect + sink connector) or streaming ingestion APIs (BigQuery streaming inserts, Snowflake Snowpipe).
   • Maintain small micro-batch windows (e.g., 30s–5min) to balance latency and throughput.

4. Exactly-once / idempotency & schema:

   • Use event ordering, transaction IDs, and primary keys to make upserts idempotent; persist offsets to ensure at-least-once processing with dedupe.
   • Manage schema evolution via schema registry (Avro/Protobuf) and connector transformations.

5. Backfill and reprocessing:

   • Keep compacted change log and snapshot metadata to support replays/backfills.
   • For late-arriving corrections, support reconciliation jobs comparing aggregates between OLTP snapshot and warehouse.

Operational considerations:

• Monitor lag (CDC lag, Kafka consumer lag), error rates, and row counts; alert on divergence.
• Throttle or schedule heavy snapshot exports during low-traffic windows.
• Test recovery: simulate snapshot + CDC replay to validate consistency.
• Security: encrypt streams, limit DB user permissions for CDC.

Expected outcomes / metrics:

• OLTP CPU/IO impact minimal (no full-table scans), CDC lag < 1–5 minutes, consistency guarantees via snapshot+offset coordination, warehouse kept near-real-time with bounded latency and ability to backfill/repair when mismatches appear.

Overview: Validate an online resharding tool end-to-end before production by exercising correctness, safety, and performance across unit, integration, chaos, and performance tests. Each test type below includes scope, representative test cases, environment/data, and pass/fail criteria.

Unit Tests

• Scope: Individual modules: shard-mapping logic, partition key hash functions, chunk-split/merge algorithms, metadata updates, retry/backoff, and safety gates.
• Cases: deterministic hash outputs, off-by-one boundaries for split points, idempotent metadata writes, failure path handling, validation of precondition checks.
• Environment: CI with mocked storage/metadata APIs.
• Success: 100% critical-path coverage, all assertions pass; no regressions; branch coverage >= 80% for core modules.

Integration Tests

• Scope: End-to-end in a staging cluster using real DB instances (smaller dataset), controller + worker processes, metadata service, client drivers.
• Cases: full reshard operation from planning to cutover; mid-stream reads/writes consistency; client reconnects; schema and index preservation; incremental migration with partial failures.
• Environment: isolated staging that mirrors production networking/versions.
• Success: No data loss (row counts and checksums match before/after), strong/causal consistency guarantees upheld per spec, successful cutover, all client ops succeed with <= production error budget (e.g., <0.1% transient errors).

Chaos Tests

• Scope: Inject infrastructure and process failures to validate safety and recovery.
• Cases: kill worker during chunk move, network partitions between controller and workers, delayed metadata updates, disk full on source/target, clock skew, double-run of same job.
• Environment: staging with chaos tool (Chaos Monkey/Gremlin/Kubernetes podkill).
• Success: Tool either retries/rolls back to a safe state or completes without data loss. Invariants maintained: no dangling references, metadata consistent, idempotency prevents double-apply. Recovery time within SLO (e.g., automated resume within X minutes). All invariants verified via checksums and compare-tool.

Performance & Load Tests

• Scope: Throughput and latency during reshard under load representative of peak traffic.
• Cases: high write throughput during chunk migration, many small writes vs large batched writes, concurrent reshard jobs, varying chunk sizes.
• Metrics: migration throughput (rows/sec), replication lag, client latency P50/P95/P99, CPU/memory IO on source/target, time-to-complete.
• Environment: performance cluster with synthetic load generator and production-like dataset sizes or scaled-down with extrapolation.
• Success: Migration completes within target window (e.g., <= maintenance window), client latency increase within acceptable delta (e.g., <20% P95), replication lag stays below threshold, no data loss, system resources remain within acceptable utilization (e.g., CPU < 80%).

Additional Safety & Release Criteria

• Automated pre-checks: validate cluster topology, free disk, schema compatibility, snapshot/backup taken.
• Dry-run/plan verification: tool produces deterministic plan; human sign-off required for production runs.
• Canary rollout: run on single shard or low-risk tenant first, verify metrics for X hours before full rollout.
• Observability: dashboards, log alerts for retries/errors, audit trail for all metadata changes.
• Rollback: tested automatic rollback or manual safe-stop procedure; rollback validated by integration/chaos tests.

Validation Artifacts

• Test harnesses, checksum/compare utilities, sample datasets, automated test suites in CI, runbooks for failures, post-test reports with metrics and invariant checks.

By enforcing these tests and clear success criteria you ensure correctness, availability, and safety before any production resharding.

Situation: One shard suddenly sees an enormous spike of requests concentrated on a very small set of keys — classic hot-key/DDoS behavior. Below is a defendable, engineer-level plan covering detection, immediate mitigation, and long-term architecture changes (including rate-limiting and sharding tactics).

Detection logic

• Per-key telemetry: maintain sliding-window QPS and request-count histograms (e.g., 1s, 1m, 5m windows).
• Baseline/anomaly rules: trigger when key-QPS > 5× historical median and absolute QPS > S (e.g., 100 req/s) for T seconds (e.g., 30s), or z-score > 4.
• Multi-signal: correlate origin IPs, API keys, geo, user agents; use sample-based sketches (Count-Min) to find heavy hitters cheaply.
• Alerting + automatic trigger flags for mitigation.

Immediate mitigations

• Apply per-key and per-client emergency caps: return 429 for requests over token-bucket threshold.
• Graceful degradation: serve cached responses for hot keys, disable noncritical features, or serve stale/readonly data.
• Blackhole malicious IPs / apply WAF rules and CDN edge blocking for obvious distributed sources.
• Route hot-key traffic to separate queue with rate-limited worker pool (backpressure) to protect main shard.
• Short-lived circuit breaker on the key (e.g., reject >X errors or >Y latency).

Long-term architectural changes

• Tiered caching: move hot-key responses into an in-memory cache (Redis/varnish) with TTLs and write-through to reduce shard load.
• Hot-key isolation: allow moving very hot keys to dedicated shards or “pinned” partitions; support manual and automated hot-key migration.
• Adaptive sharding: use consistent hashing with virtual nodes plus load-aware rebalancing. Add hotspot-aware hashing (split key namespace or use key-suffix randomization for high-traffic keys).
• Rate-limiting infrastructure: distributed token-bucket implementation (local token buckets + remote coordinator or Redis/leaky-bucket) with quota per API-key/account/IP and per-key caps.
• Read replicas and CQRS: offload reads to replicas; adopt eventual consistency where acceptable so reads don’t hit primary.
• Observability & autoscaling: per-shard cost metrics, SLOs, autoscale workers/ingress in response to healthy load signals.
• Abuse controls: quota by account, per-API key throttles, authentication gating for expensive endpoints, and progressive penalties.
• Testing & drills: chaos tests for hot keys, canary mechanisms for new mitigations.

Trade-offs and reasoning

• Immediate caps and caching protect availability quickly but may deny legitimate spikes—use progressive throttling and quick rollback.
• Isolating hot keys increases operational complexity but prevents single-key blast radius.
• Distributed rate limits add latency and state; prefer local buckets with periodic reconciliation for performance.

Example quick rule (practical): if key QPS > max(5×baseline, 200) for 30s → pin key to isolation queue, serve cached results for 60s, return 429 after local token-bucket of 100 req/s, alert ops to decide migration.

This plan balances quick defenses and durable architecture changes to reduce future vulnerability to hot-key DDoS while preserving legitimate traffic where possible.