diff --git a/Orchestrator/docker-compose.yml b/Orchestrator/docker-compose.yml index d3bca18..80d7de1 100644 --- a/Orchestrator/docker-compose.yml +++ b/Orchestrator/docker-compose.yml @@ -100,6 +100,8 @@ services: container_name: grafana environment: - GF_SECURITY_ADMIN_PASSWORD=dbaas + - GF_AUTH_ANONYMOUS_ENABLED=true + - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin volumes: - ./grafana/provisioning:/etc/grafana/provisioning ports: diff --git a/Orchestrator/grafana/provisioning/dashboards/dashboard.yml b/Orchestrator/grafana/provisioning/dashboards/dashboard.yml index 3ebb1e3..0beb28b 100644 --- a/Orchestrator/grafana/provisioning/dashboards/dashboard.yml +++ b/Orchestrator/grafana/provisioning/dashboards/dashboard.yml @@ -3,9 +3,10 @@ apiVersion: 1 providers: - name: 'dbaas' orgId: 1 - folder: '' + folder: 'DBaaS' type: file disableDeletion: false updateIntervalSeconds: 10 + allowUiUpdates: false options: - path: /etc/grafana/provisioning/dashboards/dbaas.json + path: /etc/grafana/provisioning/dashboards diff --git a/Orchestrator/grafana/provisioning/dashboards/dbaas.json b/Orchestrator/grafana/provisioning/dashboards/dbaas.json index 09c5311..9657bac 100644 --- a/Orchestrator/grafana/provisioning/dashboards/dbaas.json +++ b/Orchestrator/grafana/provisioning/dashboards/dbaas.json @@ -1,113 +1,391 @@ { - "title": "DBaaS Overview", + "title": "DBaaS Platform", "uid": "dbaas-overview", + "description": "Distributed KV DBaaS β€” control plane, data plane, replication, and failure detection", "schemaVersion": 38, - "version": 1, + "version": 2, "refresh": "5s", + "tags": ["dbaas", "distributed-systems"], + "time": { "from": "now-15m", "to": "now" }, + "timepicker": {}, + "annotations": { + "list": [ + { + "builtIn": 1, + "datasource": "-- Grafana --", + "enable": true, + "hide": true, + "iconColor": "rgba(0, 211, 255, 1)", + "name": "Annotations & Alerts", + "type": "dashboard" + } + ] + }, "panels": [ + + { "id": 100, "type": "row", "title": "🟒 Cluster Health", "gridPos": {"x":0,"y":0,"w":24,"h":1}, "collapsed": false }, + { - "id": 1, - "type": "timeseries", - "title": "Write Rate (ops/s)", - "gridPos": {"x": 0, "y": 0, "w": 12, "h": 8}, - "targets": [ - { - "datasource": "Prometheus", - "expr": "rate(dbaas_writes_total[1m])", - "legendFormat": "writes/s" + "id": 1, "type": "stat", "title": "Live Workers", + "description": "Total worker containers currently registered in ZooKeeper", + "gridPos": {"x":0,"y":1,"w":3,"h":4}, + "fieldConfig": { + "defaults": { + "color": {"mode": "thresholds"}, + "thresholds": {"steps": [{"color":"red","value":0},{"color":"yellow","value":1},{"color":"green","value":2}]}, + "unit": "short" + } + }, + "options": {"colorMode":"background","graphMode":"none","reduceOptions":{"calcs":["lastNotNull"]}}, + "targets": [{"datasource":"Prometheus","expr":"sum(dbaas_worker_count)","legendFormat":"total"}] + }, + { + "id": 2, "type": "stat", "title": "Leader Workers", + "description": "Should always be exactly 1. Non-1 values indicate an election in progress or a split-brain.", + "gridPos": {"x":3,"y":1,"w":3,"h":4}, + "fieldConfig": { + "defaults": { + "color": {"mode": "thresholds"}, + "thresholds": {"steps": [{"color":"red","value":0},{"color":"green","value":1},{"color":"red","value":2}]}, + "unit": "short" + } + }, + "options": {"colorMode":"background","graphMode":"none","reduceOptions":{"calcs":["lastNotNull"]}}, + "targets": [{"datasource":"Prometheus","expr":"sum(dbaas_worker_count{role=\"leader\"})","legendFormat":"leaders"}] + }, + { + "id": 3, "type": "stat", "title": "Follower Workers", + "description": "Followers serving reads. Autoscaler adjusts this count based on RPS.", + "gridPos": {"x":6,"y":1,"w":3,"h":4}, + "fieldConfig": { + "defaults": { + "color": {"mode": "thresholds"}, + "thresholds": {"steps": [{"color":"yellow","value":0},{"color":"green","value":1}]}, + "unit": "short" + } + }, + "options": {"colorMode":"background","graphMode":"none","reduceOptions":{"calcs":["lastNotNull"]}}, + "targets": [{"datasource":"Prometheus","expr":"sum(dbaas_worker_count{role=\"follower\"})","legendFormat":"followers"}] + }, + { + "id": 4, "type": "stat", "title": "Leader Elections", + "description": "Cumulative leader elections since startup. A spike here indicates repeated failures.", + "gridPos": {"x":9,"y":1,"w":3,"h":4}, + "fieldConfig": { + "defaults": { + "color": {"mode": "thresholds"}, + "thresholds": {"steps": [{"color":"green","value":0},{"color":"yellow","value":2},{"color":"red","value":5}]}, + "unit": "short" + } + }, + "options": {"colorMode":"background","graphMode":"area","reduceOptions":{"calcs":["lastNotNull"]}}, + "targets": [{"datasource":"Prometheus","expr":"dbaas_leader_elections_total","legendFormat":"elections"}] + }, + { + "id": 5, "type": "stat", "title": "Election Rate (5m)", + "description": "Elections per minute in the last 5 minutes. Sustained rate > 0 means cluster instability.", + "gridPos": {"x":12,"y":1,"w":3,"h":4}, + "fieldConfig": { + "defaults": { + "color": {"mode": "thresholds"}, + "thresholds": {"steps": [{"color":"green","value":0},{"color":"orange","value":0.1},{"color":"red","value":0.5}]}, + "unit": "reqpm", + "decimals": 2 + } + }, + "options": {"colorMode":"background","graphMode":"none","reduceOptions":{"calcs":["lastNotNull"]}}, + "targets": [{"datasource":"Prometheus","expr":"rate(dbaas_leader_elections_total[5m]) * 60","legendFormat":"elections/min"}] + }, + { + "id": 6, "type": "stat", "title": "Autoscale Events", + "description": "Cumulative scale-up + scale-down decisions made by the autoscaler.", + "gridPos": {"x":15,"y":1,"w":3,"h":4}, + "fieldConfig": { + "defaults": { + "color": {"mode": "palette-classic"}, + "unit": "short" } + }, + "options": {"colorMode":"value","graphMode":"area","reduceOptions":{"calcs":["lastNotNull"]}}, + "targets": [ + {"datasource":"Prometheus","expr":"sum(dbaas_autoscale_events_total{direction=\"up\"})","legendFormat":"scale up"}, + {"datasource":"Prometheus","expr":"sum(dbaas_autoscale_events_total{direction=\"down\"})","legendFormat":"scale down"} ] }, { - "id": 2, - "type": "timeseries", - "title": "Read Rate (ops/s)", - "gridPos": {"x": 12, "y": 0, "w": 12, "h": 8}, + "id": 7, "type": "stat", "title": "Kafka WAL Offset", + "description": "Current leader's Kafka write offset. Monotonically increasing β€” shows durability progress.", + "gridPos": {"x":18,"y":1,"w":3,"h":4}, + "fieldConfig": { + "defaults": { + "color": {"mode": "fixed", "fixedColor": "blue"}, + "unit": "short" + } + }, + "options": {"colorMode":"value","graphMode":"area","reduceOptions":{"calcs":["lastNotNull"]}}, + "targets": [{"datasource":"Prometheus","expr":"dbaas_kafka_offset","legendFormat":"WAL offset"}] + }, + { + "id": 8, "type": "stat", "title": "Snapshot Offset", + "description": "Last Kafka offset captured in the leader's snapshot. Followers at this offset skip WAL replay entirely.", + "gridPos": {"x":21,"y":1,"w":3,"h":4}, + "fieldConfig": { + "defaults": { + "color": {"mode": "fixed", "fixedColor": "purple"}, + "unit": "short" + } + }, + "options": {"colorMode":"value","graphMode":"area","reduceOptions":{"calcs":["lastNotNull"]}}, + "targets": [{"datasource":"Prometheus","expr":"dbaas_snapshot_offset","legendFormat":"snapshot offset"}] + }, + + { "id": 200, "type": "row", "title": "πŸ“ˆ Throughput & Latency", "gridPos": {"x":0,"y":5,"w":24,"h":1}, "collapsed": false }, + + { + "id": 10, "type": "timeseries", "title": "Write Throughput (ops/s)", + "description": "Write operations per second reaching the leader. Driven by the workload generator.", + "gridPos": {"x":0,"y":6,"w":12,"h":8}, + "fieldConfig": { + "defaults": { + "color": {"mode": "palette-classic"}, + "custom": {"lineWidth": 2, "fillOpacity": 10}, + "unit": "ops" + } + }, + "options": {"tooltip":{"mode":"multi"},"legend":{"displayMode":"table","placement":"bottom","calcs":["mean","max","lastNotNull"]}}, "targets": [ - { - "datasource": "Prometheus", - "expr": "rate(dbaas_reads_total[1m])", - "legendFormat": "{{follower_id}}" + {"datasource":"Prometheus","expr":"rate(dbaas_writes_total[30s])","legendFormat":"writes/s (30s avg)"}, + {"datasource":"Prometheus","expr":"rate(dbaas_writes_total[1m])","legendFormat":"writes/s (1m avg)"} + ] + }, + { + "id": 11, "type": "timeseries", "title": "Read Throughput by Follower (ops/s)", + "description": "Reads per second per follower. A balanced load indicates Nginx round-robin is working correctly. Uneven distribution can appear when followers restart.", + "gridPos": {"x":12,"y":6,"w":12,"h":8}, + "fieldConfig": { + "defaults": { + "color": {"mode": "palette-classic"}, + "custom": {"lineWidth": 2, "fillOpacity": 8}, + "unit": "ops" } + }, + "options": {"tooltip":{"mode":"multi"},"legend":{"displayMode":"table","placement":"bottom","calcs":["mean","max","lastNotNull"]}}, + "targets": [ + {"datasource":"Prometheus","expr":"rate(dbaas_reads_total[30s])","legendFormat":"{{follower_id}}"} ] }, { - "id": 3, - "type": "timeseries", - "title": "Write Latency p99 (ms)", - "gridPos": {"x": 0, "y": 8, "w": 12, "h": 8}, + "id": 12, "type": "timeseries", "title": "Write Latency (ms) β€” p50 / p95 / p99", + "description": "End-to-end write latency: LevelDB apply + RabbitMQ publish + Kafka commit. p99 spikes here often correspond to Kafka acks under load.", + "gridPos": {"x":0,"y":14,"w":12,"h":8}, + "fieldConfig": { + "defaults": { + "color": {"mode": "palette-classic"}, + "custom": {"lineWidth": 2, "fillOpacity": 5}, + "unit": "ms" + } + }, + "options": {"tooltip":{"mode":"multi"},"legend":{"displayMode":"table","placement":"bottom","calcs":["mean","max","lastNotNull"]}}, "targets": [ - { - "datasource": "Prometheus", - "expr": "histogram_quantile(0.99, rate(dbaas_write_latency_ms_bucket[1m]))", - "legendFormat": "p99" + {"datasource":"Prometheus","expr":"histogram_quantile(0.50, rate(dbaas_write_latency_ms_bucket[1m]))","legendFormat":"p50"}, + {"datasource":"Prometheus","expr":"histogram_quantile(0.95, rate(dbaas_write_latency_ms_bucket[1m]))","legendFormat":"p95"}, + {"datasource":"Prometheus","expr":"histogram_quantile(0.99, rate(dbaas_write_latency_ms_bucket[1m]))","legendFormat":"p99"} + ] + }, + { + "id": 13, "type": "timeseries", "title": "Read Latency (ms) β€” p50 / p95 / p99", + "description": "Read latency per follower (LevelDB get). Should be much lower than write latency. High p99 can indicate a slow disk or a follower catching up from its WAL replay.", + "gridPos": {"x":12,"y":14,"w":12,"h":8}, + "fieldConfig": { + "defaults": { + "color": {"mode": "palette-classic"}, + "custom": {"lineWidth": 2, "fillOpacity": 5}, + "unit": "ms" } + }, + "options": {"tooltip":{"mode":"multi"},"legend":{"displayMode":"table","placement":"bottom","calcs":["mean","max","lastNotNull"]}}, + "targets": [ + {"datasource":"Prometheus","expr":"histogram_quantile(0.50, rate(dbaas_read_latency_ms_bucket[1m]))","legendFormat":"p50"}, + {"datasource":"Prometheus","expr":"histogram_quantile(0.95, rate(dbaas_read_latency_ms_bucket[1m]))","legendFormat":"p95"}, + {"datasource":"Prometheus","expr":"histogram_quantile(0.99, rate(dbaas_read_latency_ms_bucket[1m]))","legendFormat":"p99"} ] }, + + { "id": 300, "type": "row", "title": "πŸ”„ Replication Health", "gridPos": {"x":0,"y":22,"w":24,"h":1}, "collapsed": false }, + { - "id": 4, - "type": "timeseries", - "title": "Replica Lag (ms)", - "gridPos": {"x": 12, "y": 8, "w": 12, "h": 8}, + "id": 20, "type": "timeseries", "title": "Replica Lag per Follower (ms)", + "description": "Time between when the leader published a write to RabbitMQ and when the follower received it. A persistent lag > 100ms means the follower is falling behind β€” autoscaler may need to trim. A sudden spike usually indicates a follower restart.", + "gridPos": {"x":0,"y":23,"w":16,"h":8}, + "fieldConfig": { + "defaults": { + "color": {"mode": "palette-classic"}, + "custom": {"lineWidth": 2, "fillOpacity": 12}, + "unit": "ms", + "thresholds": { + "mode": "absolute", + "steps": [{"color":"green","value":0},{"color":"orange","value":100},{"color":"red","value":500}] + } + } + }, + "options": {"tooltip":{"mode":"multi"},"legend":{"displayMode":"table","placement":"bottom","calcs":["mean","max","lastNotNull"]}}, "targets": [ - { - "datasource": "Prometheus", - "expr": "dbaas_replica_lag_ms", - "legendFormat": "{{follower_id}}" + {"datasource":"Prometheus","expr":"dbaas_replica_lag_ms","legendFormat":"{{follower_id}}"} + ] + }, + { + "id": 21, "type": "stat", "title": "Max Replica Lag (current)", + "description": "Worst-case replication lag across all followers right now.", + "gridPos": {"x":16,"y":23,"w":4,"h":4}, + "fieldConfig": { + "defaults": { + "color": {"mode": "thresholds"}, + "thresholds": {"steps": [{"color":"green","value":0},{"color":"orange","value":100},{"color":"red","value":500}]}, + "unit": "ms" + } + }, + "options": {"colorMode":"background","graphMode":"none","reduceOptions":{"calcs":["lastNotNull"]}}, + "targets": [{"datasource":"Prometheus","expr":"max(dbaas_replica_lag_ms)","legendFormat":"max lag"}] + }, + { + "id": 22, "type": "stat", "title": "WAL vs Snapshot Gap", + "description": "Difference between the current WAL offset and the last snapshot offset. Measures how many writes a recovering follower must replay from Kafka. A large gap means slower follower startup.", + "gridPos": {"x":16,"y":27,"w":4,"h":4}, + "fieldConfig": { + "defaults": { + "color": {"mode": "thresholds"}, + "thresholds": {"steps": [{"color":"green","value":0},{"color":"yellow","value":500},{"color":"red","value":2000}]}, + "unit": "short" + } + }, + "options": {"colorMode":"background","graphMode":"none","reduceOptions":{"calcs":["lastNotNull"]}}, + "targets": [{"datasource":"Prometheus","expr":"dbaas_kafka_offset - dbaas_snapshot_offset","legendFormat":"offset gap"}] + }, + { + "id": 23, "type": "timeseries", "title": "RabbitMQ Sync Messages Consumed by Follower", + "description": "Cumulative sync messages (writes applied via RabbitMQ fanout) per follower. A follower that stops increasing here has lost its RabbitMQ connection.", + "gridPos": {"x":20,"y":23,"w":4,"h":8}, + "fieldConfig": { + "defaults": { + "color": {"mode": "palette-classic"}, + "custom": {"lineWidth": 2}, + "unit": "short" } + }, + "options": {"tooltip":{"mode":"multi"},"legend":{"displayMode":"list","placement":"bottom"}}, + "targets": [ + {"datasource":"Prometheus","expr":"rate(dbaas_sync_messages_total[1m])","legendFormat":"{{instance}} msgs/s"} ] }, + + { "id": 400, "type": "row", "title": "βš™οΈ Autoscaler", "gridPos": {"x":0,"y":31,"w":24,"h":1}, "collapsed": false }, + { - "id": 5, - "type": "stat", - "title": "Live Workers", - "gridPos": {"x": 0, "y": 16, "w": 6, "h": 4}, + "id": 30, "type": "timeseries", "title": "Worker Count Over Time", + "description": "How the cluster scales in response to load. Leader count should stay flat at 1; follower count tracks RPS demand. Sawtooth pattern = workload cycles.", + "gridPos": {"x":0,"y":32,"w":16,"h":8}, + "fieldConfig": { + "defaults": { + "color": {"mode": "palette-classic"}, + "custom": {"lineWidth": 2, "fillOpacity": 15, "stacking": {"mode": "normal"}}, + "unit": "short", + "min": 0 + } + }, + "options": {"tooltip":{"mode":"multi"},"legend":{"displayMode":"table","placement":"bottom","calcs":["lastNotNull","max"]}}, "targets": [ - { - "datasource": "Prometheus", - "expr": "sum(dbaas_worker_count)", - "legendFormat": "total" + {"datasource":"Prometheus","expr":"sum(dbaas_worker_count{role=\"leader\"})","legendFormat":"leader"}, + {"datasource":"Prometheus","expr":"sum(dbaas_worker_count{role=\"follower\"})","legendFormat":"followers"} + ] + }, + { + "id": 31, "type": "timeseries", "title": "Scale Events", + "description": "Autoscaler scale-up and scale-down decisions over time. Each spike corresponds to a follower being spawned or terminated.", + "gridPos": {"x":16,"y":32,"w":8,"h":8}, + "fieldConfig": { + "defaults": { + "color": {"mode": "palette-classic"}, + "custom": {"lineWidth": 2, "fillOpacity": 10}, + "unit": "short" } + }, + "options": {"tooltip":{"mode":"multi"},"legend":{"displayMode":"list","placement":"bottom"}}, + "targets": [ + {"datasource":"Prometheus","expr":"rate(dbaas_autoscale_events_total{direction=\"up\"}[1m]) * 60","legendFormat":"scale-up events/min"}, + {"datasource":"Prometheus","expr":"rate(dbaas_autoscale_events_total{direction=\"down\"}[1m]) * 60","legendFormat":"scale-down events/min"} ] }, + + { "id": 500, "type": "row", "title": "🚨 Failure Detection & Recovery", "gridPos": {"x":0,"y":40,"w":24,"h":1}, "collapsed": false }, + { - "id": 6, - "type": "stat", - "title": "Leader Elections", - "gridPos": {"x": 6, "y": 16, "w": 6, "h": 4}, + "id": 40, "type": "timeseries", "title": "HTTP Error Rate (Nginx β†’ Workers)", + "description": "Rate of 5xx responses from orchestrator's FastAPI (as seen by Prometheus instrumentator). A spike here often precedes a failure detection event.", + "gridPos": {"x":0,"y":41,"w":12,"h":8}, + "fieldConfig": { + "defaults": { + "color": {"mode": "thresholds"}, + "custom": {"lineWidth": 2, "fillOpacity": 15}, + "unit": "reqps", + "thresholds": { + "steps": [{"color":"green","value":0},{"color":"orange","value":0.01},{"color":"red","value":0.1}] + } + } + }, + "options": {"tooltip":{"mode":"multi"},"legend":{"displayMode":"list","placement":"bottom"}}, "targets": [ { "datasource": "Prometheus", - "expr": "dbaas_leader_elections_total", - "legendFormat": "elections" + "expr": "sum(rate(http_requests_total{status=~\"5..\",job=\"workers\"}[1m]))", + "legendFormat": "worker 5xx/s" + }, + { + "datasource": "Prometheus", + "expr": "sum(rate(http_requests_total{status=~\"5..\",job=\"orchestrator\"}[1m]))", + "legendFormat": "orchestrator 5xx/s" } ] }, { - "id": 7, - "type": "stat", - "title": "Kafka Write Offset", - "gridPos": {"x": 12, "y": 16, "w": 6, "h": 4}, + "id": 41, "type": "timeseries", "title": "Request Success Rate", + "description": "Percentage of successful (non-5xx) requests across leader and followers. Should stay at 100% under normal operation. Dips indicate upstream failures detected by Nginx.", + "gridPos": {"x":12,"y":41,"w":12,"h":8}, + "fieldConfig": { + "defaults": { + "color": {"mode": "thresholds"}, + "custom": {"lineWidth": 2, "fillOpacity": 10}, + "unit": "percentunit", + "min": 0, "max": 1, + "thresholds": { + "steps": [{"color":"red","value":0},{"color":"orange","value":0.95},{"color":"green","value":0.99}] + } + } + }, + "options": {"tooltip":{"mode":"multi"},"legend":{"displayMode":"list","placement":"bottom"}}, "targets": [ { "datasource": "Prometheus", - "expr": "dbaas_kafka_offset", - "legendFormat": "offset" + "expr": "sum(rate(http_requests_total{status!~\"5..\",job=\"workers\"}[1m])) / sum(rate(http_requests_total{job=\"workers\"}[1m]))", + "legendFormat": "worker success rate" } ] }, { - "id": 8, - "type": "stat", - "title": "Auto-scale Events", - "gridPos": {"x": 18, "y": 16, "w": 6, "h": 4}, - "targets": [ - { - "datasource": "Prometheus", - "expr": "sum(dbaas_autoscale_events_total)", - "legendFormat": "events" + "id": 42, "type": "timeseries", "title": "Leader Election Timeline", + "description": "Value advances every time a leader election completes. Useful for correlating with latency spikes and follower count changes β€” each step here is a failover event.", + "gridPos": {"x":0,"y":49,"w":24,"h":6}, + "fieldConfig": { + "defaults": { + "color": {"mode": "fixed", "fixedColor": "orange"}, + "custom": {"lineWidth": 2, "fillOpacity": 20, "drawStyle": "bars"}, + "unit": "short" } + }, + "options": {"tooltip":{"mode":"single"},"legend":{"displayMode":"list","placement":"bottom"}}, + "targets": [ + {"datasource":"Prometheus","expr":"increase(dbaas_leader_elections_total[1m])","legendFormat":"elections in last 1m"} ] } + ] } diff --git a/Orchestrator/orch/failure_detection.py b/Orchestrator/orch/failure_detection.py index e935442..da601cd 100644 --- a/Orchestrator/orch/failure_detection.py +++ b/Orchestrator/orch/failure_detection.py @@ -46,7 +46,6 @@ def delete_worker_znode(cont_id: str) -> None: def _resolve_cont_id(host: str) -> str: - # host can be IP or hostname (cont_id) host_clean = host.split(":")[0].strip() return state.host_to_cont_id.get(host_clean, "") @@ -55,7 +54,14 @@ def _should_skip_failover(w: state.WorkerInfo, request_uri: str) -> str | None: """Returns a reason string if failover should be skipped, else None.""" now = time.monotonic() is_write = request_uri.startswith("/write") - + + # Critical: never kill a worker that hasn't marked itself ready in ZK. + # The followerβ†’leader transition window (follower.py killed, leader.py not yet + # listening) produces genuine 502s that must not trigger a kill. + # ZK ephemeral-node expiry handles truly stuck / never-ready workers. + if not w.ready: + return "not_ready" + # Common startup grace for all workers if (now - w.first_seen_ts) < WORKER_STARTUP_GRACE_SEC: return "startup_grace" @@ -65,10 +71,26 @@ def _should_skip_failover(w: state.WorkerInfo, request_uri: str) -> str | None: return "not_leader_for_write" if w.leader_since_ts and (now - w.leader_since_ts) < LEADER_PROMOTION_GRACE_SEC: return "leader_promotion_grace" - + return None +def _probe_worker(host: str) -> bool: + host_clean = host.split(":")[0].strip() + if host_clean in {"leader_upstream", "follower_upstream", "127.0.0.1", "localhost"}: + return True # Placeholder β€” not a real worker + + url = f"http://{host_clean}:8000/health" + for _ in range(2): + try: + if requests.get(url, timeout=1.5).status_code == 200: + return True + except Exception: + pass + time.sleep(0.3) + return False + + def _probe_and_handle_failure(target_host: str, request_uri: str, upstream: str, upstream_status: str) -> None: cont_id = _resolve_cont_id(target_host) probe_key = cont_id or target_host @@ -79,11 +101,11 @@ def _probe_and_handle_failure(target_host: str, request_uri: str, upstream: str, return _failure_probe_last_ts[probe_key] = time.monotonic() - # 1. Verification Probe + # 1. Verification probe β€” if the host responds healthy, do nothing if _probe_worker(target_host): return - # 2. Guard checks + # 2. Guard checks using WorkerInfo cache w = state.workers.get(cont_id) if not w: logger.warning("Failure confirmed for unknown worker", extra={"host": target_host}) @@ -94,9 +116,9 @@ def _probe_and_handle_failure(target_host: str, request_uri: str, upstream: str, logger.info("Skipping fast-failover", extra={"cont_id": cont_id, "reason": skip_reason, "uri": request_uri}) return - # 3. Action + # 3. Confirm and act logger.info("Confirming worker failure, triggering failover", extra={"cont_id": cont_id, "uri": request_uri}) - emit(state.kafka_broker, "orchestrator", "worker_failure_confirmed", + emit(state.kafka_broker, "orchestrator", "worker_failure_confirmed", cont_id=cont_id, upstream=upstream, status=upstream_status, uri=request_uri) delete_worker_znode(cont_id) @@ -106,45 +128,28 @@ def _probe_and_handle_failure(target_host: str, request_uri: str, upstream: str, logger.warning("Failed to kill failed worker", extra={"cont_id": cont_id, "error": str(e)}) -def _probe_worker(host: str) -> bool: - host_clean = host.split(":")[0].strip() - if host_clean in {"leader_upstream", "follower_upstream", "127.0.0.1", "localhost"}: - return True # Placeholder - - url = f"http://{host_clean}:8000/health" - for _ in range(2): - try: - if requests.get(url, timeout=1.5).status_code == 200: - return True - except Exception: - pass - time.sleep(0.3) - return False - - def handle_upstream_signal(kind: str, upstream: str, upstream_status: str, request_uri: str) -> Response: """Entry point for Nginx failure/observation reports.""" hosts = [h.strip() for h in upstream.split(",") if h.strip()] statuses = [s.strip() for s in upstream_status.split(",") if s.strip()] for host, status in zip_longest(hosts, statuses, fillvalue=""): - if not host: continue - - # We care if status is 5xx or 000 (nginx failed to connect) - is_fail = not status.isdigit() or int(status) >= 500 or int(status) == 0 + if not host: + continue + is_fail = not status.isdigit() or int(status) >= 500 or status == "0" if is_fail: threading.Thread( target=_probe_and_handle_failure, args=(host, request_uri, upstream, upstream_status), - daemon=True + daemon=True, ).start() if kind == "failure": - # Return a 502/503 to nginx if this was a mandatory failure report code = 502 if statuses: - last_status = statuses[-1] - if last_status.isdigit(): code = int(last_status) + last = statuses[-1] + if last.isdigit(): + code = int(last) return JSONResponse(content={"status": "accepted"}, status_code=code) - - return {"status": "signal_accepted"} + + return Response(content="signal received", status_code=200) diff --git a/Orchestrator/orch/nginx_config.py b/Orchestrator/orch/nginx_config.py index bcfc4df..7b898b6 100644 --- a/Orchestrator/orch/nginx_config.py +++ b/Orchestrator/orch/nginx_config.py @@ -9,7 +9,7 @@ def _proxy_config_version(leader_addr: str | None, follower_addrs: list[str]) -> version_payload = { "leader": leader_addr or "", "followers": sorted(follower_addrs), - "template_v": "4", # Increment when template changes + "template_v": "5", # Increment when template changes } digest = hashlib.sha256(json.dumps(version_payload, sort_keys=True).encode("utf-8")).hexdigest() return digest[:12] @@ -30,21 +30,6 @@ def render_proxy_config(leader_addr: str | None, follower_addrs: list[str]) -> t {followers_block} }} -map $upstream_status $had_failed_upstream_attempt {{ - default 0; - "~(^|,)[ ]*(4[0-9]{{2}}|5[0-9]{{2}}|000)[ ]*(,|$)" 1; -}} - -map $upstream_addr $had_multiple_upstream_attempts {{ - default 0; - "~," 1; -}} - -map "$had_failed_upstream_attempt:$had_multiple_upstream_attempts" $report_upstream_outcome {{ - default 1; - "0:0" 0; -}} - server {{ listen 80; resolver 127.0.0.11 ipv6=off; @@ -53,6 +38,10 @@ def render_proxy_config(leader_addr: str | None, follower_addrs: list[str]) -> t location /write {{ proxy_pass http://leader_upstream; proxy_set_header X-Request-ID $request_id; + proxy_connect_timeout 500ms; + proxy_read_timeout 1s; + proxy_next_upstream error timeout invalid_header http_502 http_504; + proxy_next_upstream_tries 1; proxy_intercept_errors on; error_page 500 502 503 504 = @upstream_failure; }} @@ -61,45 +50,24 @@ def render_proxy_config(leader_addr: str | None, follower_addrs: list[str]) -> t proxy_pass http://follower_upstream; proxy_set_header X-Request-ID $request_id; proxy_connect_timeout 250ms; - proxy_read_timeout 2s; - proxy_next_upstream error timeout invalid_header http_500 http_502 http_503 http_504; - proxy_next_upstream_tries 3; - proxy_next_upstream_timeout 2s; + proxy_read_timeout 1s; + proxy_next_upstream_tries 1; proxy_intercept_errors on; error_page 500 502 503 504 = @upstream_failure; - post_action @report_upstream_outcome; - }} - - location @report_upstream_outcome {{ - internal; - if ($report_upstream_outcome = 0) {{ return 204; }} - - rewrite ^ /internal/upstream_signal break; - - proxy_intercept_errors off; - proxy_pass_request_body off; - proxy_set_header Content-Length ""; - proxy_set_header X-Upstream-Addr $upstream_addr; - proxy_set_header X-Upstream-Status $upstream_status; - proxy_set_header X-Upstream-Kind observe; - proxy_set_header X-Request-URI $request_uri; - proxy_connect_timeout 250ms; - proxy_read_timeout 250ms; - proxy_pass http://orchestrator; }} location @upstream_failure {{ internal; - + rewrite ^ /internal/upstream_signal break; - + proxy_intercept_errors off; proxy_pass_request_body off; proxy_set_header Content-Length ""; proxy_set_header X-Upstream-Addr $upstream_addr; proxy_set_header X-Upstream-Status $upstream_status; proxy_set_header X-Upstream-Kind failure; - proxy_set_header X-Request-URI $request_uri; + proxy_set_header X-Upstream-URI $request_uri; proxy_set_header X-Request-ID $request_id; proxy_connect_timeout 500ms; proxy_read_timeout 500ms; diff --git a/Orchestrator/orch/orch.py b/Orchestrator/orch/orch.py index 2d11a89..31c5ce1 100644 --- a/Orchestrator/orch/orch.py +++ b/Orchestrator/orch/orch.py @@ -192,7 +192,7 @@ def upstream_signal(request: Request): upstream = request.query_params.get("upstream") if not upstream: upstream = request.headers.get("X-Upstream-Addr", "") - + upstream_status = request.query_params.get("upstream_status") if not upstream_status: upstream_status = request.headers.get("X-Upstream-Status", "") diff --git a/Orchestrator/orch/scaling.py b/Orchestrator/orch/scaling.py index eec4f55..6b85244 100644 --- a/Orchestrator/orch/scaling.py +++ b/Orchestrator/orch/scaling.py @@ -51,28 +51,67 @@ def _current_followers(): def _set_follower_target(total_follower: int, *, trigger: str): total_follower = max(1, int(total_follower)) - followers = _current_followers() - cur_follower = len(followers) + + # 1. Gather all current workers to identify the leader and followers + try: + children = state.zk.get_children("/t") + worker_nodes = [c for c in children if c.startswith("worker")] + except Exception: + worker_nodes = [] + + leader_count = 0 + all_followers = [] + for wname in worker_nodes: + try: + data, _ = state.zk.get("/t/" + wname) + res_dict = json.loads(data.decode("utf-8")) + if res_dict.get("leader"): + leader_count += 1 + else: + res_dict["name"] = wname + all_followers.append(res_dict) + except Exception: + pass + + cur_follower = len(all_followers) dif_follower = total_follower - cur_follower state.req_followers = total_follower if dif_follower > 0: for i in range(dif_follower): state.pending_spawns += 1 - spawn_worker() - if state.autoscale_events_counter is not None: - state.autoscale_events_counter.labels(direction="up").inc() - emit(state.kafka_broker, "orchestrator", "follower_spawned", trigger=trigger) + try: + spawn_worker() + if state.autoscale_events_counter is not None: + state.autoscale_events_counter.labels(direction="up").inc() + emit(state.kafka_broker, "orchestrator", "follower_spawned", trigger=trigger) + except Exception as e: + state.pending_spawns = max(0, state.pending_spawns - 1) + logger.error("Scale up failed", extra={"error": str(e)}) elif dif_follower < 0: + # SAFETY GUARD: Do NOT scale down if there is no leader. + # During a leader crash, throughput drops to 0; we must wait for election + # to finish before trimming "idle" followers. + if leader_count == 0: + logger.info("Scaling down deferred: no leader elected", extra={"target": total_follower, "current": cur_follower}) + return + dif_follower = -dif_follower - followers = sorted(followers, key=lambda i: i['proc_id'], reverse=True) - followers = followers[:dif_follower] - for i in followers: - logger.info("Killing worker", extra={"cont_id": i["cont_id"]}) + all_followers = sorted(all_followers, key=lambda i: i['proc_id'], reverse=True) + to_trim = all_followers[:dif_follower] + for i in to_trim: + logger.info("Trimming follower", extra={"cont_id": i["cont_id"]}) emit(state.kafka_broker, "orchestrator", "worker_crashing", cont_id=i["cont_id"]) - state.zk.delete("/t/"+i["name"]) - kill_worker(i["cont_id"]) + # Use quiet=True to handle race conditions where node is already gone + try: + state.zk.delete("/t/"+i["name"]) + except Exception: + pass + try: + kill_worker(i["cont_id"]) + except Exception: + pass if state.autoscale_events_counter is not None: state.autoscale_events_counter.labels(direction="down").inc() emit(state.kafka_broker, "orchestrator", "follower_terminated", trigger=trigger, cont_id=i["cont_id"]) diff --git a/Orchestrator/orch/state.py b/Orchestrator/orch/state.py index ac4a9a8..02b08dd 100644 --- a/Orchestrator/orch/state.py +++ b/Orchestrator/orch/state.py @@ -6,13 +6,14 @@ class WorkerInfo(BaseModel): ip: str = "" znode_path: str = "" is_leader: bool = False + ready: bool = False # True only after worker calls _set_ready() proc_id: int = 0 first_seen_ts: float = 0.0 leader_since_ts: float | None = None zk = None scheduler = None -req_followers = 1 +req_followers = 2 prev_request_count = 0 pending_spawns = 0 last_seen_workers = 0 diff --git a/Orchestrator/orch/zk_watcher.py b/Orchestrator/orch/zk_watcher.py index 877f0eb..e8a11f4 100644 --- a/Orchestrator/orch/zk_watcher.py +++ b/Orchestrator/orch/zk_watcher.py @@ -52,7 +52,7 @@ def get_proxy_targets(worker_nodes): addr = f"{res_dict['cont_id']}:8000" if res_dict.get("leader"): leader_addr = addr - else: + elif res_dict.get("ready", False): follower_addrs.append(addr) except Exception: pass @@ -65,24 +65,79 @@ def update_proxy_from_workers(worker_nodes, *, trigger: str): reconcile_nginx_upstream(leader_addr, follower_addrs, trigger=trigger) +def _build_worker_state(worker_nodes: list) -> tuple[dict, dict]: + """Re-read ZK and build fresh WorkerInfo + host lookup dicts. + + Called twice in watch_workers: once at the start (initial state) and once + after election (so the promoted leader's role/timestamps are reflected before + nginx is updated). + """ + new_workers: dict[str, state.WorkerInfo] = {} + new_host_to_cont_id: dict[str, str] = {} + now = time.monotonic() + + for wname in worker_nodes: + try: + znode_path = "/t/" + wname + data, _ = state.zk.get(znode_path) + res_dict = json.loads(data.decode("utf-8")) + cont_id = (res_dict.get("cont_id") or wname).strip() + cont_ip = (res_dict.get("ip") or "").strip() + is_leader = bool(res_dict.get("leader")) + ready = bool(res_dict.get("ready", False)) + + old_info = state.workers.get(cont_id) + first_seen = old_info.first_seen_ts if old_info else now + leader_since = old_info.leader_since_ts if old_info else None + if is_leader and (not old_info or not old_info.is_leader or leader_since is None): + leader_since = now + + info = state.WorkerInfo( + cont_id=cont_id, + ip=cont_ip, + znode_path=znode_path, + is_leader=is_leader, + ready=ready, + proc_id=res_dict.get("proc_id", 0), + first_seen_ts=first_seen, + leader_since_ts=leader_since if is_leader else None, + ) + new_workers[cont_id] = info + new_host_to_cont_id[cont_id] = cont_id + if cont_ip: + new_host_to_cont_id[cont_ip] = cont_id + else: + try: + resolved_ip = socket.gethostbyname(cont_id) + if resolved_ip: + new_host_to_cont_id[resolved_ip] = cont_id + except Exception: + pass + except Exception: + pass + + return new_workers, new_host_to_cont_id + + def sync_proxy_from_zk(): if state.zk is None: return try: worker_nodes = [c for c in state.zk.get_children("/t") if c.startswith("worker")] + new_workers, new_host_to_cont_id = _build_worker_state(worker_nodes) + state.workers = new_workers + state.host_to_cont_id = new_host_to_cont_id + update_proxy_from_workers(worker_nodes, trigger="orchestrator_startup") except Exception as e: logger.warning("Failed to load workers for proxy sync", extra={"error": str(e)}) - return - - update_proxy_from_workers(worker_nodes, trigger="orchestrator_startup") def watch_workers(children): logger.info("ZK children changed", extra={"children": children}) worker_nodes = [c for c in children if c.startswith("worker")] - # 0. Build current worker identity maps + # 0. Build current worker identity maps and update state caches new_workers: dict[str, state.WorkerInfo] = {} new_host_to_cont_id: dict[str, str] = {} current_ids = {} @@ -96,36 +151,31 @@ def watch_workers(children): cont_id = (res_dict.get("cont_id") or wname).strip() cont_ip = (res_dict.get("ip") or "").strip() is_leader = bool(res_dict.get("leader")) + ready = bool(res_dict.get("ready", False)) current_ids[wname] = cont_id - - # Carry over timestamps for existing workers + old_info = state.workers.get(cont_id) first_seen = old_info.first_seen_ts if old_info else now - leader_since = (old_info.leader_since_ts if old_info else None) - - # Handle leader transition - if is_leader: - is_new_leader = not (old_info and old_info.is_leader) - if is_new_leader or leader_since is None: - leader_since = now + leader_since = old_info.leader_since_ts if old_info else None + if is_leader and (not old_info or not old_info.is_leader or leader_since is None): + leader_since = now info = state.WorkerInfo( cont_id=cont_id, ip=cont_ip, znode_path=znode_path, is_leader=is_leader, + ready=ready, proc_id=res_dict.get("proc_id", 0), first_seen_ts=first_seen, - leader_since_ts=leader_since if is_leader else None + leader_since_ts=leader_since if is_leader else None, ) - new_workers[cont_id] = info new_host_to_cont_id[cont_id] = cont_id if cont_ip: new_host_to_cont_id[cont_ip] = cont_id else: - # Fallback IP resolution try: resolved_ip = socket.gethostbyname(cont_id) if resolved_ip: @@ -134,6 +184,7 @@ def watch_workers(children): pass except Exception: pass + new_count = len(worker_nodes) state.workers = new_workers state.host_to_cont_id = new_host_to_cont_id @@ -201,7 +252,10 @@ def watch_workers(children): for follower in follower_nodes[:excess]: logger.info("Trimming extra follower", extra={"cont_id": follower["cont_id"]}) emit(state.kafka_broker, "orchestrator", "worker_crashing", cont_id=follower["cont_id"]) - state.zk.delete("/t/" + follower["name"]) + try: + state.zk.delete("/t/" + follower["name"]) + except Exception: + pass kill_worker(follower["cont_id"]) if state.autoscale_events_counter is not None: state.autoscale_events_counter.labels(direction="down").inc() @@ -209,9 +263,9 @@ def watch_workers(children): # 3. Promote the new leader if election occurred if best_dict: - emit(state.kafka_broker, "orchestrator", "leader_election_started", - reason="no leader found") + emit(state.kafka_broker, "orchestrator", "leader_election_started", reason="no leader found") best_dict["leader"] = True + best_dict["ready"] = False # Force recovery period β€” probe guard relies on this res_bytes = json.dumps(best_dict).encode('utf-8') state.zk.set("/t/" + best_name, res_bytes) duration_ms = round((time.monotonic() - election_start) * 1000, 2) @@ -222,7 +276,14 @@ def watch_workers(children): emit(state.kafka_broker, "orchestrator", "leader_elected", new_leader=best_dict.get("cont_id", ""), duration_ms=duration_ms) - # 4. Update routing to point to the correct leader and followers + # 4. Refresh state caches BEFORE updating nginx so the promoted leader's + # WorkerInfo (ready=False, leader_since_ts=now) is in place before 502s arrive. + # The not-ready guard in _should_skip_failover depends on this ordering. + new_workers, new_host_to_cont_id = _build_worker_state(worker_nodes) + state.workers = new_workers + state.host_to_cont_id = new_host_to_cont_id + + # 5. Update routing to point to the correct leader and followers. update_proxy_from_workers(worker_nodes, trigger="topology_change") diff --git a/Orchestrator/prometheus/prometheus.yml b/Orchestrator/prometheus/prometheus.yml index 6e62c3e..453f4e4 100644 --- a/Orchestrator/prometheus/prometheus.yml +++ b/Orchestrator/prometheus/prometheus.yml @@ -1,5 +1,6 @@ global: - scrape_interval: 30s + scrape_interval: 10s + evaluation_interval: 10s scrape_configs: - job_name: orchestrator @@ -9,4 +10,4 @@ scrape_configs: - job_name: workers http_sd_configs: - url: 'http://orchestrator:80/api/v1/prometheus/targets' - refresh_interval: 30s + refresh_interval: 15s diff --git a/Orchestrator/worker/follower.py b/Orchestrator/worker/follower.py index f8afff9..1e71aea 100644 --- a/Orchestrator/worker/follower.py +++ b/Orchestrator/worker/follower.py @@ -37,17 +37,22 @@ stats_lock = threading.Lock() def telemetry_worker(): - """Background thread to emit aggregated metrics every 5 seconds.""" + """Background thread to emit aggregated metrics every 1 second.""" while True: - time.sleep(5) + time.sleep(1) with stats_lock: - avg_latency = (telemetry_stats["read_latency_sum"] / telemetry_stats["read_count"]) if telemetry_stats["read_count"] > 0 else 0 - emit(KAFKA_BROKER, f"follower-{FOLLOWER_ID}", "telemetry_snapshot", - read_count=telemetry_stats["read_count"], - avg_read_latency_ms=round(avg_latency, 2), - lag_ms=round(telemetry_stats["latest_lag"], 2)) + # Atomic swap to avoid missing requests during calculation + stats_copy = telemetry_stats.copy() telemetry_stats["read_count"] = 0 telemetry_stats["read_latency_sum"] = 0 + # latest_lag is not reset as it's a current measurement + + avg_latency = (stats_copy["read_latency_sum"] / stats_copy["read_count"]) if stats_copy["read_count"] > 0 else 0 + + emit(KAFKA_BROKER, f"follower-{FOLLOWER_ID}", "telemetry_snapshot", + read_count=stats_copy["read_count"], + avg_read_latency_ms=round(avg_latency, 2), + replica_lag_ms=stats_copy["latest_lag"]) threading.Thread(target=telemetry_worker, daemon=True).start() @@ -64,6 +69,20 @@ def telemetry_worker(): _zk.start() _zk.ensure_path("/t/snapshots") +def _set_ready(): + try: + for c in _zk.get_children("/t"): + if c.startswith("worker"): + data, _ = _zk.get(f"/t/{c}") + res = json.loads(data.decode()) + if res.get("cont_id") == FOLLOWER_ID: + res["ready"] = True + _zk.set(f"/t/{c}", json.dumps(res).encode()) + logger.info("Follower marked as ready") + return + except Exception as e: + logger.warning("Failed to mark follower as ready", extra={"error": str(e)}) + ############ SyncQ Consumer (background thread) ############ def sync_consumer(): @@ -252,4 +271,5 @@ def _start_sync(): emit(KAFKA_BROKER, f"follower-{FOLLOWER_ID}", "recovery_complete") t = threading.Thread(target=_start_sync, daemon=True) t.start() + _set_ready() uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/Orchestrator/worker/leader.py b/Orchestrator/worker/leader.py index e66ca33..367dd01 100644 --- a/Orchestrator/worker/leader.py +++ b/Orchestrator/worker/leader.py @@ -32,6 +32,8 @@ kv = KVStore("/data/rideshare.ldb") write_counter = 0 +write_counter_lock = threading.Lock() +last_snapshot_time = time.monotonic() ############ Telemetry Aggregation ############ @@ -39,18 +41,21 @@ stats_lock = threading.Lock() def telemetry_worker(): - """Background thread to emit aggregated metrics every 5 seconds.""" + """Background thread to emit aggregated metrics every 1 second.""" while True: - time.sleep(5) + time.sleep(1) with stats_lock: - avg_latency = (write_stats["latency_sum"] / write_stats["count"]) if write_stats["count"] > 0 else 0 - emit(KAFKA_BROKER, f"leader-{LEADER_ID}", "telemetry_snapshot", - count=write_stats["count"], - avg_latency_ms=round(avg_latency, 2), - kafka_offset=write_stats["max_offset"]) + # Atomic swap + stats_copy = write_stats.copy() write_stats["count"] = 0 write_stats["latency_sum"] = 0 + avg_latency = (stats_copy["latency_sum"] / stats_copy["count"]) if stats_copy["count"] > 0 else 0 + emit(KAFKA_BROKER, f"leader-{LEADER_ID}", "telemetry_snapshot", + count=stats_copy["count"], + avg_latency_ms=round(avg_latency, 2), + kafka_offset=stats_copy["max_offset"]) + threading.Thread(target=telemetry_worker, daemon=True).start() ############ Prometheus metrics ############ @@ -66,6 +71,20 @@ def telemetry_worker(): _zk.start() _zk.ensure_path("/t/snapshots") +def _set_ready(): + try: + for c in _zk.get_children("/t"): + if c.startswith("worker"): + data, _ = _zk.get(f"/t/{c}") + res = json.loads(data.decode()) + if res.get("cont_id") == LEADER_ID: + res["ready"] = True + _zk.set(f"/t/{c}", json.dumps(res).encode()) + logger.info("Leader marked as ready") + return + except Exception as e: + logger.warning("Failed to mark leader as ready", extra={"error": str(e)}) + ############ Kafka Producer ############ def _make_kafka_producer(): @@ -209,12 +228,17 @@ def write(req: WriteRequest): write_stats["latency_sum"] += latency_ms write_stats["max_offset"] = max(write_stats["max_offset"], offset) - write_counter += 1 - if write_counter >= SNAPSHOT_INTERVAL: - write_counter = 0 - _snapshotter.trigger() + with write_counter_lock: + global write_counter, last_snapshot_time + write_counter += 1 + should_snap = write_counter >= SNAPSHOT_INTERVAL or (time.monotonic() - last_snapshot_time) > 120 + if should_snap: + write_counter = 0 + last_snapshot_time = time.monotonic() + _snapshotter.trigger() return {"status": "ok"} if __name__ == "__main__": + _set_ready() uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/Orchestrator/worker/leader_snapshot.py b/Orchestrator/worker/leader_snapshot.py index 4cb19f7..9a774f1 100644 --- a/Orchestrator/worker/leader_snapshot.py +++ b/Orchestrator/worker/leader_snapshot.py @@ -78,7 +78,12 @@ def _take_snapshot(self): write_snapshot(self._kv.snapshot_dict()) write_meta({"kafka_offset": offset, "kafka_partition": KAFKA_PARTITION, "ts": datetime.utcnow().isoformat()}) + + # 1. Update global latest offset zk_set(self._zk, "/t/snapshots/latest_offset", str(offset).encode()) + # 2. Update leader-specific offset (so orchestrator sees our progress during elections) + zk_set(self._zk, f"/t/snapshots/{self._leader_id}", str(offset).encode()) + self._publish({"type": SYNC_MSG_SNAPSHOT, "kafka_offset": offset, "kafka_partition": KAFKA_PARTITION, "ts": datetime.utcnow().isoformat()}) # Update gauge and emit event diff --git a/Orchestrator/worker/worker.py b/Orchestrator/worker/worker.py index 29d24c2..da849dc 100644 --- a/Orchestrator/worker/worker.py +++ b/Orchestrator/worker/worker.py @@ -25,6 +25,7 @@ def get_container_ip() -> str: ############ Creating Z-node ############ cont_id = socket.gethostname() cont_ip = get_container_ip() + # Connect to Orchestrator with retry url = f'http://{ORCHESTRATOR_HOST}/getpid/{cont_id}' while True: @@ -41,7 +42,8 @@ def get_container_ip() -> str: "cont_id":cont_id, "ip":cont_ip, "proc_id":proc_id, - "leader":False + "leader":False, + "ready":False } res_bytes = json.dumps(data).encode('utf-8') diff --git a/README.md b/README.md index 177f043..20f3ad3 100644 --- a/README.md +++ b/README.md @@ -6,66 +6,50 @@ ![ZooKeeper](https://img.shields.io/badge/Coordination-ZooKeeper-yellow) ![Kafka](https://img.shields.io/badge/WAL-Kafka-black) -A distributed Database-as-a-Service (DBaaS) platform built around a primary-backup replicated KV store. The platform handles read/write routing, autoscaling, leader election, and fast failure recovery β€” and ships with a full operator console for running workloads and observing the cluster in real time. +A distributed key-value Database-as-a-Service built from scratch. Primary-backup replication, automatic leader election, RPS-based autoscaling, and fast failure recovery β€” all observable through a live operator console. --- -## What's in here - -| Component | Description | -|-----------|-------------| -| **Orchestrator** (FastAPI) | Control plane: ZooKeeper watches, leader election, autoscaling, Nginx config authority, failure handling | -| **Nginx** | Data-plane router: routes `/write` to the leader, `/read` across followers, reports upstream failures back to the orchestrator | -| **Worker** (Python) | Shared container image; starts as follower, promoted to leader by the orchestrator via ZooKeeper znode update | -| **Workload Generator** (FastAPI + Locust) | Synthetic load service with a full runtime-configurable parameter set | -| **Frontend** (React + Vite) | Operator console: topology graph, live metrics, autoscaling config, workload lab with time-series charts | -| **Kafka** | Durable write-ahead log; workers replay it on startup for state recovery | -| **RabbitMQ** | Replication fanout: leader publishes writes, followers consume | -| **ZooKeeper** | Worker membership, leader role, snapshot offsets | -| **Prometheus + Grafana + Tempo** | Metrics, dashboards, distributed traces | + --- -## Quick Start +## How it works -### Prerequisites +Writes go through Nginx to a single elected leader. The leader applies the operation to its local LevelDB, publishes it to a RabbitMQ fanout for fast follower sync, and commits it to Kafka as a durable WAL. Reads are load-balanced across followers, which serve from their own LevelDB replica. -- Docker Desktop (macOS: enable **"Allow the default Docker socket to be used"** in Settings β†’ Advanced so the orchestrator can manage worker containers) -- Docker Compose v2 +The orchestrator watches ZooKeeper for membership changes and handles everything that happens in the control plane: leader election when a leader is missing, spawning or trimming followers to match a target RPS, and pushing updated Nginx upstream configs when topology changes. It never touches the read/write request path. -### Start the full stack +When Nginx detects an upstream failure it calls back to the orchestrator, which probes the worker and β€” if dead β€” deletes its ZooKeeper znode immediately rather than waiting for session expiry. This cuts failover from ~30s to ~2s. -```bash -docker compose -f docker-compose-local.yml up --build -``` +--- -Docker waits for RabbitMQ and ZooKeeper to pass their health checks before starting the orchestrator. The cluster is ready when you see `"Scheduler started"` in the orchestrator logs. +## Screenshots -### Service endpoints +**Topology graph + event stream during a leader failover** + +![Topology graph](_docs/screenshots/topology-failover.png) +*Leader crash β†’ election β†’ new leader promoted β†’ Nginx config reconciled. Visible in the event log in real time.* -| Service | URL | Notes | -|---------|-----|-------| -| **Frontend** | http://localhost:5173 | Operator console (Observability + Workload tabs) | -| **Orchestrator API** | http://localhost:8080 | Control plane REST + WebSocket | -| **Nginx (data plane)** | http://localhost:80 | Send reads/writes here | -| **Workload Generator** | http://localhost:8089 | Job control API | -| **Grafana** | http://localhost:3001 | Dashboards (admin / `dbaas`) | -| **Prometheus** | http://localhost:9090 | Raw metrics | -| **Tempo** | http://localhost:3200 | Distributed traces | +**Workload tab β€” live KPIs under load** + +![Workload tab](_docs/screenshots/workload-kpis.png) +*p50/p95/p99 latency, reads/writes per second, error count, rolling time-series chart. Autoscaler spawning followers visible in the topology graph simultaneously.* -### Quick health check - -```bash -curl http://localhost:8080/health # orchestrator -curl http://localhost:80/health # nginx -curl http://localhost:8089/health # workload generator -``` --- -## Architecture Overview +## Architecture -```ascii +``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ FRONTEND (React / Vite) β”‚ β”‚ - Observability tab: topology graph, metrics, scaling config, event log β”‚ @@ -82,7 +66,7 @@ curl http://localhost:8089/health # workload generator β”‚ - upstream failure handling β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ - Prometheus HTTP SD β”‚ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ HTTP traffic - β”‚ POST /internal/reconcile β–Ό + β”‚ POST /internal/reconcile β–Ό β”‚ (on topology change) β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β–Ό β”‚ target: nginx:80 β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” @@ -95,127 +79,118 @@ curl http://localhost:8089/health # workload generator β”‚ --bootstrap: pulls config from orchestrator on container start β”‚ β”‚ --serve: handles POST /internal/reconcile β†’ fetch + write + reload β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ - β”‚ writes β”‚ reads (round-robin) + β”‚ writes β”‚ reads (round-robin) β–Ό β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ LEADER WORKER β”‚ β”‚ FOLLOWER WORKER(S) β”‚ β”‚ - LevelDB writes β”‚ β”‚ - LevelDB reads β”‚ β”‚ - RabbitMQ fanout publish │─────────▢│ - RabbitMQ sync consumer β”‚ -β”‚ - Kafka WAL publish β”‚ β”‚ - snapshot recovery β”‚ -β”‚ - snapshot producer β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +β”‚ - Kafka WAL publish β”‚ β”‚ - snapshot recovery on startup β”‚ +β”‚ - periodic snapshots β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ KAFKA + RABBITMQ + ZOOKEEPER β”‚ -β”‚ - Kafka: durable write-ahead log β”‚ -β”‚ - RabbitMQ: replication fanout to followers β”‚ +β”‚ - Kafka: durable write-ahead log (dbaas-writes WAL, dbaas.events stream) β”‚ +β”‚ - RabbitMQ: replication fanout to followers (sync exchange) β”‚ β”‚ - ZooKeeper: worker membership, leader role, snapshot offsets β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` -**Data flow for a write:** Client β†’ Nginx `/write` β†’ Leader β†’ RabbitMQ fanout β†’ all Followers. The write is also published to Kafka as a durable WAL entry. +--- -**Data flow for a read:** Client β†’ Nginx `/read` β†’ Follower pool (round-robin across all registered followers; Nginx retries on failure up to 3 times). If a follower returns 5xx, a `post_action` callback notifies the orchestrator, which probes and removes the unhealthy node. +## Stack -**Failure detection:** Two complementary paths. First, Nginx intercepts 5xx errors and fires callbacks to the orchestrator, which probes the worker's `/health` endpoint and, if confirmed dead, removes its ZooKeeper znode and kills the container β€” bypassing ZooKeeper session expiry for faster failover. Second, the orchestrator's ZooKeeper `ChildrenWatch` detects znode removals (session expiry, explicit delete) and immediately triggers leader election and Nginx config reconciliation. +| Layer | Technology | +|-------|-----------| +| Control plane | Python, FastAPI, Kazoo (ZooKeeper), APScheduler | +| Data plane | Nginx, LevelDB (`plyvel`) | +| Replication | RabbitMQ (fanout), Kafka (WAL + replay) | +| Coordination | ZooKeeper | +| Load testing | Locust (embedded runner) | +| Frontend | React, Vite, Recharts | +| Observability | Prometheus, Grafana, OpenTelemetry / Tempo | +| Infrastructure | Docker, Docker Compose | --- -## Configuration Reference +## Interesting parts -### Orchestrator environment variables - -| Variable | Default | Description | -|---|---|---| -| `SCALING_INTERVAL_SECONDS` | `30` | How often the autoscaler evaluates load | -| `SCALING_RPS_PER_WORKER` | `10.0` | RPS target per follower worker | -| `SNAPSHOT_INTERVAL` | `2000` | Write operations between worker state snapshots | -| `DOCKER_NETWORK` | `local_dbaas_network` | Docker network for spawned worker containers | -| `ZOOKEEPER_HOST` | `zookeeper` | ZooKeeper hostname | -| `RABBITMQ_HOST` | `rabbitmq` | RabbitMQ hostname | -| `KAFKA_BROKER` | `kafka:9092` | Kafka broker address | -| `NGINX_URL` | `http://nginx` | Nginx base URL (for version polling) | -| `NGINX_RECONCILE_URL` | `http://nginx:8081/internal/reconcile` | Nginx reconcile sidecar URL | -| `FAILURE_PROBE_COOLDOWN_SEC` | `1.5` | Minimum seconds between probes of the same worker | +**Leader election** β€” when the orchestrator detects no leader in ZooKeeper, it ranks candidates by their last snapshot offset (most caught-up replica first), with `proc_id` as a deterministic tie-break. The elected node's znode is updated; the worker process picks up the change via `DataWatch` and swaps its subprocess from `follower.py` to `leader.py`. -The autoscaling parameters (`SCALING_INTERVAL_SECONDS`, `SCALING_RPS_PER_WORKER`) can also be updated live at runtime via `POST /api/v1/config/scaling` without restarting the orchestrator. The Observability tab sidebar exposes this as a form. +**Dual-channel replication** β€” writes go to RabbitMQ fanout for low-latency follower propagation *and* to Kafka as a durable WAL. They serve different purposes: RabbitMQ minimises replica lag under normal operation; Kafka is what followers replay on startup. A follower that crashed and recovered locally skips the leader fetch entirely if its snapshot is fresh enough. -### Workload Generator environment variables +**Nginx as active failure detector** β€” Nginx intercepts upstream 5xx errors and calls back to the orchestrator via internal locations (`@upstream_failure`, `@report_upstream_outcome`). The orchestrator sends two health probes (750ms timeout, 150ms apart) before acting. On confirmed failure it deletes the ZooKeeper znode directly, which triggers election and Nginx reconcile without waiting for ZooKeeper session expiry. -| Variable | Default | Description | -|---|---|---| -| `TARGET_HOST` | `http://nginx:80` | Where to send synthetic traffic | -| `ORCHESTRATOR_URL` | `http://orchestrator:80` | Used to reset autoscaler on job stop | -| `READ_RATIO` | `80` | Initial read percentage (0–100) | -| `KEY_SPACE` | `1000` | Initial key space size | -| `KEY_DISTRIBUTION` | `uniform` | Initial distribution (`uniform`, `zipf`, `hotspot`) | -| `MAX_RPS_PER_USER` | `5` | Initial per-user RPS cap | +**Two autoscaling loops** β€” an interval-based scheduler derives RPS from Nginx `stub_status` counter deltas and adjusts follower count to a target. A separate ZooKeeper `ChildrenWatch` handles crashes between scheduler ticks. They share a `pending_spawns` counter to avoid double-spawning during churn. Both `scaling_interval_seconds` and `rps_per_worker` are adjustable at runtime without restart. -All parameters can be overridden per job start via the API or the Workload UI. +**Versioned Nginx config** β€” the orchestrator renders Nginx config and hashes it (SHA-256 over sorted `{leader, followers}`). The reconcile sidecar skips the reload if the current version matches. This makes every topology event trigger a reconcile call without causing unnecessary Nginx reloads. --- -## Operator Console +## Quick start -Open http://localhost:5173 after starting the stack. +```bash +# Requires Docker Desktop with the default Docker socket enabled +docker compose -f docker-compose-local.yml up --build +``` -**Observability tab:** -- Live topology graph showing leader/follower roles and health -- Metrics panel (request rate, error rate) -- **Autoscaling Strategy panel** β€” tune `scaling_interval_seconds` and `scaling_rps_per_worker` live -- Event log streamed from Kafka via WebSocket -- Node detail panel with manual crash controls (crash by ID) +The cluster is ready when the orchestrator logs show `"Scheduler started"`. -**Workload tab:** -- Configure concurrent users, spawn rate, read/write ratio, key space, per-user RPS cap, and key distribution -- Start/stop workload jobs; "Apply & Restart" while running to change parameters mid-test -- Configured Ceiling and Estimated Throughput prediction cards -- Live KPI grid (p50/p95/p99 latency, reads/sec, writes/sec, error count, runner state) -- Rolling time-series chart -- Error distribution table +| Service | URL | +|---------|-----| +| Operator console | http://localhost:5173 | +| Orchestrator API | http://localhost:8080 | +| Nginx (data plane) | http://localhost:80 | +| Workload Generator | http://localhost:8089 | +| Grafana | http://localhost:3001 | +| Prometheus | http://localhost:9090 | + +```bash +# Quick health check +curl http://localhost:8080/health +curl http://localhost:80/health +curl http://localhost:8089/health +``` --- -## Key API Endpoints +## Try it -### Orchestrator (port 8080) +**Trigger a failover manually:** +1. Open the operator console β†’ Observability tab +2. Click a worker node β†’ crash it from the Node Detail panel +3. Watch the event log: `worker_crashing` β†’ `leader_election_started` β†’ `leader_elected` β†’ Nginx reconcile -| Method | Path | Description | -|---|---|---| -| `GET` | `/health` | Liveness | -| `WS` | `/stream` | Kafka event stream (WebSocket) | -| `GET` | `/list_workers` | Live worker list with roles | -| `GET` | `/api/v1/config/scaling` | Current autoscaling params | -| `POST` | `/api/v1/config/scaling` | Update autoscaling params live | -| `POST` | `/api/autoscale/reset` | Reset autoscaler baseline (called by workload generator on stop) | -| `POST` | `/api/v1/crash/worker/{cont_id}` | Kill specific worker | -| `GET` | `/api/v1/prometheus/targets` | Prometheus HTTP SD format (live worker discovery) | -| `GET` | `/metrics` | Prometheus metrics | - -### Workload Generator (port 8089) - -| Method | Path | Description | -|---|---|---| -| `GET` | `/health` | Liveness | -| `GET` | `/api/jobs` | Current job state + config | -| `GET` | `/api/jobs/stats` | Live Locust stats snapshot | -| `POST` | `/api/jobs/start` | Start or restart a workload job | -| `POST` | `/api/jobs/stop` | Stop current job + reset autoscaler | +**Run a load test and watch the autoscaler:** +1. Go to the Workload tab, set concurrent users and a key distribution (`zipf` is interesting) +2. Start the job and watch followers spawn in the topology graph as RPS climbs +3. Stop the job β€” autoscaler resets and trims back to 1 follower --- -## Telemetry +## Configuration -- **Prometheus** scrapes the orchestrator at `/metrics` and dynamically discovers workers via the HTTP SD endpoint at `/api/v1/prometheus/targets` -- **Grafana** (http://localhost:3001, admin / `dbaas`) has pre-provisioned dashboards -- **Tempo** collects distributed traces from the orchestrator (OpenTelemetry via OTLP gRPC on port 4317) -- Orchestrator emits structured JSON logs to stdout +Key environment variables (can also be set live via the API for autoscaling params): + +| Variable | Default | Description | +|---|---|---| +| `SCALING_INTERVAL_SECONDS` | `30` | Autoscaler evaluation interval | +| `SCALING_RPS_PER_WORKER` | `10.0` | RPS target per follower | +| `SNAPSHOT_INTERVAL` | `2000` | Writes between leader snapshots | +| `FAILURE_PROBE_COOLDOWN_SEC` | `1.5` | Minimum time between probes of the same worker | --- -## Architecture Documentation +## Documentation -For a full breakdown of the architecture, design decisions, and known limitations, see [_docs/v2.md](_docs/v2.md). +Detailed architecture, component internals, and design decisions are in [`_docs/system/`](_docs/system/README.md): -Earlier iteration notes: [_docs/v1.md](_docs/v1.md) | [_docs/v0.md](_docs/v0.md) +- [Architecture & layout](_docs/system/processes/architecture-and-layout.md) +- [Orchestrator β€” election, autoscaler, failure detection](_docs/system/components/orchestrator.md) +- [Workers β€” write path, striped locks, snapshot protocol](_docs/system/components/workers.md) +- [Nginx routing & reconcile sidecar](_docs/system/components/nginx-routing.md) +- [Data path & failure scenario walkthrough](_docs/system/processes/data-path-and-recovery.md) +- [Scaling & failover](_docs/system/processes/scaling-and-failover.md) +- [Design decisions & tradeoffs](_docs/system/decisions/major-decisions.md) diff --git a/_docs/screenshots/grafana-dashboard.png b/_docs/screenshots/grafana-dashboard.png new file mode 100644 index 0000000..c9eb962 Binary files /dev/null and b/_docs/screenshots/grafana-dashboard.png differ diff --git a/_docs/screenshots/topology-failover.png b/_docs/screenshots/topology-failover.png new file mode 100644 index 0000000..14d6dd0 Binary files /dev/null and b/_docs/screenshots/topology-failover.png differ diff --git a/_docs/screenshots/workload-kpis.png b/_docs/screenshots/workload-kpis.png new file mode 100644 index 0000000..956c372 Binary files /dev/null and b/_docs/screenshots/workload-kpis.png differ diff --git a/_docs/system/README.md b/_docs/system/README.md new file mode 100644 index 0000000..fffc8e6 --- /dev/null +++ b/_docs/system/README.md @@ -0,0 +1,84 @@ +# Distributed DBaaS β€” System Documentation + +A distributed key-value store built to exercise real control-plane concerns: leader election, follower replication, autoscaling, and failure recovery β€” all observable in real time through an operator console. + +**Stack:** Python (FastAPI, LevelDB, Kafka-python, Kazoo, Pika, Locust), React, ZooKeeper, Kafka, RabbitMQ, Nginx, Docker, Prometheus, OpenTelemetry + +--- + +## Architecture + +``` +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ FRONTEND (React / Vite) β”‚ +β”‚ - Observability tab: topology graph, metrics, scaling config, event log β”‚ +β”‚ - Workload tab: workload config, live KPIs, time-series chart β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ HTTP / WS β”‚ HTTP + β–Ό β–Ό +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ ORCHESTRATOR (FastAPI) β”‚ β”‚ WORKLOAD GENERATOR (FastAPI) β”‚ +β”‚ - ZooKeeper children watch β”‚ β”‚ - Locust Environment + Runner β”‚ +β”‚ - leader election β”‚ β”‚ - /api/jobs start / stop / stats β”‚ +β”‚ - RPS-based autoscaling β”‚ β”‚ - runtime workload config β”‚ +β”‚ - Nginx config source of truth β”‚ β”‚ - resets autoscaler on stop β”‚ +β”‚ - upstream failure handling β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +β”‚ - Prometheus HTTP SD β”‚ β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ HTTP traffic + β”‚ POST /internal/reconcile β–Ό + β”‚ (on topology change) β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” + β–Ό β”‚ target: nginx:80 β”‚ +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ NGINX (data-plane router) port 80 β”‚ +β”‚ /write β†’ leader_upstream β”‚ +β”‚ /read β†’ follower_upstream (round-robin across followers, retry on fail) β”‚ +β”‚ /api/stats β†’ stub_status (used by orchestrator for autoscaling) β”‚ +β”‚ β”‚ +β”‚ RECONCILE SIDECAR (reconcile.py) port 8081 β”‚ +β”‚ --bootstrap: pulls config from orchestrator on container start β”‚ +β”‚ --serve: handles POST /internal/reconcile β†’ fetch + write + reload β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ writes β”‚ reads (round-robin) + β–Ό β–Ό +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ LEADER WORKER β”‚ β”‚ FOLLOWER WORKER(S) β”‚ +β”‚ - LevelDB writes β”‚ β”‚ - LevelDB reads β”‚ +β”‚ - RabbitMQ fanout publish │─────────▢│ - RabbitMQ sync consumer β”‚ +β”‚ - Kafka WAL publish β”‚ β”‚ - snapshot recovery β”‚ +β”‚ - snapshot producer β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ + β–Ό +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ KAFKA + RABBITMQ + ZOOKEEPER β”‚ +β”‚ - Kafka: durable write-ahead log (dbaas-writes WAL, dbaas.events stream) β”‚ +β”‚ - RabbitMQ: replication fanout to followers (sync exchange) β”‚ +β”‚ - ZooKeeper: worker membership, leader role, snapshot offsets β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +``` + +--- + +## How It Works in One Paragraph + +Writes go through Nginx to a single elected leader. The leader applies the op to LevelDB, publishes it to a RabbitMQ fanout exchange for fast follower sync, then commits it to Kafka as a durable WAL. Reads are load-balanced across followers, which serve stale-but-fast reads from their local LevelDB copy. ZooKeeper holds ephemeral worker registrations β€” the orchestrator watches those for membership changes, runs election when a leader is missing, and reconciles Nginx upstream config on every topology change. The autoscaler runs on a separate interval loop, deriving RPS from Nginx's `stub_status` counter delta and spawning/trimming followers to match a target RPS-per-worker. When Nginx detects an upstream failure it calls back to the orchestrator, which probes the worker and β€” if confirmed dead β€” deletes its ZK znode immediately rather than waiting for session expiry. + +--- + +## Reading Order + +1. [Architecture & Folder Layout](./processes/architecture-and-layout.md) +2. [Orchestrator](./components/orchestrator.md) +3. [Workers (Leader/Follower)](./components/workers.md) +4. [Nginx Routing & Reconcile](./components/nginx-routing.md) +5. [Workload Generator](./components/workload-generator.md) +6. [Frontend Console](./components/frontend-console.md) +7. [Write/Read + Snapshot/Recovery](./processes/data-path-and-recovery.md) +8. [Scaling & Failover](./processes/scaling-and-failover.md) +9. [Major Decisions & Tradeoffs](./decisions/major-decisions.md) + +--- + +## Scope + +Single-cluster, single Kafka partition (`dbaas-writes`, partition 0). One leader at a time. Follower count is dynamic, controlled by the autoscaler. This keeps the operational surface small while exercising the interesting control-plane behavior β€” election, failover, routing reconciliation, autoscaling β€” without the noise of multi-cluster coordination. diff --git a/_docs/system/components/frontend-console.md b/_docs/system/components/frontend-console.md new file mode 100644 index 0000000..a3ff860 --- /dev/null +++ b/_docs/system/components/frontend-console.md @@ -0,0 +1,35 @@ +# Frontend Console + +The frontend is an operator console β€” it's meant to let you watch the system respond to load and failures, not just display static state. + +## What It Shows + +- **Topology graph** β€” live leader/follower nodes with role and status (healthy / crashing / recovering) +- **Event stream** β€” real-time feed from the orchestrator WebSocket (`/stream`), showing elections, joins, terminations, snapshot events, telemetry +- **Scaling config** β€” live controls for `scaling_interval_seconds` and `scaling_rps_per_worker`, applied without redeploy +- **Workload controls** β€” start/stop load tests with configurable user count, RPS cap, key distribution, read ratio +- **Metrics** β€” observed throughput and latency from the workload generator + +## Data Sources + +- `useWebSocket` β€” persistent WebSocket to `/stream`; reconnects automatically on disconnect +- `useTopology` β€” HTTP polling of `/list_workers` for authoritative topology snapshots +- Workload generator APIs β€” polled for job state and live stats + +## Topology State: Hybrid Event + Poll + +`useTopology` merges two signals. HTTP polling gives authoritative ground truth but is slow. WebSocket events are fast but don't include full topology state. + +The hook handles specific events to update UI state immediately without waiting for the next poll: + +- `worker_crashing` β†’ node is immediately marked `crashing` in the local ref (`crashingIds`) +- `worker_joined` β†’ node is marked `recovering` +- `recovery_complete` β†’ `recovering` status cleared + +Events like `leader_elected`, `follower_spawned`, `worker_terminated` trigger a topology refresh (new HTTP fetch) to get the updated node list. This way the graph responds in near-real-time to control-plane events while staying consistent with backend state on every refresh. + +The `normalizeWorkerIdentity` helper in `useTopology.js` handles the fact that different event types use different field names for the same concept (`id`, `name`, `container_id`, `cont_id`) β€” this was a pragmatic fix to avoid coupling the frontend to a single event schema. + +## Crash Button + +The console exposes a per-node crash action (`POST /api/v1/crash/worker/{cont_id}`) that kills a specific container via the orchestrator. This is useful for manually triggering failover and watching the election + routing convergence play out in the event stream. diff --git a/_docs/system/components/nginx-routing.md b/_docs/system/components/nginx-routing.md new file mode 100644 index 0000000..162387e --- /dev/null +++ b/_docs/system/components/nginx-routing.md @@ -0,0 +1,44 @@ +# Nginx Routing and Reconcile + +Nginx is the only externally visible endpoint. All client traffic enters here. + +## Route Map + +``` +/write β†’ leader_upstream (single server, proxy_intercept_errors on) +/read β†’ follower_upstream (round-robin, proxy_next_upstream with 3 retries, 2s timeout) +/health β†’ 200 ok +/api/stats β†’ stub_status (used by autoscaler for cumulative request count) +``` + +`/read` uses `proxy_next_upstream` with a 250ms connect timeout and 2s read timeout. On failure, Nginx retries up to 3 times across the follower pool before giving up. + +## Config Versioning + +`nginx_config.py` computes a 12-character version string (SHA-256 hex digest) over `{"leader": addr, "followers": sorted(addrs)}`. This version is embedded as the first line of the config file (`# proxy_config_version: `) and sent as the `X-Proxy-Config-Version` response header. + +The reconcile sidecar compares the desired version against a persisted version file (`/var/run/nginx-proxy-config.version`). If they match, the reconcile is a no-op. This makes topology updates idempotent β€” if the orchestrator calls reconcile twice with the same topology, Nginx is not reloaded the second time. + +## Reconcile Sidecar (`reconcile.py`) + +The sidecar runs as a separate process inside the Nginx container. It has two modes: + +- `--bootstrap`: called at container startup; fetches config from orchestrator and writes it to disk before Nginx starts, without sending a reload signal +- `--serve`: starts a `ThreadingHTTPServer` on port 8081, listening for `POST /internal/reconcile` + +When the orchestrator calls `POST /internal/reconcile`, the sidecar: +1. Fetches the desired config from `GET /internal/nginx/config` on the orchestrator +2. Compares version against the persisted version file +3. If different: atomic file write (temp file β†’ `os.rename`), persist new version, `nginx -s reload` + +Atomic write prevents Nginx from ever seeing a partial config. + +## Upstream Failure Signaling + +Nginx has two internal locations that call back to the orchestrator: + +**`@upstream_failure`** β€” fires on 502/503/504 from `/write` or `/read`. The orchestrator gets `kind=failure`, resolves the upstream host to a container ID via ZooKeeper, and starts a health probe. + +**`@report_upstream_outcome`** β€” fires via `post_action` on `/read` for any non-2xx outcome (controlled by the `$report_upstream_outcome` map variable). The orchestrator gets `kind=observe` and probes all failing upstream addresses in the response. + +This gives the orchestrator a faster failure signal than ZooKeeper session expiry, which can take 30+ seconds. diff --git a/_docs/system/components/orchestrator.md b/_docs/system/components/orchestrator.md new file mode 100644 index 0000000..4aea724 --- /dev/null +++ b/_docs/system/components/orchestrator.md @@ -0,0 +1,66 @@ +# Orchestrator + +The orchestrator is the control plane. It does not touch the read/write request path β€” its job is membership, election, scaling, and routing. + +## Modules + +| File | What it does | +|------|-------------| +| `orch.py` | FastAPI app, startup lifespan, API surface, Kafkaβ†’WebSocket relay | +| `zk_watcher.py` | `ChildrenWatch` callback β€” join/leave events, election, recovery spawn, target trim | +| `scaling.py` | Interval autoscaler β€” Nginx RPS delta β†’ follower target | +| `failure_detection.py` | Nginx failure callbacks β†’ health probe β†’ znode delete + container kill | +| `docker_utils.py` | Worker spawn/kill via Docker SDK, Nginx reconcile trigger | +| `nginx_config.py` | Nginx config template renderer + config version hash | +| `state.py` | Shared in-process state: ZK client, scheduler, follower target, pending spawn count | + +## Leader Election + +Triggered from `watch_workers` when no worker znode has `leader: true`. + +``` +for each candidate worker: + read snapshot offset from /t/snapshots/{cont_id} + track best by: highest offset, then lowest proc_id as tie-break + +set best_candidate["leader"] = True +zk.set("/t/" + best_name, updated_bytes) +``` + +The goal is to pick the replica least likely to be missing recent writes. `proc_id` tie-break is deterministic β€” oldest surviving worker wins ties. Election duration is measured and emitted as `leader_elected` with `duration_ms`. + +## Autoscaler + +`auto_scale()` runs on a configurable interval (default controlled by `SCALING_INTERVAL_SECONDS`). + +``` +delta = nginx_cumulative_requests - prev_sample +current_rps = delta / interval_seconds +required_followers = ceil(current_rps / rps_per_worker) # min 1 +``` + +It calls `_set_follower_target()`, which compares against live ZK follower count and spawns or trims accordingly. Trims remove the highest-`proc_id` followers first (newest workers go first). Both `scaling_interval_seconds` and `scaling_rps_per_worker` are adjustable at runtime via `POST /api/v1/config/scaling` without a redeploy β€” the scheduler job is rescheduled in-place. + +Nginx's cumulative counter resets on reload. `return_count()` handles this by treating a negative delta as `current_requests` (the new baseline after reset). + +## Failure Detection + +Nginx calls `/internal/upstream_signal` with `kind=failure` (immediate upstream error) or `kind=observe` (post-action non-2xx report). Both paths feed `_probe_and_handle_failure()`. + +The probe logic: +- Per-worker cooldown of 1.5 seconds (`FAILURE_PROBE_COOLDOWN_SEC`) +- Per-worker inflight set β€” only one probe per worker at a time +- Two `GET /health` attempts, 0.75s timeout each, with 150ms between +- On confirmed failure: emit `worker_failure_confirmed`, delete ZK znode, kill container + +Deleting the znode immediately triggers `ChildrenWatch`, which kicks off election and Nginx reconcile without waiting for ZooKeeper session expiry (typically 30–40s). + +## Kafka β†’ WebSocket Relay + +A background thread in `orch.py` consumes `dbaas.events` and fans out each message to all connected WebSocket clients via `asyncio.run_coroutine_threadsafe`. The orchestrator is the single aggregation point β€” workers and the workload generator both publish to `dbaas.events`; the frontend only needs one WebSocket connection. + +## Tradeoffs + +- Module-level shared state in `state.py` is straightforward but would need rethinking for multi-process or multi-instance deployments. +- Docker socket control is flexible but couples the orchestrator tightly to Docker. A Kubernetes deployment would swap `docker_utils.py` for a k8s client. +- Fast znode deletion speeds up failover but means the orchestrator, not ZooKeeper session expiry, is authoritative for membership cleanup. That's a deliberate choice β€” ZK session timeout is too slow for a good user experience. diff --git a/_docs/system/components/workers.md b/_docs/system/components/workers.md new file mode 100644 index 0000000..eeea8b1 --- /dev/null +++ b/_docs/system/components/workers.md @@ -0,0 +1,58 @@ +# Workers + +Every worker container runs `worker.py`, which registers a ZooKeeper znode and starts follower.py as a subprocess. If the orchestrator sets `leader: true` on that znode, `worker.py`'s `DataWatch` callback kills the follower process and starts `leader.py` in its place. The leader process runs until the container is killed; there's no in-process demotion path. + +## Leader + +`leader.py` is a FastAPI app serving `/write`. + +**Write path** (inside a single stripe lock): +1. Apply op to local LevelDB (`kv.apply_op(op)`) +2. Publish sync message to RabbitMQ `sync` fanout (persistent delivery, `delivery_mode=2`) +3. Publish op to Kafka `dbaas-writes` with `acks='all'`; record the returned offset +4. Return `{"status": "ok"}` + +Steps 2 and 3 happen inside the stripe lock, so apply + both publishes are atomic per-key. The RabbitMQ channel uses publisher confirms (`confirm_delivery()`). + +**Striped locks:** 64 stripes, keyed by `hash(key) % 64`. Same-key writes serialize; different keys that fall on different stripes run in parallel. `clear_all` has no key β€” it hashes to a fixed stripe, serializing it against itself and any key that maps to the same stripe. This is a simple way to get per-key ordering without a global write lock. + +**RabbitMQ connection:** `pika.BlockingConnection` is not thread-safe, so `_pika_lock` serializes all publishes (both the write threadpool and the snapshot background thread share the same connection). + +**Snapshot trigger:** every `SNAPSHOT_INTERVAL` writes (default 2000), `_snapshotter.trigger()` fires. The trigger is non-blocking β€” it starts a daemon thread if one isn't already running, otherwise drops the request. + +**Telemetry:** a background thread aggregates write count, average latency, and max Kafka offset every 5 seconds and emits a `telemetry_snapshot` event to `dbaas.events`. Prometheus metrics (`dbaas_writes_total`, `dbaas_write_latency_ms`, `dbaas_kafka_offset`) are exposed via `prometheus-fastapi-instrumentator`. + +## Follower + +`follower.py` is a FastAPI app serving `/read`. On startup it runs recovery, then starts the RabbitMQ sync consumer in a daemon thread. + +**Read ops:** `get`, `prefix_scan`, `count_prefix` β€” all served from local LevelDB. Response includes `X-Replica-Lag` header (milliseconds since the leader published the last sync message). Lag is measured from the `ts` field embedded in each sync message. + +**Sync consumer:** exclusive, auto-delete queue bound to the `sync` fanout exchange. `prefetch_count=50`. On snapshot signal (`SYNC_MSG_SNAPSHOT`), spawns a background thread to fetch + store the snapshot without blocking message consumption. + +**Dead-letter handling:** message failures are tracked by SHA1 of the message body. After `MAX_SYNC_RETRIES` (default 3) failed attempts, the message is nacked, published to a per-follower dead-letter queue (`sync.dead.{follower_id}`), and a `sync_message_dead_lettered` event is emitted. This prevents a single bad message from stalling replication indefinitely. + +## Snapshot Protocol + +**Leader side (`LeaderSnapshotter`):** +1. Fetch current end offset from Kafka (`end_offsets - 1`) +2. Write full LevelDB snapshot to disk (`latest`) + metadata (`latest.meta`: `kafka_offset`, `kafka_partition`, `ts`) +3. Set `/t/snapshots/latest_offset` in ZooKeeper +4. Publish `SYNC_MSG_SNAPSHOT` to RabbitMQ fanout so all followers get notified +5. Wait up to 30 seconds for all live followers to POST `/snapshot/ack` with their confirmed offset +6. GC: call Kafka `delete_records` up to the minimum confirmed offset across all acks (or snapshot offset if no followers acked in time) + +**Follower side (startup recovery in `follower_snapshot.py`):** +1. Resolve leader address from ZooKeeper +2. Read `/t/snapshots/latest_offset` from ZooKeeper +3. If local snapshot metadata has `kafka_offset >= zk_offset`: restore from local disk +4. Otherwise: `GET /snapshot/latest.meta` then `GET /snapshot/latest` from leader +5. Replay Kafka from `snapshot_offset + 1` to current end offset + +This means a freshly started follower can recover from a local snapshot if it's recent enough, skipping the leader fetch entirely. + +**Follower background snapshot fetch (`fetch_and_store`):** triggered by the `SYNC_MSG_SNAPSHOT` signal on the RabbitMQ sync stream. Fetches the snapshot, persists it locally, updates ZooKeeper, and POSTs an ack to the leader. + +## Storage + +LevelDB via `kv_store.py`. Op model: `put`, `delete`, `clear_all`. Snapshot format is a JSON dict of all KV pairs; restore is a full replace. Full snapshots are simple and correct, but expensive for large datasets β€” acceptable for the current scale, and a known limitation for larger ones. diff --git a/_docs/system/components/workload-generator.md b/_docs/system/components/workload-generator.md new file mode 100644 index 0000000..dd8bc63 --- /dev/null +++ b/_docs/system/components/workload-generator.md @@ -0,0 +1,28 @@ +# Workload Generator + +`WorkloadGenerator/` is a self-contained FastAPI service that wraps an embedded Locust environment. It exists so load-test concerns stay entirely out of the orchestrator. + +## Runtime Model + +Locust's `LocalRunner` runs inside the same process as the FastAPI app. Locust uses gevent; FastAPI/uvicorn uses asyncio. A `_pump_gevent` coroutine runs continuously during the lifespan, calling `gevent.sleep(0.05)` + `asyncio.sleep(0.05)` to keep both event loops ticking. A `threading.Lock` serializes `start`/`stop` transitions to prevent state corruption if requests arrive concurrently. + +## Workload Parameters + +| Parameter | Default | Notes | +|-----------|---------|-------| +| `concurrent_users` | 10 | Locust user count | +| `spawn_rate` | 5.0 | Users spawned per second | +| `read_ratio` | 80 | Percentage of ops that are reads (0–100) | +| `key_space` | 1000 | Number of distinct keys | +| `key_distribution` | `uniform` | `uniform`, `zipf`, or `hotspot` | +| `max_rps_per_user` | 5 | Per-user request cap; implemented as `wait_time = 1/max_rps_per_user` | + +`key_distribution` lets you test different access patterns against the autoscaler and replica lag behavior. `zipf` creates hot keys; `hotspot` concentrates writes on a small key subset. + +## Stop Behavior + +On `POST /api/jobs/stop`, the workload generator stops the Locust runner and immediately calls `POST /api/autoscale/reset` on the orchestrator. This resets the autoscaler's request counter baseline so the RPS delta from the load test doesn't carry over into the next scaling interval and cause unnecessary follower trimming. + +## Stats + +`GET /api/jobs/stats` returns per-endpoint Locust stats including `current_rps`, `avg_response_time`, `p95_response_time`, `p99_response_time`, and error counts. The frontend polls this to show live throughput and latency. diff --git a/_docs/system/decisions/major-decisions.md b/_docs/system/decisions/major-decisions.md new file mode 100644 index 0000000..8aa83ef --- /dev/null +++ b/_docs/system/decisions/major-decisions.md @@ -0,0 +1,71 @@ +# Major Decisions and Tradeoffs + +These are the non-obvious choices in the design. Each has a clear reason and a known cost. + +--- + +## 1. Control plane out of the request path + +The orchestrator never proxies reads or writes. It manages topology β€” ZooKeeper membership, Nginx config, worker lifecycle β€” and nothing else. Nginx talks directly to workers. + +The alternative (orchestrator as a smart proxy) would simplify topology discovery but puts the control plane in the latency-critical path. Every orchestrator restart or deployment would interrupt traffic. Keeping them separate means topology operations are independent of data operations. + +Cost: the routing layer (Nginx) has to stay in sync with what the orchestrator knows. That synchronization is the reconcile protocol β€” a source of bugs if it drifts. + +--- + +## 2. RabbitMQ for fast sync + Kafka for durable WAL + +Replication uses two channels with different goals. RabbitMQ fanout gives followers near-real-time updates under normal operation β€” consumers are already connected, so publish latency is low. Kafka provides the durable replay source that followers need on startup and that the leader needs to anchor snapshot GC. + +A single Kafka topic for both would work but followers would see higher replication lag (Kafka's consumer poll interval vs. RabbitMQ's push model). A single RabbitMQ fanout without Kafka would mean no durable log β€” a crashed follower with a stale snapshot has no way to catch up. + +Cost: two brokers to operate, two failure modes to reason about. A message published to RabbitMQ but not yet consumed by a follower when that follower crashes is lost β€” the follower recovers from the Kafka WAL instead, so this is correct, but it means follower recovery always involves a Kafka replay even if the lag was small. + +--- + +## 3. Leader election by snapshot offset, tie-break by proc_id + +When no leader exists, the orchestrator picks the follower with the highest snapshot offset recorded in ZooKeeper (`/t/snapshots/{cont_id}`). Ties go to the lowest `proc_id` (oldest surviving worker). + +This minimizes how many writes the new leader is missing relative to the previous leader. It's an approximation β€” snapshot offset is the last checkpoint, not the true write frontier β€” but it's the best available signal without a dedicated consensus log. In the worst case (leader crashed between snapshots), the new leader may be missing up to `SNAPSHOT_INTERVAL` writes that existed in the old leader's Kafka WAL. Those writes were already published to RabbitMQ and applied by followers, so followers may actually be ahead of the new leader temporarily. This is a known consistency gap in the current design. + +`proc_id` tie-break is deterministic and avoids split-brain during simultaneous election attempts (though ZooKeeper's sequential write semantics already prevent that). + +--- + +## 4. Nginx config versioning with a hash digest + +Config version is a 12-character SHA-256 hex digest over `{"leader": addr, "followers": sorted(addrs)}`. The reconcile sidecar skips the Nginx reload if the current version file matches the desired version. + +This makes reconcile idempotent. The orchestrator calls reconcile aggressively β€” on every `ChildrenWatch` event, on startup, on failure detection. Without versioning, every topology event would reload Nginx even if nothing changed, causing unnecessary connection drops. + +The hash is short (12 chars) purely for readability in logs. Collision probability is negligible for this use case. + +--- + +## 5. Reconcile sidecar rather than exec or file mount + +The original approach was to have the orchestrator write the Nginx config directly via a Docker volume mount. The reconcile sidecar (`reconcile.py`) replaced that: the orchestrator calls an HTTP endpoint, and the sidecar fetches and applies the config from inside the container. + +The sidecar owns the reload β€” it does an atomic file write (temp file β†’ `os.rename`) before calling `nginx -s reload`, so Nginx never reads a partial config. The orchestrator doesn't need filesystem access to the Nginx container. + +Cost: another process to keep alive, another endpoint to monitor. + +--- + +## 6. Orchestrator deletes failed worker znodes rather than waiting for session expiry + +ZooKeeper session expiry is the guaranteed cleanup path β€” ephemeral znodes disappear when the session times out. But the default timeout is 30–40 seconds, which is too long for acceptable failover UX. + +When the orchestrator confirms a failure via its health probe, it explicitly deletes the znode (`zk.delete("/t/worker...")`). This triggers `ChildrenWatch` immediately, which runs election and Nginx reconcile in ~1–2 seconds rather than 30+. + +The cost is that the orchestrator is now responsible for membership cleanup, not just ZooKeeper. If the orchestrator itself crashes between detecting the failure and deleting the znode, cleanup falls back to session expiry. This is acceptable β€” session expiry is a backstop, not the primary path. + +--- + +## 7. Single-partition Kafka WAL + +`dbaas-writes` uses a single partition. This gives strict write ordering for free β€” offset is a total order, snapshot GC is unambiguous, follower replay is a simple sequential seek. + +The ceiling is write throughput limited to what one partition can sustain. Partitioned WAL would require per-shard leaders, offset coordination across partitions, and more complex snapshot logic. That's the right direction for scaling writes horizontally, but it's a significant increase in complexity for a prototype. diff --git a/_docs/system/processes/architecture-and-layout.md b/_docs/system/processes/architecture-and-layout.md new file mode 100644 index 0000000..bf7d587 --- /dev/null +++ b/_docs/system/processes/architecture-and-layout.md @@ -0,0 +1,59 @@ +# Architecture and Folder Layout + +## Four Planes + +``` +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ Control Plane β”‚ +β”‚ Orchestrator/orch/ β€” ZK watcher, election, autoscaler, β”‚ +β”‚ failure detection, routing reconcile β”‚ +β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ +β”‚ Data Plane β”‚ +β”‚ Orchestrator/nginx/ β€” Nginx routes /write β†’ leader, β”‚ +β”‚ /read β†’ follower pool (round-robin) β”‚ +β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ +β”‚ Storage / Replication Plane β”‚ +β”‚ Orchestrator/worker/ β€” leader.py, follower.py, LevelDB, β”‚ +β”‚ Kafka WAL, RabbitMQ sync, snapshots β”‚ +β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ +β”‚ Operations Plane β”‚ +β”‚ WorkloadGenerator/ β€” Locust load tests β”‚ +β”‚ frontend/ β€” operator console β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +``` + +The control plane never touches the request path. The orchestrator spawns and kills workers, manages ZooKeeper membership, and pushes new Nginx configs β€” but `/write` and `/read` traffic never flows through it. + +## Folder Layout + +``` +Orchestrator/ + orch/ control-plane service (orch.py, zk_watcher.py, scaling.py, + β”‚ failure_detection.py, docker_utils.py, nginx_config.py, state.py) + worker/ worker entrypoint + leader/follower runtimes + β”‚ (worker.py, leader.py, follower.py, kv_store.py, + β”‚ leader_snapshot.py, follower_snapshot.py, snapshot.py) + nginx/ Nginx image + reconcile sidecar (reconcile.py) +WorkloadGenerator/ + app.py FastAPI wrapper + Locust lifecycle + workload.py WorkloadUser task logic, key distribution +frontend/ + src/ + hooks/ useTopology.js, useWebSocket.js + pages/ Workload.jsx + components/ TopologyGraph, EventLog, ScalingConfig, MetricsPanel, ... +``` + +## Runtime Contracts + +| What | Where | +|------|-------| +| Worker registrations | ZooKeeper `/t/worker*` (ephemeral sequential) | +| Snapshot offsets | ZooKeeper `/t/snapshots/{cont_id}`, `/t/snapshots/latest_offset` | +| WAL | Kafka topic `dbaas-writes`, partition 0 | +| Control/telemetry events | Kafka topic `dbaas.events` | +| Fast replication | RabbitMQ `sync` fanout exchange | +| Nginx config delivery | Orchestrator `/internal/nginx/config` β†’ reconcile sidecar | +| Failure signaling | Nginx `@upstream_failure` β†’ orchestrator `/internal/upstream_signal` | + +Workers register under `/t/worker` with `{cont_id, proc_id, leader: false}`. The orchestrator sets `leader: true` on the elected node; worker.py's `DataWatch` picks that up and kills the follower subprocess, starting leader.py in its place. diff --git a/_docs/system/processes/data-path-and-recovery.md b/_docs/system/processes/data-path-and-recovery.md new file mode 100644 index 0000000..f807c1b --- /dev/null +++ b/_docs/system/processes/data-path-and-recovery.md @@ -0,0 +1,82 @@ +# Data Path and Recovery + +## Write Path + +``` +Client + β†’ POST /write β†’ Nginx + β†’ leader_upstream (single server) + β†’ Leader: + stripe_lock(hash(key) % 64): + kv.apply_op(op) # LevelDB + rabbitmq.publish(sync) # RabbitMQ fanout, delivery_mode=2 + kafka.send(dbaas-writes, acks='all') # returns offset + β†’ {"status": "ok"} +``` + +All three operations (LevelDB write, RabbitMQ publish, Kafka publish) happen inside the stripe lock. The lock scope is intentional β€” it ensures that for any given key, the apply and both publishes are ordered consistently. Different keys that land on different stripes run in parallel. + +## Read Path + +``` +Client + β†’ GET /read?query=... β†’ Nginx + β†’ follower_upstream (round-robin across followers) + β†’ Follower: + kv.get / kv.prefix_scan / kv.count_prefix # LevelDB read + β†’ {"result": ...} + X-Replica-Lag: # time since last sync message from leader +``` + +Reads are eventually consistent. The `X-Replica-Lag` header tells the client how stale the follower may be. + +## Follower Replication (Steady State) + +Followers consume the RabbitMQ `sync` fanout exchange via exclusive auto-delete queues. Each sync message from the leader contains the write op plus a `ts` timestamp. The follower applies the op to its local LevelDB and records the lag. This is the fast path β€” under normal conditions, replication lag is in the low milliseconds. + +## Snapshot and WAL GC + +The leader snapshots every `SNAPSHOT_INTERVAL` writes (default 2000). The snapshot process: +1. Fetch current Kafka end offset +2. Write full LevelDB state to `latest` + metadata to `latest.meta` (includes `kafka_offset`, `kafka_partition`, timestamp) +3. Set `/t/snapshots/latest_offset` in ZooKeeper +4. Publish snapshot signal to RabbitMQ fanout +5. Wait up to 30 seconds for live followers to ack via `POST /snapshot/ack` +6. Kafka `delete_records` up to `min(confirmed acks)`, keeping WAL bounded + +If followers don't all ack within 30 seconds, GC proceeds up to the minimum of whatever acks arrived. Missing followers (crashed between snapshot signal and ack) are excluded. + +## Follower Startup Recovery + +On startup, `follower_snapshot.py::recover()` runs before the HTTP server starts accepting traffic: + +``` +1. Resolve leader address from ZooKeeper (/t/worker* where leader=true) +2. Read /t/snapshots/latest_offset from ZooKeeper +3. If local snapshot metadata kafka_offset >= zk_offset: + restore from local disk (no leader fetch) + Else: + GET /snapshot/latest.meta from leader + GET /snapshot/latest from leader + write to local disk +4. Replay Kafka from snapshot_offset + 1 to current end offset +``` + +A follower that recently crashed and has a fresh local snapshot skips the leader fetch entirely β€” only the delta from the WAL needs replaying. A brand-new follower (no local snapshot) fetches the full snapshot from the leader and replays from there. + +## Failure Scenario: Leader Crash + +This is what happens end-to-end when the leader container dies: + +1. **Nginx** receives a 502/503 on `/write`. The `@upstream_failure` internal location fires. +2. **Nginx** calls `GET /internal/upstream_signal?kind=failure&upstream=&...` on the orchestrator. +3. **Orchestrator** (`failure_detection.py`) resolves the upstream address to a container ID via ZooKeeper, checks the per-worker cooldown (1.5s) and inflight set, then starts a probe thread. +4. **Probe thread** sends two `GET /health` requests (0.75s timeout, 150ms apart). Both fail. +5. **Orchestrator** emits `worker_failure_confirmed`, deletes the leader's ZK znode, kills the container. +6. **ZooKeeper** `ChildrenWatch` fires in `zk_watcher.py`. No `leader=true` node exists. +7. **Election runs**: candidates are ranked by snapshot offset, then proc_id. Best candidate's znode is updated to `leader: true`. +8. **Elected worker**: `worker.py`'s `DataWatch` callback fires, kills follower.py, starts leader.py. +9. **Orchestrator** calls Nginx reconcile. The reconcile sidecar fetches new config (new leader address), writes it atomically, reloads Nginx. +10. **Nginx** now routes `/write` to the new leader. Total downtime is typically a few seconds. + +Meanwhile, if the follower count drops below target (because the crashed node was also counted), `watch_workers` spawns a replacement. diff --git a/_docs/system/processes/scaling-and-failover.md b/_docs/system/processes/scaling-and-failover.md new file mode 100644 index 0000000..46d4518 --- /dev/null +++ b/_docs/system/processes/scaling-and-failover.md @@ -0,0 +1,57 @@ +# Scaling and Failover + +## Two Control Loops + +The system uses two independent loops that both affect follower count. Understanding their interaction is important. + +**Loop 1 β€” Autoscaler** (`scaling.py`, interval-driven): +- Reads Nginx cumulative request count from `/api/stats` (`stub_status`) +- Computes delta from previous sample to get request count over the interval +- `current_rps = delta / interval_seconds` +- `required_followers = ceil(current_rps / rps_per_worker)`, floor at 1 +- Calls `_set_follower_target()` which spawns or trims to match the target + +**Loop 2 β€” ZK Watcher** (`zk_watcher.py`, event-driven): +- Fires on every ZooKeeper `/t` children change (worker join or leave) +- Recomputes current leader/follower counts +- Spawns if `len(current_workers) + pending_spawns < req_followers + 1` +- Trims if excess followers exist after a target reduction +- Triggers election if no leader exists +- Calls `reconcile_nginx_upstream` unconditionally on every topology change + +Loop 2 is the recovery path. If a worker crashes between autoscaler ticks, the ZK watcher fires immediately and restores the count without waiting for the next scheduler interval. + +## `pending_spawns` Coordination + +Both loops share `state.pending_spawns`. When the orchestrator calls `spawn_worker()`, it increments `pending_spawns` before the container comes up. When the ZK watcher sees a new worker registration, it decrements `pending_spawns` by the number of new arrivals. + +This prevents both loops from deciding independently that "not enough workers exist" and double-spawning. Without it, a crash during an autoscale-up event could cause a spawn storm. + +## Autoscaler Reset on Workload Stop + +When the workload generator stops a job, it calls `POST /api/autoscale/reset` on the orchestrator. This resets `prev_request_count` to the current Nginx counter and sets `req_followers` back to 1. Without this, the high RPS from the load test would inflate the next autoscaler delta calculation, causing a spurious scale-down that trims followers that should stay alive. + +## Scaling Config at Runtime + +`POST /api/v1/config/scaling` updates `scaling_interval_seconds` and `scaling_rps_per_worker` in `state.py` and reschedules the APScheduler job with the new interval in-place. No restart required. This is useful when tuning sensitivity β€” a short interval reacts faster but creates more churn; a high `rps_per_worker` target keeps the cluster smaller. + +## Failure Recovery Timeline + +When a worker crashes: + +``` +0ms Nginx detects upstream failure, calls /internal/upstream_signal +~10ms Orchestrator probe thread starts +~1.5s Two health probes complete (750ms timeout Γ— 2, 150ms gap) +~1.5s ZK znode deleted, container killed +~1.5s ChildrenWatch fires +~1.5s Election runs (if leader crashed) +~2s New leader.py starts, Nginx reconcile triggered +~2-3s Nginx config updated, traffic routes to new leader +``` + +Compare to ZooKeeper session expiry alone: typically 30–40 seconds. The Nginx callback + active probe path cuts observed downtime by an order of magnitude. + +## Follower Crash (No Election Needed) + +If a follower crashes, ZK `ChildrenWatch` fires, compares current count to `req_followers + 1`, and spawns a replacement. The new follower runs startup recovery (local snapshot or leader fetch + Kafka replay) before starting to serve reads. Nginx reconcile removes the dead follower from the upstream pool. diff --git a/docker-compose-local.yml b/docker-compose-local.yml index df6e2be..709990e 100644 --- a/docker-compose-local.yml +++ b/docker-compose-local.yml @@ -161,6 +161,8 @@ services: container_name: grafana environment: - GF_SECURITY_ADMIN_PASSWORD=dbaas + - GF_AUTH_ANONYMOUS_ENABLED=true + - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin - GF_LOG_LEVEL=warn - GF_ANALYTICS_CHECK_FOR_UPDATES=false - GF_ANALYTICS_REPORTING_ENABLED=false diff --git a/frontend/src/components/MetricsPanel.jsx b/frontend/src/components/MetricsPanel.jsx index 0a79457..a24ef7d 100644 --- a/frontend/src/components/MetricsPanel.jsx +++ b/frontend/src/components/MetricsPanel.jsx @@ -35,7 +35,7 @@ function useAllMetrics(events, nowMs) { }; } const b = buckets[bucketKey]; - + const wCount = Number(e.count ?? 0); const rCount = Number(e.read_count ?? 0); const wLat = Number(e.avg_latency_ms ?? 0); @@ -69,13 +69,13 @@ function useAllMetrics(events, nowMs) { datasets.writes.push({ time: label, value: b ? (b.writes / BUCKET_SEC) : 0 }); datasets.reads.push({ time: label, value: b ? (b.reads / BUCKET_SEC) : 0 }); - datasets.writeLat.push({ - time: label, - value: (b?.writeLatCount > 0) ? Number((b.writeLatSum / b.writeLatCount).toFixed(2)) : 0 + datasets.writeLat.push({ + time: label, + value: (b?.writeLatCount > 0) ? Number((b.writeLatSum / b.writeLatCount).toFixed(2)) : 0 }); - datasets.readLat.push({ - time: label, - value: (b?.readLatCount > 0) ? Number((b.readLatSum / b.readLatCount).toFixed(2)) : 0 + datasets.readLat.push({ + time: label, + value: (b?.readLatCount > 0) ? Number((b.readLatSum / b.readLatCount).toFixed(2)) : 0 }); } return datasets; @@ -99,10 +99,10 @@ export function MetricsPanel({ events }) { - val.toFixed(val > 0 && val < 5 ? 1 : 0)} + val.toFixed(val > 0 && val < 5 ? 1 : 0)} /> @@ -119,4 +119,4 @@ export function MetricsPanel({ events }) { {renderChart('Read Latency avg (ms)', readLat, 'var(--color-follower)')} ); -} +}; diff --git a/frontend/src/hooks/useTopology.js b/frontend/src/hooks/useTopology.js index 1f22455..65ad360 100644 --- a/frontend/src/hooks/useTopology.js +++ b/frontend/src/hooks/useTopology.js @@ -17,7 +17,7 @@ export function workerIdsMatch(a, b) { } const isLeaderRole = (w) => ( - w.is_leader || + w.is_leader || ['leader', 'master'].includes(String(w.role || w.type).toLowerCase()) ); @@ -36,7 +36,7 @@ export function useTopology(url, events = []) { const newNodes = []; const newEdges = []; - + // Identify leader first for edge links const leader = workers.find(isLeaderRole); const leaderId = leader ? normalizeWorkerIdentity(leader) : null; @@ -44,8 +44,7 @@ export function useTopology(url, events = []) { workers.forEach(w => { const wId = normalizeWorkerIdentity(w); const isLeader = isLeaderRole(w); - - // Determine status based on active patches and reported status + let status = w.status || 'healthy'; if (crashingIds.current.has(wId)) status = 'crashing'; else if (recoveringIds.current.has(wId)) status = 'recovering'; @@ -81,7 +80,6 @@ export function useTopology(url, events = []) { const contId = latest.cont_id || (latest.source?.startsWith('follower-') ? latest.source.slice(9) : null); const handleEvent = async () => { - // Patch local state for immediate visual feedback if (type === 'worker_crashing' && contId) { crashingIds.current.add(contId); setNodes(prev => prev.map(n => workerIdsMatch(n.id, contId) ? { ...n, status: 'crashing' } : n)); @@ -97,10 +95,10 @@ export function useTopology(url, events = []) { // Refresh topology for structural changes const refreshSignals = [ - 'leader_elected', - 'worker_joined', 'follower_spawned', - 'worker_terminated', 'follower_terminated', - 'recovery_complete' + 'leader_elected', + 'worker_joined', 'follower_spawned', 'slave_spawned', + 'worker_terminated', 'follower_terminated', 'slave_terminated', + 'recovery_complete', ]; if (refreshSignals.includes(type)) {