From eeaf769030da2d04f6b57f5d11de30536f5a3877 Mon Sep 17 00:00:00 2001 From: wmousa Date: Sat, 4 Jul 2026 02:06:58 +0200 Subject: [PATCH 01/10] fix: prevent cluster activation from wedging in in_activation Activation of a fresh cluster died mid-flight with "RuntimeError: Node not found" from patch_cr_node_status and left the cluster permanently in_activation, since the activation thread had no failure handling. Root cause: patch_cr_node_status replaces the CR's whole status.nodes list via read-modify-write with no concurrency control. During activation, many writers (node registration, monitor health patches, port events) patch the same list concurrently, so a reader can observe a list momentarily missing a node and a writer can silently drop a concurrent update. - patch_cr_node_status: retry when the node is not (yet) present, send resourceVersion with the status patch so conflicting writes 409 and retry on fresh data instead of clobbering each other, and log+return False instead of raising - CR status mirroring must never crash the caller - cluster_activate: wrap the implementation so any unhandled failure reverts in_activation to the prior status before re-raising, keeping the cluster retryable - create_lvstore call in activation: add the same error handling as the sibling recreate_lvstore path --- simplyblock_core/cluster_ops.py | 30 +++++++++- simplyblock_core/utils/__init__.py | 95 +++++++++++++++++++++--------- 2 files changed, 94 insertions(+), 31 deletions(-) diff --git a/simplyblock_core/cluster_ops.py b/simplyblock_core/cluster_ops.py index 6c86cbf33..137031107 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" diff --git a/simplyblock_core/utils/__init__.py b/simplyblock_core/utils/__init__.py index a5eee2872..d1858c4dc 100644 --- a/simplyblock_core/utils/__init__.py +++ b/simplyblock_core/utils/__init__.py @@ -2527,7 +2527,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 +2539,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 +2596,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( From fc96909cc033823abe07a048391f298ba2d71680 Mon Sep 17 00:00:00 2001 From: wmousa Date: Mon, 6 Jul 2026 12:57:10 +0200 Subject: [PATCH 02/10] fix: make storage node restarts and cluster activation crash-safe Cluster activation and node restarts could both wedge permanently when the driving process died or raced concurrent writers: - cluster_activate died on a transient "Node not found" from patch_cr_node_status and left the cluster stuck in in_activation. The CR status.nodes list is rewritten wholesale by concurrent writers, so reads can transiently miss a node and writes can drop concurrent updates. - A node whose restart was interrupted (tasks-runner pod evicted while its host drains, node crash) stayed orphaned in RESTARTING with no task and no owner. The k8s operator holds its drain slot until the node is online, deadlocking MachineConfig rollouts cluster-wide (observed 2026-07-04: every MCO reboot froze the rollout until manual intervention). Changes: - patch_cr_node_status: retry on transient not-found, compare-and-swap via resourceVersion (409 -> reread and retry), log-and-return instead of raising - CR status mirroring must never crash business logic - cluster_activate: wrap the implementation so any unhandled failure reverts in_activation to the prior status, keeping retry possible - create_lvstore call in activation: same error handling as the sibling recreate_lvstore path - restart_storage_node: every restart now ensures a persistent NODE_RESTART task, claims its lease, and heartbeats it (30s) while driving the restart; TASK_LEASE_TTL_SEC drops 1200->180 so a live tasks-runner takes over a dead driver's restart within ~3 minutes - tasks_runner_restart: watchdog that detects nodes stuck in RESTARTING/IN_SHUTDOWN with no owning task, verifies SPDK is dead (_reset_if_transient), resets to OFFLINE and queues an auto-restart --- simplyblock_core/constants.py | 19 +++-- .../controllers/tasks_controller.py | 42 ++++++++++ .../services/tasks_runner_restart.py | 81 +++++++++++++++++++ simplyblock_core/storage_node_ops.py | 33 ++++++++ 4 files changed, 169 insertions(+), 6 deletions(-) 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..ab9c5c580 100644 --- a/simplyblock_core/controllers/tasks_controller.py +++ b/simplyblock_core/controllers/tasks_controller.py @@ -69,6 +69,48 @@ 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"] + + +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_restart.py b/simplyblock_core/services/tasks_runner_restart.py index 0cb4806ad..884b8ede2 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: @@ -566,4 +642,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..4c395dda7 100755 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -2569,6 +2569,38 @@ def restart_storage_node( 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 + _snode_pre = db_ctrl.get_storage_node_by_id(node_id) + _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 +2617,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" From 3ccc65b78549830c9e741c6673c0fc1584c2a3f7 Mon Sep 17 00:00:00 2001 From: hamdykhader Date: Mon, 6 Jul 2026 17:15:08 +0300 Subject: [PATCH 03/10] Add migration tasks before node status is set to Online --- simplyblock_core/storage_node_ops.py | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 4c395dda7..4fabf9f2c 100755 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -3416,6 +3416,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 @@ -3438,16 +3449,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 From 2bfd9cc7d81d99b26bd41697b4cb752d979f478f Mon Sep 17 00:00:00 2001 From: wmousa Date: Mon, 6 Jul 2026 16:19:11 +0200 Subject: [PATCH 04/10] fix: return 409 from shutdown API when preconditions are not met The /shutdown endpoint answered 202 and evaluated its safety guards (active migration tasks, active restart task, concurrent peer restart/shutdown, node state) only in the fire-and-forget background thread. A refusal was invisible to the caller: during a MachineConfig rollout the k8s operator took the 202 as "shutdown in progress" and switched to polling for the node to go offline - forever. The blocking migrations finished minutes later, but nothing re-issued the shutdown, stalling the node drain until manual intervention (2026-07-06). Extract the guards into check_node_shutdown_preconditions(), a read-only validator (force downgrades refusals to warnings, matching the previous behavior), call it synchronously in the endpoint and return 409 with the refusal reason. shutdown_storage_node performs the same validation through the shared helper, so CLI and other callers keep the same protection. A 409 also makes the operator re-issue the shutdown on each reconcile instead of entering its poll-forever state, so drains now proceed automatically once the blocking condition clears. --- simplyblock_core/storage_node_ops.py | 132 +++++++++++------- .../api/v2/cluster/storage_node/__init__.py | 12 +- 2 files changed, 93 insertions(+), 51 deletions(-) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 4fabf9f2c..78f978365 100755 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -3874,6 +3874,85 @@ 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. + """ + 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. @@ -3936,56 +4015,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_web/api/v2/cluster/storage_node/__init__.py b/simplyblock_web/api/v2/cluster/storage_node/__init__.py index 942242c99..4d2339b70 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,16 @@ 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". + allowed, reason = storage_node_ops.check_node_shutdown_preconditions( + storage_node.get_id(), force=force) + if not allowed: + raise HTTPException(409, reason) + Thread( target=storage_node_ops.shutdown_storage_node, args=(storage_node.get_id(), force) From b3892a3056282b5e9b4649047651e4cc38a79ff7 Mon Sep 17 00:00:00 2001 From: wmousa Date: Mon, 6 Jul 2026 16:38:28 +0200 Subject: [PATCH 05/10] Fix: Instantiate db_controller = DBController() at the top of the validator --- simplyblock_core/storage_node_ops.py | 1 + 1 file changed, 1 insertion(+) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 78f978365..d14c74ea9 100755 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -3889,6 +3889,7 @@ def check_node_shutdown_preconditions(node_id, force=False): 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: From 711673fe2ce771101d66b8a2f9aad2d9f5eb065b Mon Sep 17 00:00:00 2001 From: wmousa Date: Mon, 6 Jul 2026 18:06:19 +0200 Subject: [PATCH 06/10] fix: heartbeat task leases during long-running task execution TASK_LEASE_TTL_SEC was reduced to 180s for fast ownership transfer, but runners only refreshed a lease on task writes. Long-blocking work (add_node can run for many minutes without touching its task) let a live owner's lease go stale mid-execution: 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 is destructive: the re-entrant cleanup kills the in-flight add's SPDK and deletes its half-created node record. Add tasks_controller.task_lease_heartbeat(), a context manager that refreshes this host's lease every TASK_LEASE_HEARTBEAT_SEC and stops itself if the lease is lost, and wrap task execution at every claim site (node_add, restart, migration, lvol_migration, port_allow, cluster_expand). Leases are per task, so concurrent adds of different nodes remain fully parallel; takeover now occurs only on actual owner death. --- .../controllers/tasks_controller.py | 36 +++++++++++++++++++ .../services/tasks_runner_cluster_expand.py | 3 +- .../services/tasks_runner_lvol_migration.py | 3 +- .../services/tasks_runner_migration.py | 3 +- .../services/tasks_runner_node_add.py | 5 ++- .../services/tasks_runner_port_allow.py | 3 +- .../services/tasks_runner_restart.py | 6 +++- 7 files changed, 53 insertions(+), 6 deletions(-) diff --git a/simplyblock_core/controllers/tasks_controller.py b/simplyblock_core/controllers/tasks_controller.py index ab9c5c580..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 @@ -92,6 +94,40 @@ def _mutate(t): 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 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..b7394e2fd 100644 --- a/simplyblock_core/services/tasks_runner_node_add.py +++ b/simplyblock_core/services/tasks_runner_node_add.py @@ -92,7 +92,10 @@ 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. + with tasks_controller.task_lease_heartbeat(task): + res = process_task(task, cl) if res: if task.status == JobSchedule.STATUS_DONE: break 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 884b8ede2..a80f9285d 100644 --- a/simplyblock_core/services/tasks_runner_restart.py +++ b/simplyblock_core/services/tasks_runner_restart.py @@ -620,7 +620,11 @@ def _watchdog_orphaned_transitional_nodes(cluster_id): 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) From 416c78bdabaa93cd44b20b34c745af107960f944 Mon Sep 17 00:00:00 2001 From: wmousa Date: Tue, 7 Jul 2026 00:18:43 +0200 Subject: [PATCH 07/10] fix: survive the CPU-topology reboot during node-add Applying the CPU topology during add_node reboots the node, and two independent defects turned that expected reboot into failed or stalled adds (2026-07-06, workers 0/4): - load_kernel_module wrote its modules-load.d persistence into the storage-node agent container's ephemeral filesystem, where the host's systemd-modules-load never sees it. vfio-pci/uio_pci_generic were therefore gone after the reboot and bind_device_to_spdk answered 500 on every retry. Write through the /host/etc/modules-load.d hostPath mount when present (added to the CSI chart's storage-node daemonset), falling back to the local path for docker/bare-metal deployments. - The node-add runner was blind to the reboot it triggered: clean failed attempts retried after a flat 10s, burning max_retry against a node that is down for 5-8 minutes, while excepting attempts grew the exponential backoff so the runner kept sleeping long after the node was back. After a failed attempt whose node agent address is unreachable, wait for it to answer again (poll 15s, bounded by NODE_REBOOT_WAIT_MAX_SEC=900, matching the topology job's reboot budget) without consuming retries, then retry promptly with the backoff reset. An existing add task declares the node is meant to join, so waiting out unreachability is the correct convergence behavior. Requires the matching CSI chart change mounting the host's /etc/modules-load.d at /host/etc/modules-load.d in the agent. --- .../services/tasks_runner_node_add.py | 58 ++++++++++++++++++- simplyblock_core/utils/__init__.py | 23 ++++++-- 2 files changed, 76 insertions(+), 5 deletions(-) diff --git a/simplyblock_core/services/tasks_runner_node_add.py b/simplyblock_core/services/tasks_runner_node_add.py index b7394e2fd..b0c02a6b9 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. @@ -99,7 +148,14 @@ def _run_task(task_uuid, cluster_id): if res: if task.status == JobSchedule.STATUS_DONE: break - else: + # Reboot-aware wait: an attempt that failed because the node went + # down (topology reboot) neither burns the backoff nor retries + # blindly — wait for the agent to answer, then retry promptly on + # a fresh schedule. + if _wait_node_reachable(task, task_uuid): + 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/utils/__init__.py b/simplyblock_core/utils/__init__.py index d1858c4dc..1f4ec1654 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") From 79d6da88f781b9ea9290b917127df45d2a2ece73 Mon Sep 17 00:00:00 2001 From: wmousa Date: Tue, 7 Jul 2026 23:38:39 +0200 Subject: [PATCH 08/10] fix: make node-add interruption-harmless via idempotent re-entry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When CPU topology is enabled, adding a node reboots it — killing the agent that serves spdk_process_start and, if the tasks-runner pod is co-located on that storage node, the pod driving the add too (taking every parallel add with it). The reboot is unavoidable, so recovery must be clean. Two gaps prevented that: - add_node's stale-record cleanup only matched on SSD-PCIe overlap, so an attempt interrupted before SSD assignment left an orphaned IN_CREATION StorageNode. The lease-based retry then built a DUPLICATE node and the total_mem loop double-counted the orphan's hugepages. Add an idempotent re-entry cleanup keyed on api_endpoint+socket that kills any half-started SPDK and drops the stale record before the hugepage/mesh work runs. - The guaranteed topology reboot makes spdk_process_start fail and add_node return False, which process_task counted as a consumed retry. The runner now detects (via agent reachability) that the failure was a reboot and rolls the retry back, so the expected reboot never erodes the max_retry budget; it then waits for the agent to return and retries promptly. Combined with the existing task-lease takeover, an add interrupted by any reboot — including the co-located-pod case — resumes and converges automatically with no duplicate nodes and no wasted retries. --- .../services/tasks_runner_node_add.py | 19 ++++++++++--- simplyblock_core/storage_node_ops.py | 28 +++++++++++++++++++ 2 files changed, 43 insertions(+), 4 deletions(-) diff --git a/simplyblock_core/services/tasks_runner_node_add.py b/simplyblock_core/services/tasks_runner_node_add.py index b0c02a6b9..52a2db0f5 100644 --- a/simplyblock_core/services/tasks_runner_node_add.py +++ b/simplyblock_core/services/tasks_runner_node_add.py @@ -143,16 +143,27 @@ def _run_task(task_uuid, cluster_id): break # 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 - # Reboot-aware wait: an attempt that failed because the node went - # down (topology reboot) neither burns the backoff nor retries - # blindly — wait for the agent to answer, then retry promptly on - # a fresh schedule. + # 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: diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index d14c74ea9..085a2b413 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: From fd1805e0fb2dd2d91eeb465106d8aff04b758f38 Mon Sep 17 00:00:00 2001 From: wmousa Date: Wed, 8 Jul 2026 01:49:52 +0200 Subject: [PATCH 09/10] fix(shutdown API): only gate forced shutdown-preconditions on the graceful path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit check_node_shutdown_preconditions can only refuse a graceful shutdown; with force=True every guard downgrades to a warning and returns allowed, so calling it unconditionally was dead work that also broke the mocked endpoint test (MagicMock unpacks to 0 values). Move it inside the existing `if not force` block. fix(restart): reuse pre-status node read instead of re-fetching The transferable-ownership block re-read the node via get_storage_node_by_id immediately after the pre_status read — a wasted FDB round-trip that also consumed an extra read the restart-wrapper's tests count against its contract, breaking the orphan-RESTARTING cleanup path. Reuse the already-read node. --- simplyblock_core/storage_node_ops.py | 10 +++++++-- .../api/v2/cluster/storage_node/__init__.py | 21 +++++++++++-------- 2 files changed, 20 insertions(+), 11 deletions(-) diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 085a2b413..5f85440d7 100755 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -2591,8 +2591,10 @@ 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") @@ -2611,7 +2613,11 @@ def restart_storage_node( if pre_status not in (StorageNode.STATUS_RESTARTING, StorageNode.STATUS_IN_SHUTDOWN, None): try: from simplyblock_core.controllers import tasks_controller - _snode_pre = db_ctrl.get_storage_node_by_id(node_id) + # 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): diff --git a/simplyblock_web/api/v2/cluster/storage_node/__init__.py b/simplyblock_web/api/v2/cluster/storage_node/__init__.py index 4d2339b70..a479b512d 100644 --- a/simplyblock_web/api/v2/cluster/storage_node/__init__.py +++ b/simplyblock_web/api/v2/cluster/storage_node/__init__.py @@ -214,15 +214,18 @@ 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". - allowed, reason = storage_node_ops.check_node_shutdown_preconditions( - storage_node.get_id(), force=force) - if not allowed: - raise HTTPException(409, 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, From 0da861657c58148847d8e4d0f6a07cdffbbfa45f Mon Sep 17 00:00:00 2001 From: wmousa Date: Fri, 10 Jul 2026 14:43:34 +0200 Subject: [PATCH 10/10] feat(cpu-topology): scale MCP maxUnavailable to parallel-add, narrow to FTT on activate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deleting/creating the storage MachineConfigPool with a hardcoded maxUnavailable=1 serialized every first-time CPU-topology reboot pool-wide: with N nodes, a node added early could sit behind all the others' reboots before its own, so node-add's wait-for-reboot was unbounded and unpredictable. No tuning of the job's wait fixes this — the delay lives in MCO's serialized rollout, not the job. Make the reboot concurrency phase-aware instead: - During node-add (pre-activation, nodes carry no data) the MCP is created with maxUnavailable = PARALLEL_ADD_NUMBER, so the first-time topology reboots roll in parallel waves instead of a one-at-a-time queue. - At the end of cluster_activate (success path, before flipping to ACTIVE) the pool is narrowed to cluster.max_fault_tolerance, so any later MachineConfig/KubeletConfig rollout never reboots more storage nodes at once than the data plane can absorb. Post-activation expansion keeps this value (later adds only append to the nodeSelector), so it stays safe on a live cluster. Changes: - utils.set_storage_mcp_max_unavailable(): patch spec.maxUnavailable on the storage- MCP; floors at 1 (0 wedges the pool), OpenShift-only (no-ops on k3s / when the pool is absent), never raises. - cluster_ops._cluster_activate(): narrow to max_fault_tolerance before ACTIVE. - oc_storage_cpu_topology.yaml.j2: create MCP with maxUnavailable={{ MCP_MAX_UNAVAILABLE }}. - kubernetes.py: thread MCP_MAX_UNAVAILABLE from env PARALLEL_ADD_NUMBER (default 1). Default PARALLEL_ADD_NUMBER=1 preserves current one-at-a-time behavior; set it to the parallel-add count to enable parallel bring-up. Use max_fault_tolerance-1 for the steady-state value if you want headroom for an unplanned failure concurrent with a rollout. --- simplyblock_core/cluster_ops.py | 12 +++++ simplyblock_core/utils/__init__.py | 52 +++++++++++++++++++ .../api/internal/storage_node/kubernetes.py | 8 +++ .../templates/oc_storage_cpu_topology.yaml.j2 | 6 ++- 4 files changed, 77 insertions(+), 1 deletion(-) diff --git a/simplyblock_core/cluster_ops.py b/simplyblock_core/cluster_ops.py index 137031107..88983a2bb 100755 --- a/simplyblock_core/cluster_ops.py +++ b/simplyblock_core/cluster_ops.py @@ -1011,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/utils/__init__.py b/simplyblock_core/utils/__init__.py index 1f4ec1654..96a3e5783 100644 --- a/simplyblock_core/utils/__init__.py +++ b/simplyblock_core/utils/__init__.py @@ -2492,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, 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/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