diff --git a/simplyblock_core/cluster_ops.py b/simplyblock_core/cluster_ops.py index 6c86cbf33..88983a2bb 100755 --- a/simplyblock_core/cluster_ops.py +++ b/simplyblock_core/cluster_ops.py @@ -614,6 +614,27 @@ def set_name(cl_id, name) -> Cluster: def cluster_activate(cl_id, force=False, force_lvstore_create=False) -> None: cluster = db_controller.get_cluster_by_id(cl_id) + prev_status = cluster.status + if prev_status == Cluster.STATUS_IN_ACTIVATION: + prev_status = Cluster.STATUS_UNREADY + try: + _cluster_activate(cl_id, force=force, force_lvstore_create=force_lvstore_create) + except Exception: + # Never leave the cluster wedged in in_activation: this often runs in + # a fire-and-forget thread, and an unhandled failure would otherwise + # block any retry (the activate API rejects in_activation clusters). + # The expected-failure paths inside _cluster_activate restore the + # status themselves; this only catches what they missed. + cluster = db_controller.get_cluster_by_id(cl_id) + if cluster.status == Cluster.STATUS_IN_ACTIVATION: + logger.error("Cluster activation failed unexpectedly; reverting status " + f"from {Cluster.STATUS_IN_ACTIVATION} to {prev_status}") + set_cluster_status(cl_id, prev_status) + raise + + +def _cluster_activate(cl_id, force=False, force_lvstore_create=False) -> None: + cluster = db_controller.get_cluster_by_id(cl_id) if cluster.status == Cluster.STATUS_ACTIVE: logger.warning("Cluster is ACTIVE") @@ -757,8 +778,13 @@ def cluster_activate(cl_id, force=False, force_lvstore_create=False) -> None: set_cluster_status(cl_id, ols_status) raise ValueError("Failed to activate cluster") else: - ret = storage_node_ops.create_lvstore(snode, cluster.distr_ndcs, cluster.distr_npcs, cluster.distr_bs, - cluster.distr_chunk_bs, cluster.page_size_in_blocks, max_size) + try: + ret = storage_node_ops.create_lvstore(snode, cluster.distr_ndcs, cluster.distr_npcs, cluster.distr_bs, + cluster.distr_chunk_bs, cluster.page_size_in_blocks, max_size) + except Exception as e: + logger.error(e) + set_cluster_status(cl_id, ols_status) + raise ValueError("Failed to activate cluster") snode = db_controller.get_storage_node_by_id(snode.get_id()) if ret: snode.lvstore_status = "ready" @@ -985,6 +1011,18 @@ def cluster_activate(cl_id, force=False, force_lvstore_create=False) -> None: logger.error("Pass 4: set non_optimized ANA on %s for %s failed: %s", sec_node.get_id(), lv.nqn, e) + # The cluster is now active and about to serve IO. During node-add the + # storage MCP was created wide (= parallel-add count) so the first-time + # CPU-topology reboots happened in one parallel wave rather than a + # serialized queue. Now narrow it to the cluster's fault tolerance so any + # future MachineConfig/KubeletConfig rollout never reboots more storage + # nodes at once than the data plane can absorb. Done here (success path, + # before flipping to ACTIVE) so a failed/aborted activation leaves the + # cluster non-serving with the wide value — harmless, since no data is at + # risk until it goes ACTIVE. (Use max_fault_tolerance - 1 instead if you + # want headroom for an unplanned failure concurrent with a rollout.) + utils.set_storage_mcp_max_unavailable(cl_id, cluster.max_fault_tolerance) + set_cluster_status(cl_id, Cluster.STATUS_ACTIVE) logger.info("Cluster activated successfully") diff --git a/simplyblock_core/constants.py b/simplyblock_core/constants.py index e6288e1c6..d5e4628ee 100644 --- a/simplyblock_core/constants.py +++ b/simplyblock_core/constants.py @@ -100,12 +100,19 @@ def get_config_var(name, default=None): # A JobSchedule's lease is held by the runner host (by hostname) that last # touched it. Another host may only take over a task whose lease is older than -# this — i.e. the owning runner is presumed dead. Must exceed the longest -# single task_runner() invocation that does not write the task back (the -# restart runner can block on RPCs for several minutes), so a live owner is -# never falsely preempted. A runner restarting on the SAME host re-claims its -# own tasks immediately regardless of this value (owner id is the hostname). -TASK_LEASE_TTL_SEC = 1200 +# this — i.e. the owning runner is presumed dead. A live owner refreshes the +# lease every TASK_LEASE_HEARTBEAT_SEC from a background thread while driving +# a restart (restart_storage_node wrapper), so the TTL no longer needs to +# exceed the longest blocking RPC — it only needs to be several heartbeats +# wide so a momentarily slow (but alive) owner is never falsely preempted. +# Keeping it short is what makes ownership transfer fast: when the driving +# process dies (pod evicted while its host drains, CLI killed), a live +# tasks-runner claims the stale lease and resumes the restart instead of the +# node staying orphaned in RESTARTING (2026-07-04 MCO rollout deadlock). +# A runner restarting on the SAME host re-claims its own tasks immediately +# regardless of this value (owner id is the hostname). +TASK_LEASE_HEARTBEAT_SEC = 30 +TASK_LEASE_TTL_SEC = 180 # Node-add concurrency: the cross-node mesh section of add_node is serialized # per cluster behind a ClusterAddNodeLock. The holder refreshes the lock every diff --git a/simplyblock_core/controllers/tasks_controller.py b/simplyblock_core/controllers/tasks_controller.py index ece14cccb..d26bea511 100644 --- a/simplyblock_core/controllers/tasks_controller.py +++ b/simplyblock_core/controllers/tasks_controller.py @@ -1,7 +1,9 @@ # coding=utf-8 +import contextlib import datetime import logging import socket +import threading import time import uuid @@ -69,6 +71,82 @@ def _mutate(t): return decision["won"] +def refresh_task_lease(task, owner=None): + """Heartbeat: refresh this host's lease on a task it already owns, so a + live owner is never preempted while blocking on long RPCs. Returns False + (without touching the task) if the task is done or owned by another host — + the caller lost the lease and should treat the takeover as authoritative.""" + owner = owner or _RUNNER_HOST + now = str(datetime.datetime.now(datetime.timezone.utc)) + refreshed = {"ok": False} + + def _mutate(t): + if t.status == JobSchedule.STATUS_DONE: + return False + if t.owner != owner: + return False + t.updated_at = now + refreshed["ok"] = True + return True + + if db.atomic_update(task, _mutate) is None: + return False + return refreshed["ok"] + + +@contextlib.contextmanager +def task_lease_heartbeat(task, owner=None): + """Refresh this host's lease on `task` every TASK_LEASE_HEARTBEAT_SEC for + the duration of the with-block. + + Every runner that executes long-blocking work under a claimed lease MUST + wrap that work in this: since TASK_LEASE_TTL_SEC (180s) is far shorter + than a node add / restart / migration, a lease that is only refreshed on + task writes goes stale mid-execution, and a second runner host (e.g. the + new pod during a rolling update) would claim the task and double-drive + it — for node-add that means killing the in-flight add's SPDK and + deleting its half-created node record. + + The heartbeat stops on its own if the lease is lost to another host + (refresh_task_lease returns False) — the takeover is authoritative. + """ + stop = threading.Event() + + def _beat(): + while not stop.wait(constants.TASK_LEASE_HEARTBEAT_SEC): + try: + if not refresh_task_lease(task, owner): + return + except Exception as e: + logger.debug(f"Lease heartbeat failed for task {task.uuid}: {e}") + + thread = threading.Thread(target=_beat, daemon=True) + thread.start() + try: + yield + finally: + stop.set() + + +def ensure_node_restart_task(node): + """Return the id of an unfinished FN_NODE_RESTART task for this node, + creating one if none exists. Used by explicit restarts + (restart_storage_node wrapper) to make their ownership transferable: the + driver claims and heartbeats the task's lease, and if the driving process + dies mid-restart, a live tasks-runner claims the stale lease and resumes. + + Deliberately bypasses the auto_restart_disabled guard of + add_node_to_auto_restart: that flag means "no UNATTENDED restart after a + deliberate shutdown" — an explicit restart is precisely the operator + intervention the flag waits for, and this task only continues that + expressed intent. On success the ONLINE transition cancels the task + (cancel_pending_node_restart_tasks via set_node_status).""" + existing = _validate_new_task_node_restart(node.cluster_id, node.get_id()) + if existing: + return existing + return _add_task(JobSchedule.FN_NODE_RESTART, node.cluster_id, node.get_id(), "", max_retry=11) + + def _validate_new_task_dev_restart(cluster_id, node_id, device_id): tasks = db.get_job_tasks(cluster_id) for task in tasks: diff --git a/simplyblock_core/services/tasks_runner_cluster_expand.py b/simplyblock_core/services/tasks_runner_cluster_expand.py index 058a16ca5..b4f3fe53b 100644 --- a/simplyblock_core/services/tasks_runner_cluster_expand.py +++ b/simplyblock_core/services/tasks_runner_cluster_expand.py @@ -125,7 +125,8 @@ def main(): f"Cluster-expand task {task.uuid} owned by " f"another runner host; skipping") break - res = process_task(task) + with tasks_controller.task_lease_heartbeat(task): + res = process_task(task) if res: if task.status == JobSchedule.STATUS_DONE: break diff --git a/simplyblock_core/services/tasks_runner_lvol_migration.py b/simplyblock_core/services/tasks_runner_lvol_migration.py index 55af8b98f..98dbe29a2 100644 --- a/simplyblock_core/services/tasks_runner_lvol_migration.py +++ b/simplyblock_core/services/tasks_runner_lvol_migration.py @@ -2258,6 +2258,7 @@ def _fail_task(task, migration_or_msg, reason=None): if not tasks_controller.claim_task(task): logger.info(f"LVol-migration task {task.uuid} owned by another runner host; skipping") continue - task_runner(task) + with tasks_controller.task_lease_heartbeat(task): + task_runner(task) time.sleep(3) diff --git a/simplyblock_core/services/tasks_runner_migration.py b/simplyblock_core/services/tasks_runner_migration.py index 3eb8f362c..8592f9beb 100644 --- a/simplyblock_core/services/tasks_runner_migration.py +++ b/simplyblock_core/services/tasks_runner_migration.py @@ -293,7 +293,8 @@ def _set_master_task_status(master_task, status): if not tasks_controller.claim_task(task): logger.info(f"Migration task {task.uuid} owned by another runner host; skipping") continue - res = task_runner(task) + with tasks_controller.task_lease_heartbeat(task): + res = task_runner(task) update_master_task(task) if res: node_task = tasks_controller.get_active_node_tasks(task.cluster_id, task.node_id) diff --git a/simplyblock_core/services/tasks_runner_node_add.py b/simplyblock_core/services/tasks_runner_node_add.py index 4eae2c6c5..52a2db0f5 100644 --- a/simplyblock_core/services/tasks_runner_node_add.py +++ b/simplyblock_core/services/tasks_runner_node_add.py @@ -1,4 +1,5 @@ # coding=utf-8 +import socket import threading import time from concurrent.futures import ThreadPoolExecutor @@ -71,6 +72,54 @@ def process_task(task, cl): return False +# Applying the CPU topology during add_node makes the node reboot +# (kubeletconfig / MCP update). The in-flight attempt then fails, but the +# right reaction is neither a quick blind retry (the node is down for +# 5-8 minutes; each attempt burns one of max_retry) nor exponential backoff +# (which keeps sleeping long after the node is back). Bound how long we are +# willing to wait for the node's agent to answer again — matches the +# topology job's own reboot budget (sleep 900). +NODE_REBOOT_WAIT_MAX_SEC = 900 +NODE_REBOOT_POLL_SEC = 15 + + +def _node_api_reachable(task, timeout=5): + """TCP-level reachability of the node agent (host:port from the task's + node_addr). During add the StorageNode record may not exist yet, so this + intentionally checks the address, not the DB object.""" + addr = (task.function_params or {}).get("node_addr", "") + if ":" not in addr: + return True # can't tell — let the normal retry path decide + host, _, port = addr.rpartition(":") + try: + with socket.create_connection((host, int(port)), timeout=timeout): + return True + except Exception: + return False + + +def _wait_node_reachable(task, task_uuid): + """After a failed attempt, if the node is unreachable (rebooting for the + CPU-topology change), wait for it to answer again — up to + NODE_REBOOT_WAIT_MAX_SEC — instead of consuming retries against a node + that cannot possibly respond. Returns True if a wait took place.""" + if _node_api_reachable(task): + return False + logger.info( + f"Node-add task {task_uuid}: node agent unreachable (rebooting for " + f"CPU topology?); waiting up to {NODE_REBOOT_WAIT_MAX_SEC}s for it to return") + deadline = time.time() + NODE_REBOOT_WAIT_MAX_SEC + while time.time() < deadline: + time.sleep(NODE_REBOOT_POLL_SEC) + if _node_api_reachable(task): + logger.info(f"Node-add task {task_uuid}: node agent reachable again; retrying add") + return True + logger.warning( + f"Node-add task {task_uuid}: node agent still unreachable after " + f"{NODE_REBOOT_WAIT_MAX_SEC}s; resuming normal retry schedule") + return True + + def _run_task(task_uuid, cluster_id): """Worker thread: drive one node-add task to completion (or suspension), then drop it from the in-flight set so a later cycle can retry it. @@ -92,11 +141,32 @@ def _run_task(task_uuid, cluster_id): if not tasks_controller.claim_task(task): logger.info(f"Node-add task {task_uuid} owned by another runner host; skipping") break - res = process_task(task, cl) + # add_node blocks for many minutes with no task writes; heartbeat + # the lease so another runner host never sees it stale mid-add. + retry_before = task.retry + with tasks_controller.task_lease_heartbeat(task): + res = process_task(task, cl) if res: if task.status == JobSchedule.STATUS_DONE: break - else: + # Reboot-aware handling: an attempt that failed because the node + # went down (CPU-topology reboot) is expected, not a real failure. + # add_node catches the interrupted spdk_process_start and RETURNS + # False, so process_task already consumed a retry — roll it back so + # the one guaranteed topology reboot per node doesn't eat the + # retry budget. Then wait for the agent to answer and retry + # promptly on a fresh schedule (no blind fast-retry, no runaway + # backoff). The re-run is idempotent: add_node cleans up its own + # stale IN_CREATION record on re-entry. + if _wait_node_reachable(task, task_uuid): + if task.retry > retry_before: + task.retry = retry_before + if task.status != JobSchedule.STATUS_DONE: + task.status = JobSchedule.STATUS_SUSPENDED + task.write_to_db(db.kv_store) + delay_seconds = constants.TASK_EXEC_INTERVAL_SEC + continue + if not res: # Cap the exponential backoff so a permanently failing node-add # can't grow the sleep without bound. delay_seconds = min( diff --git a/simplyblock_core/services/tasks_runner_port_allow.py b/simplyblock_core/services/tasks_runner_port_allow.py index f14e6d563..e76cafb68 100644 --- a/simplyblock_core/services/tasks_runner_port_allow.py +++ b/simplyblock_core/services/tasks_runner_port_allow.py @@ -1084,7 +1084,8 @@ def _main(): if not tasks_controller.claim_task(task): logger.info(f"Port-allow task {task.uuid} owned by another runner host; skipping") continue - exec_port_allow_task(task) + with tasks_controller.task_lease_heartbeat(task): + exec_port_allow_task(task) time.sleep(5) diff --git a/simplyblock_core/services/tasks_runner_restart.py b/simplyblock_core/services/tasks_runner_restart.py index 0cb4806ad..a80f9285d 100644 --- a/simplyblock_core/services/tasks_runner_restart.py +++ b/simplyblock_core/services/tasks_runner_restart.py @@ -488,6 +488,82 @@ def _restart_backoff_seconds(retry): return min(exp, constants.RESTART_TASK_EXEC_INTERVAL_MAX_SEC) +# Watchdog for orphaned transitional states. A node whose restart/shutdown +# flow is interrupted (this runner's pod evicted mid-restart during a node +# drain, node crash, ...) is left in STATUS_RESTARTING / STATUS_IN_SHUTDOWN +# with no pending task and no live process owning the transition. Those +# states are locked against outside writers (set_node_status) and the only +# sanctioned cleanup, _reset_if_transient, runs solely while a task for that +# node is being processed — so an ownerless node is wedged forever. The k8s +# operator's nodedrain controller then holds its drain slot waiting for the +# node to come online, deadlocking MachineConfig rollouts cluster-wide +# (incident 2026-07-04: every MCO reboot wedged the rollout until the node +# was manually reset). +# +# First-seen tracking is in-memory: a runner restart resets the clock, which +# only delays recovery by one grace period. Two grace tiers: when the node's +# SPDK pod is absent, nothing can be mid-flight on the data plane and we +# recover fast; when a pod exists, an unseen foreground CLI restart (which +# holds no task and looks ownerless to this check) may be driving it, and +# resetting under it would kill the SPDK it just started — so wait long +# enough for any legitimate restart to finish. +_transitional_first_seen: dict = {} +ORPHANED_STATE_GRACE_SEC = 20 * 60 +ORPHANED_STATE_FAST_GRACE_SEC = 5 * 60 + + +def _spdk_pod_exists(node): + """Whether the node's SPDK pod exists (kubernetes mode). Used only to + pick the watchdog grace tier — on any doubt return True so the + conservative (long) tier applies.""" + try: + cluster = db.get_cluster_by_id(node.cluster_id) + if cluster.mode != "kubernetes": + return True + utils.load_kube_config_with_fallback() + from kubernetes import client as k8s_client + namespace = getattr(node, "cr_namespace", "") or constants.K8S_NAMESPACE + prefix = f"snode-spdk-pod-{node.rpc_port}-" + for pod in k8s_client.CoreV1Api().list_namespaced_pod(namespace=namespace).items: + if pod.metadata.name.startswith(prefix): + return True + return False + except Exception as e: + logger.debug(f"SPDK pod lookup failed for {node.get_id()}: {e}") + return True + + +def _watchdog_orphaned_transitional_nodes(cluster_id): + """Detect nodes stuck in a transitional CP state with no restart task + owning them, and route them through the sanctioned recovery: verify the + data plane is down, reset to OFFLINE (_reset_if_transient), then queue a + normal auto-restart task.""" + for node in db.get_storage_nodes_by_cluster_id(cluster_id): + node_id = node.get_id() + if node.status not in (StorageNode.STATUS_RESTARTING, StorageNode.STATUS_IN_SHUTDOWN): + _transitional_first_seen.pop(node_id, None) + continue + # An unfinished restart task owns this state; its own flow calls + # _reset_if_transient when appropriate. + if not _validate_no_task_node_restart(cluster_id, node_id): + _transitional_first_seen.pop(node_id, None) + continue + first_seen = _transitional_first_seen.setdefault(node_id, time.time()) + elapsed = time.time() - first_seen + grace = ORPHANED_STATE_GRACE_SEC if _spdk_pod_exists(node) else ORPHANED_STATE_FAST_GRACE_SEC + if elapsed < grace: + continue + logger.warning( + f"Node {node_id} stuck in {node.status} for {int(elapsed)}s with no " + f"restart task owning it; attempting reset to OFFLINE") + _reset_if_transient(node_id) + node = db.get_storage_node_by_id(node_id) + if node.status == StorageNode.STATUS_OFFLINE: + _transitional_first_seen.pop(node_id, None) + if tasks_controller.add_node_to_auto_restart(node): + logger.info(f"Queued auto-restart for recovered node {node_id}") + + logger.info("Starting Tasks runner...") while True: try: @@ -544,7 +620,11 @@ def _restart_backoff_seconds(retry): logger.info(f"Restart task {task.uuid} owned by another runner host; skipping") continue retry_before = task.retry - res = task_runner(task) + # Device restarts (and parts of node restarts outside the + # restart_storage_node wrapper) block without task writes; + # heartbeat the lease so it never goes stale mid-execution. + with tasks_controller.task_lease_heartbeat(task): + res = task_runner(task) task = db.get_task_by_id(task.uuid) if res or task.status == JobSchedule.STATUS_DONE: _restart_next_attempt.pop(task.uuid, None) @@ -566,4 +646,9 @@ def _restart_backoff_seconds(retry): _restart_next_attempt[task.uuid] = ( time.time() + _restart_backoff_seconds(task.retry)) + try: + _watchdog_orphaned_transitional_nodes(cl.get_id()) + except Exception as e: + logger.error(f"Orphaned-node watchdog failed for cluster {cl.get_id()}: {e}") + time.sleep(constants.TASK_EXEC_INTERVAL_SEC) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 44351b056..5f85440d7 100755 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -1901,6 +1901,34 @@ def add_node(cluster_id, node_addr, iface_name, data_nics_list, start_storage_node_api_container(mgmt_ip, cluster_ip) node_socket = node_config.get("socket") + # Idempotent re-entry. A prior add attempt for this (node, socket) may + # have been interrupted — most commonly because the node rebooted for + # the CPU-topology change and killed the agent serving the add, or + # because the tasks-runner pod driving the add was itself co-located on + # a rebooting storage node and died mid-flight. That leaves a + # StorageNode stuck in IN_CREATION. The lease-based takeover re-runs the + # add, so this must start from a clean slate: kill any half-started + # SPDK and drop the stale record. Without this the retry builds a + # DUPLICATE node (observed 2026-07-06) and the total_mem loop below + # double-counts the orphan's hugepages. This matches on + # api_endpoint+socket — unlike the SSD-overlap cleanup above, which + # misses an attempt interrupted before SSD assignment (empty ssd_pcie). + for n in db_controller.get_storage_nodes_by_cluster_id(cluster_id): + if (n.api_endpoint == node_addr and n.socket == node_socket + and n.status == StorageNode.STATUS_IN_CREATION): + logger.warning( + f"Found stale IN_CREATION node {n.get_id()} for {node_addr} " + f"socket {node_socket} from an interrupted add; cleaning up before retry") + try: + n.client(timeout=20).spdk_process_kill(n.rpc_port, n.cluster_id) + except Exception: + logger.warning("Failed to kill SPDK for stale in_creation node", exc_info=True) + try: + storage_events.snode_delete(n) + except Exception: + logger.warning("snode_delete event failed for stale node", exc_info=True) + n.remove(db_controller.kv_store) + total_mem = minimum_hp_memory for n in db_controller.get_storage_nodes_by_cluster_id(cluster_id): if n.api_endpoint == node_addr and n.socket == node_socket: @@ -2563,12 +2591,50 @@ def restart_storage_node( failed — that's the case the cleanup is for.""" db_ctrl = DBController() pre_status = None + _snode_pre = None try: - pre_status = db_ctrl.get_storage_node_by_id(node_id).status + _snode_pre = db_ctrl.get_storage_node_by_id(node_id) + pre_status = _snode_pre.status except Exception: logger.warning(f"Could not read pre-call status for {node_id}; " f"skipping orphan-RESTARTING cleanup as a precaution") + # Transferable ownership: ensure a persistent NODE_RESTART task exists, + # claim its lease for this host, and heartbeat it while this process + # drives the restart. If this process dies mid-restart (pod evicted while + # its host drains, CLI killed), the lease goes stale within + # TASK_LEASE_TTL_SEC and a live tasks-runner claims the task and resumes + # the restart — instead of the node staying orphaned in RESTARTING and + # deadlocking node drains (2026-07-04 MCO rollout incident). On success + # the ONLINE transition auto-cancels the task. Pre-status RESTARTING / + # IN_SHUTDOWN means another caller owns the transition — don't touch + # its task, and don't add our own. + _hb_stop = threading.Event() + if pre_status not in (StorageNode.STATUS_RESTARTING, StorageNode.STATUS_IN_SHUTDOWN, None): + try: + from simplyblock_core.controllers import tasks_controller + # Reuse the node read for pre_status above — pre_status is only + # non-None (and thus inside this block) when that read succeeded, + # so _snode_pre is populated. Re-fetching here would be a wasted + # FDB round-trip and, more subtly, breaks callers/tests that count + # get_storage_node_by_id calls against the wrapper's contract. + _task_id = tasks_controller.ensure_node_restart_task(_snode_pre) + _hb_task = db_ctrl.get_task_by_id(_task_id) if _task_id else None + if _hb_task and tasks_controller.claim_task(_hb_task): + def _lease_heartbeat(): + while not _hb_stop.wait(constants.TASK_LEASE_HEARTBEAT_SEC): + try: + if not tasks_controller.refresh_task_lease(_hb_task): + # Lost the lease (another host took over) — + # stop heartbeating; the node-status lock + # still serializes the actual restart work. + return + except Exception as hb_e: + logger.debug(f"Restart lease heartbeat failed for {node_id}: {hb_e}") + threading.Thread(target=_lease_heartbeat, daemon=True).start() + except Exception as e: + logger.warning(f"Could not set up transferable restart ownership for {node_id}: {e}") + result = False try: result = _restart_storage_node_impl( @@ -2585,6 +2651,7 @@ def restart_storage_node( # is also down) undiagnosable from the logs. logger.error("restart_storage_node raised unexpectedly", exc_info=True) finally: + _hb_stop.set() # Trust the DB. If the impl raised after the ONLINE write was # already committed, the node IS factually online — peers see # ONLINE, IO is being served — and the only thing that "failed" @@ -3383,6 +3450,17 @@ def _abort_restart(reason): except Exception as ana_e: logger.error("ANA failback during restart of %s failed: %s", snode.get_id(), ana_e) + # Start data migration + online_devices_list = [] + for dev in snode.nvme_devices: + if dev.status in [NVMeDevice.STATUS_ONLINE, + NVMeDevice.STATUS_CANNOT_ALLOCATE, + NVMeDevice.STATUS_FAILED_AND_MIGRATED]: + online_devices_list.append(dev.get_id()) + if online_devices_list: + logger.info(f"Starting migration task for node {snode.get_id()}") + tasks_controller.add_device_mig_task_for_node(snode.get_id()) + logger.info("Setting node status to Online") if not set_node_status(snode.get_id(), StorageNode.STATUS_ONLINE, caused_by="restart"): # See twin call site above (single-leader restart path) for @@ -3405,16 +3483,6 @@ def _abort_restart(reason): lvol_list = db_controller.get_lvols_by_node_id(snode.get_id()) logger.info(f"Found {len(lvol_list)} lvols") - # Phase 10: start data migration, set node online - online_devices_list = [] - for dev in snode.nvme_devices: - if dev.status in [NVMeDevice.STATUS_ONLINE, - NVMeDevice.STATUS_CANNOT_ALLOCATE, - NVMeDevice.STATUS_FAILED_AND_MIGRATED]: - online_devices_list.append(dev.get_id()) - if online_devices_list: - logger.info(f"Starting migration task for node {snode.get_id()}") - tasks_controller.add_device_mig_task_for_node(snode.get_id()) return True @@ -3840,6 +3908,86 @@ def _detach_one_peer(peer): return detached[0] +def check_node_shutdown_preconditions(node_id, force=False): + """Read-only validation of everything that can refuse a graceful node + shutdown. Returns (allowed, reason). + + Exists so API endpoints can evaluate the guards SYNCHRONOUSLY and return + an actionable error (409) to the caller. Previously these checks only ran + inside shutdown_storage_node in the endpoint's fire-and-forget background + thread: the API had already answered 202, so a refusal (e.g. active + migration tasks) was invisible to the caller — the k8s operator polled + for the node to go offline forever and node drains stalled even after + the blocking condition cleared (2026-07-06 MCO rollout incident). + + With force=True the refusals downgrade to warnings and the shutdown is + allowed, mirroring shutdown_storage_node's historical behavior. + """ + db_controller = DBController() + try: + snode = db_controller.get_storage_node_by_id(node_id) + except KeyError: + return False, f"Storage node not found: {node_id}" + + # Guard: no concurrent shutdown + restart (design: mutual exclusion) + for peer in db_controller.get_storage_nodes_by_cluster_id(snode.cluster_id): + if peer.get_id() != node_id and peer.status == StorageNode.STATUS_RESTARTING: + reason = (f"Node {peer.get_id()} is restarting in this cluster, " + f"cannot shutdown {node_id} concurrently") + if force is False: + logger.error(reason) + return False, reason + logger.warning("%s — proceeding with force", reason) + if peer.get_id() != node_id and peer.status == StorageNode.STATUS_IN_SHUTDOWN: + reason = (f"Node {peer.get_id()} is already shutting down in this cluster, " + f"cannot shutdown {node_id} concurrently") + if force is False: + logger.error(reason) + return False, reason + logger.warning("%s — proceeding with force", reason) + + task_id = tasks_controller.get_active_node_restart_task(snode.cluster_id, snode.get_id()) + if task_id: + reason = f"Restart task found: {task_id}, can not shutdown storage node" + if force is False: + logger.error(reason) + return False, reason + logger.warning("%s — proceeding with force", reason) + + tasks = tasks_controller.get_active_node_tasks(snode.cluster_id, snode.get_id()) + if tasks: + if not force and _allow_shutdown_with_migration_tasks(snode, db_controller): + logger.warning( + "Migration task found: %s, proceeding with shutdown because FTT=2 allows node outage", + len(tasks), + ) + elif force: + logger.warning( + "Migration task found: %s, proceeding with forced shutdown", + len(tasks), + ) + else: + reason = f"Migration task found: {len(tasks)}, can not shutdown storage node or use --force" + logger.error(reason) + return False, reason + + if snode.status not in ( + StorageNode.STATUS_ONLINE, + StorageNode.STATUS_SUSPENDED, + StorageNode.STATUS_DOWN, + ): + if force: + logger.warning( + "Node status is %s, proceeding with force", snode.status) + else: + reason = (f"Node is in {snode.status} state; only online/suspended/down " + f"can be gracefully shut down. Use --force.") + logger.error(reason) + return False, reason + + return True, "" + + def shutdown_storage_node(node_id, force=False, keep_auto_restart=False): """Gracefully terminate a storage node. @@ -3902,56 +4050,9 @@ def shutdown_storage_node(node_id, force=False, keep_auto_restart=False): # this for its own non-force shutdown endpoint, where the policy # decision belongs. - # Guard: no concurrent shutdown + restart (design: mutual exclusion) - for peer in db_controller.get_storage_nodes_by_cluster_id(snode.cluster_id): - if peer.get_id() != node_id and peer.status == StorageNode.STATUS_RESTARTING: - logger.error( - f"Node {peer.get_id()} is restarting in this cluster, " - f"cannot shutdown {node_id} concurrently") - if force is False: - return False - if peer.get_id() != node_id and peer.status == StorageNode.STATUS_IN_SHUTDOWN: - logger.error( - f"Node {peer.get_id()} is already shutting down in this cluster, " - f"cannot shutdown {node_id} concurrently") - if force is False: - return False - - task_id = tasks_controller.get_active_node_restart_task(snode.cluster_id, snode.get_id()) - if task_id: - logger.error(f"Restart task found: {task_id}, can not shutdown storage node") - if force is False: - return False - - tasks = tasks_controller.get_active_node_tasks(snode.cluster_id, snode.get_id()) - if tasks: - if not force and _allow_shutdown_with_migration_tasks(snode, db_controller): - logger.warning( - "Migration task found: %s, proceeding with shutdown because FTT=2 allows node outage", - len(tasks), - ) - elif force: - logger.warning( - "Migration task found: %s, proceeding with forced shutdown", - len(tasks), - ) - else: - logger.error(f"Migration task found: {len(tasks)}, can not shutdown storage node or use --force") - return False - - if snode.status not in ( - StorageNode.STATUS_ONLINE, - StorageNode.STATUS_SUSPENDED, - StorageNode.STATUS_DOWN, - ): - if force: - logger.warning( - "Node status is %s, proceeding with force", snode.status) - else: - logger.error( - "Node is in %s state; only online/suspended/down can be " - "gracefully shut down. Use --force.", snode.status) - return False + allowed, _reason = check_node_shutdown_preconditions(node_id, force=force) + if not allowed: + return False # Step 1: mark the node in_shutdown. set_node_status fans out a # node_status event to peers so their cluster maps see "this node diff --git a/simplyblock_core/utils/__init__.py b/simplyblock_core/utils/__init__.py index a5eee2872..96a3e5783 100644 --- a/simplyblock_core/utils/__init__.py +++ b/simplyblock_core/utils/__init__.py @@ -2441,10 +2441,21 @@ def render_and_deploy_alerting_configs(contact_point, grafana_endpoint, cluster_ print(f"File moved to {prometheus_file_path} successfully.") +# hostPath mount of the host's /etc/modules-load.d inside the k8s storage-node +# agent (see storage-node.yaml in the CSI chart). Writing to the container's +# own /etc/modules-load.d is a silent no-op there: the file lands in the +# ephemeral container layer, the host's systemd-modules-load never sees it, +# and modules are gone after the next node reboot — which wedged node-adds +# resuming after the CPU-topology reboot (2026-07-06, bind_device_to_spdk 500). +HOST_MODULES_LOAD_DIR = "/host/etc/modules-load.d" + + def load_kernel_module(module): """ - Loads a kernel module using modprobe and ensures it is persistent across reboots - by creating a module file in /etc/modules-load.d/.conf. + Loads a kernel module using modprobe and ensures it is persistent across + reboots by creating a module file in modules-load.d/.conf — on the + HOST when running inside the k8s agent (via the HOST_MODULES_LOAD_DIR + hostPath mount), otherwise locally. """ try: # Attempt to load the module immediately @@ -2456,8 +2467,12 @@ def load_kernel_module(module): # Ensure persistence across reboots try: - path = f"/etc/modules-load.d/{module}.conf" - os.makedirs("/etc/modules-load.d", exist_ok=True) + if os.path.isdir(HOST_MODULES_LOAD_DIR): + target_dir = HOST_MODULES_LOAD_DIR + else: + target_dir = "/etc/modules-load.d" + os.makedirs(target_dir, exist_ok=True) + path = f"{target_dir}/{module}.conf" with open(path, "w") as f: f.write(f"{module}\n") @@ -2477,6 +2492,58 @@ def load_kube_config_with_fallback(): config.load_kube_config() +def set_storage_mcp_max_unavailable(cluster_id: str, max_unavailable: int) -> bool: + """Set spec.maxUnavailable on the cluster's storage MachineConfigPool. + + This controls how many storage nodes MCO reboots at once for a + MachineConfig / KubeletConfig rollout (the CPU-topology apply). It is + flipped between two phases: + + * During initial node-add (cluster not yet active, nodes carry no data): + a WIDE value (= the parallel-add count) so the first-time topology + reboots happen in one wave instead of a serialized, one-at-a-time + queue — a node added early is otherwise stuck behind every other + node's reboot before its own. + * After cluster activation (nodes now serve IO): NARROWED to the + cluster's fault tolerance, so any later rollout never reboots more + storage nodes at once than the data plane can absorb. + + OpenShift-only: MachineConfigPool is an OCP CRD, so this is a no-op on k3s + or when CPU topology was never applied (pool absent). Never raises — this + is rollout pacing, not business logic, and must not crash activation or + node-add. Returns True on success, False otherwise. + """ + # maxUnavailable=0 wedges the pool (MCO can never take a node), so floor + # at 1. The MCP name mirrors the CPU-topology job template + # (storage-). + value = max(int(max_unavailable), 1) + mcp_name = f"storage-{first_six_chars(cluster_id)}" + try: + load_kube_config_with_fallback() + api = client.CustomObjectsApi() + api.patch_cluster_custom_object( + group="machineconfiguration.openshift.io", + version="v1", + plural="machineconfigpools", + name=mcp_name, + body={"spec": {"maxUnavailable": value}}, + ) + logger.info(f"Set MachineConfigPool {mcp_name} maxUnavailable={value}") + return True + except ApiException as e: + if e.status == 404: + logger.info(f"MachineConfigPool {mcp_name} not found " + f"(non-OpenShift or CPU topology not applied); " + f"skipping maxUnavailable update") + else: + logger.warning(f"Failed to set maxUnavailable on MCP {mcp_name}: " + f"{e.reason} {e.body}") + return False + except Exception as e: + logger.warning(f"Failed to set maxUnavailable on MCP {mcp_name}: {e}") + return False + + def patch_cr_status( *, group: str, @@ -2527,7 +2594,7 @@ def patch_cr_node_status( node_mgmt_ip: str, updates: Optional[Dict[str, Any]] = None, remove: bool = False, -): +) -> bool: """ Patch status.nodes[*] fields for a specific node identified by UUID. @@ -2539,18 +2606,38 @@ def patch_cr_node_status( {"health": "true"} {"status": "offline"} {"capacity": {"sizeUsed": 1234}} + + Returns True on success, False on failure. Never raises: this mirrors + state into the CR for observability and must not crash business logic + (an unhandled raise here killed cluster_activate mid-flight, leaving the + cluster wedged in in_activation). """ load_kube_config_with_fallback() api = client.CustomObjectsApi() - try: - cr = api.get_namespaced_custom_object( - group=group, - version=version, - namespace=namespace, - plural=plural, - name=name, - ) + # status.nodes is replaced wholesale by several concurrent writers (node + # registration, monitor health patches, port events), so a read can + # transiently miss a node that a concurrent writer is about to (re)add, + # and a blind write can silently drop a concurrent update. Retry with + # optimistic locking instead of failing hard. + attempts = 5 + for attempt in range(attempts): + if attempt: + time.sleep(2) + + try: + cr = api.get_namespaced_custom_object( + group=group, + version=version, + namespace=namespace, + plural=plural, + name=name, + ) + except ApiException as e: + logger.error( + f"Failed to read CR {name}: {e.reason} {e.body}" + ) + return False status_nodes = cr.get("status", {}).get("nodes", []) found = False @@ -2576,28 +2663,45 @@ def patch_cr_node_status( if not found: if remove: # Node already absent from status — nothing to do. - return - raise RuntimeError( - f"Node not found (uuid={node_uuid}, mgmtIp={node_mgmt_ip})" - ) + return True + # Node not (yet) in status — likely racing a concurrent + # whole-list rewrite; retry on fresh data. + continue - api.patch_namespaced_custom_object_status( - group=group, - version=version, - namespace=namespace, - plural=plural, - name=name, - body={ - "status": { - "nodes": new_status_nodes - } - }, - ) + try: + api.patch_namespaced_custom_object_status( + group=group, + version=version, + namespace=namespace, + plural=plural, + name=name, + body={ + # resourceVersion makes this a compare-and-swap: if + # another writer replaced status.nodes since our read, + # the API returns 409 and we retry on fresh data instead + # of clobbering their update. + "metadata": {"resourceVersion": cr["metadata"]["resourceVersion"]}, + "status": { + "nodes": new_status_nodes + } + }, + ) + return True + except ApiException as e: + if e.status == 409: + # Lost-update conflict — retry on fresh data. + continue + logger.error( + f"Failed to patch node for {name}: {e.reason} {e.body}" + ) + return False - except ApiException as e: - logger.error( - f"Failed to patch node for {name}: {e.reason} {e.body}" - ) + logger.error( + f"Failed to patch CR {name} after {attempts} attempts: node not " + f"found or persistent write conflicts (uuid={node_uuid}, " + f"mgmtIp={node_mgmt_ip})" + ) + return False def patch_cr_lvol_status( diff --git a/simplyblock_web/api/internal/storage_node/kubernetes.py b/simplyblock_web/api/internal/storage_node/kubernetes.py index db86a6fe1..645c8a1b7 100644 --- a/simplyblock_web/api/internal/storage_node/kubernetes.py +++ b/simplyblock_web/api/internal/storage_node/kubernetes.py @@ -288,6 +288,13 @@ def spdk_process_start(body: SPDKParams): if isinstance(skip_kubelet_configuration, str): skip_kubelet_configuration = skip_kubelet_configuration.strip().lower() in ("true") reserved_system_cpus = os.environ.get("RESERVED_SYSTEM_CPUS", "0,1") + # How many storage nodes MCO may reboot at once when the MCP is first + # created for CPU-topology apply. Defaults to 1 (conservative, one-at-a-time + # — current behavior). Set to the parallel-add count so the initial + # (pre-activation, data-less) first-time reboots roll in one wave instead of + # a one-at-a-time queue; cluster_activate later narrows the pool to the + # cluster's fault tolerance for safe steady-state rollouts. + mcp_max_unavailable = os.environ.get("PARALLEL_ADD_NUMBER", "1") openshift_mcp = os.environ.get("OPENSHIFT_MCP") node_prepration_core_name = "snode-spdk-core-isolate-" @@ -330,6 +337,7 @@ def spdk_process_start(body: SPDKParams): 'FW_PORT': body.firewall_port, 'CPU_TOPOLOGY_ENABLED': cpu_topology_enabled, 'RESERVED_SYSTEM_CPUS': reserved_system_cpus, + 'MCP_MAX_UNAVAILABLE': mcp_max_unavailable, 'OPENSHIFT_MCP': openshift_mcp, 'TLS_SERVE': settings.tls_serve, 'TLS_CONNECT': settings.tls_connect, diff --git a/simplyblock_web/api/v2/cluster/storage_node/__init__.py b/simplyblock_web/api/v2/cluster/storage_node/__init__.py index 942242c99..a479b512d 100644 --- a/simplyblock_web/api/v2/cluster/storage_node/__init__.py +++ b/simplyblock_web/api/v2/cluster/storage_node/__init__.py @@ -205,7 +205,7 @@ def resume(cluster: Cluster, storage_node: StorageNode) -> Response: return Response(status_code=204) -@instance_api.post('/shutdown', name='clusters:storage-nodes:shutdown', status_code=202, responses={202: {"content": None}}) +@instance_api.post('/shutdown', name='clusters:storage-nodes:shutdown', status_code=202, responses={202: {"content": None}, 409: {"description": "Shutdown preconditions not met; retry later or use force"}}) def shutdown(cluster: Cluster, storage_node: StorageNode, force: bool = False) -> Response: if not force: from simplyblock_core.storage_node_ops import _check_ftt_allows_node_removal @@ -214,6 +214,19 @@ def shutdown(cluster: Cluster, storage_node: StorageNode, force: bool = False) - if not allowed: raise ValueError(reason) + # Evaluate every condition that would make the background shutdown bail + # BEFORE answering: a refusal after 202 is invisible to the caller (the + # k8s operator polled forever for a shutdown that had already been + # rejected because migration tasks were running — 2026-07-06 node-drain + # stall). 409 tells the caller "not now, retry later or use force". + # Only meaningful on the graceful path: with force=True every guard in + # check_node_shutdown_preconditions downgrades to a warning and returns + # allowed, so there is nothing to synchronously refuse. + allowed, reason = storage_node_ops.check_node_shutdown_preconditions( + storage_node.get_id()) + if not allowed: + raise HTTPException(409, reason) + Thread( target=storage_node_ops.shutdown_storage_node, args=(storage_node.get_id(), force) diff --git a/simplyblock_web/templates/oc_storage_cpu_topology.yaml.j2 b/simplyblock_web/templates/oc_storage_cpu_topology.yaml.j2 index ef051f85c..7e149a323 100755 --- a/simplyblock_web/templates/oc_storage_cpu_topology.yaml.j2 +++ b/simplyblock_web/templates/oc_storage_cpu_topology.yaml.j2 @@ -102,7 +102,11 @@ spec: operator: In values: - $HOSTNAME - maxUnavailable: 1 + # Wide during initial node-add so first-time CPU-topology + # reboots roll in parallel (nodes are empty pre-activation); + # cluster_activate narrows this to the fault tolerance once the + # cluster serves IO. + maxUnavailable: {{ MCP_MAX_UNAVAILABLE }} MCPEOF fi