Docs

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:

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

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)

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;

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", "")

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"])

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