diff --git a/simplyblock_cli/cli-reference.yaml b/simplyblock_cli/cli-reference.yaml index 7d227bef1..47bee110e 100644 --- a/simplyblock_cli/cli-reference.yaml +++ b/simplyblock_cli/cli-reference.yaml @@ -297,14 +297,11 @@ commands: type: bool action: store_true - name: remove - help: "Removes a storage node from the cluster." - description: > - The storage node cannot be used or added any more. Any data residing on this storage node will be migrated to - the remaining storage nodes. The user must ensure that there is sufficient free space in remaining cluster to - allow for successful node removal. - - !!! danger - If there isn't enough storage available, the cluster may run full and switch to read-only mode. + help: 'Removes a storage node from the cluster (inverse of cluster expansion). + The node must be ONLINE and all other nodes ONLINE; it must host no logical + volumes or snapshots (migrate those first). Runs as a background task: shuts + the node down, rewires LVS replicas onto other nodes, then removes, fails and + migrates its devices before marking the node removed.' arguments: - name: node_id help: Storage node id @@ -312,7 +309,7 @@ commands: completer: _completer_get_sn_list type: str - name: "--force-remove" - help: "Force remove all logical volumes and snapshots." + help: "Cancel any active tasks on the node before starting removal." dest: force_remove type: bool action: store_true @@ -449,21 +446,16 @@ commands: type: bool action: store_true - name: suspend - help: "DEPRECATED: the suspension phase was removed from graceful shutdown (it caused writer conflicts on sec/tert lvstores). This command is now a no-op returning success. Use `sn shutdown`." - usage: The node will stop accepting new IO, but will finish processing any IO, which has been received already. + help: "Exclude node from lvol allocation." + usage: The node will be excluded from lvol allocation. It will not be able to create new logical volumes on this node. arguments: - name: node_id help: Storage node id dest: node_id completer: _completer_get_sn_list type: str - - name: "--force" - help: "Ignored (kept for backwards compatibility)." - dest: force - type: bool - action: store_true - name: resume - help: "DEPRECATED: counterpart to `sn suspend`, also a no-op." + help: "Include node in lvol allocation." arguments: - name: node_id help: Storage node id diff --git a/simplyblock_cli/cli.py b/simplyblock_cli/cli.py index 2a0ebbca9..a1dc05480 100755 --- a/simplyblock_cli/cli.py +++ b/simplyblock_cli/cli.py @@ -165,9 +165,9 @@ def init_storage_node__delete(self, subparser): subcommand.add_argument('--force', help='Force delete storage node from DB. Ensure you know what you\'re doing.', dest='force_remove', action='store_true') def init_storage_node__remove(self, subparser): - subcommand = self.add_sub_command(subparser, 'remove', 'Removes a storage node from the cluster.') + subcommand = self.add_sub_command(subparser, 'remove', 'Removes a storage node from the cluster (inverse of cluster expansion). The node must be ONLINE and all other nodes ONLINE; it must host no logical volumes or snapshots (migrate those first). Runs as a background task: shuts the node down, rewires LVS replicas onto other nodes, then removes, fails and migrates its devices before marking the node removed.') subcommand.add_argument('node_id', help='Storage node id', type=str).completer = self._completer_get_sn_list - subcommand.add_argument('--force-remove', help='Force remove all logical volumes and snapshots.', dest='force_remove', action='store_true') + subcommand.add_argument('--force-remove', help='Cancel any active tasks on the node before starting removal.', dest='force_remove', action='store_true') def init_storage_node__list(self, subparser): subcommand = self.add_sub_command(subparser, 'list', 'Lists all storage nodes.') @@ -209,12 +209,11 @@ def init_storage_node__shutdown(self, subparser): subcommand.add_argument('--force', help='Force node shutdown.', dest='force', action='store_true') def init_storage_node__suspend(self, subparser): - subcommand = self.add_sub_command(subparser, 'suspend', 'DEPRECATED: the suspension phase was removed from graceful shutdown (it caused writer conflicts on sec/tert lvstores). This command is now a no-op returning success. Use `sn shutdown`.') + subcommand = self.add_sub_command(subparser, 'suspend', 'Exclude node from lvol allocation.') subcommand.add_argument('node_id', help='Storage node id', type=str).completer = self._completer_get_sn_list - subcommand.add_argument('--force', help='Ignored (kept for backwards compatibility).', dest='force', action='store_true') def init_storage_node__resume(self, subparser): - subcommand = self.add_sub_command(subparser, 'resume', 'DEPRECATED: counterpart to `sn suspend`, also a no-op.') + subcommand = self.add_sub_command(subparser, 'resume', 'Include node in lvol allocation.') subcommand.add_argument('node_id', help='Storage node id', type=str).completer = self._completer_get_sn_list def init_storage_node__get_io_stats(self, subparser): diff --git a/simplyblock_cli/clibase.py b/simplyblock_cli/clibase.py index 67ca984eb..f6fabd669 100755 --- a/simplyblock_cli/clibase.py +++ b/simplyblock_cli/clibase.py @@ -301,7 +301,7 @@ def storage_node__shutdown(self, sub_command, args): return ret def storage_node__suspend(self, sub_command, args): - ret = storage_ops.suspend_storage_node(args.node_id, args.force) + ret = storage_ops.suspend_storage_node(args.node_id) if isinstance(ret, tuple): ok, reason = ret if not ok: diff --git a/simplyblock_core/cluster_ops.py b/simplyblock_core/cluster_ops.py index 895acdb22..cf19d0a6a 100755 --- a/simplyblock_core/cluster_ops.py +++ b/simplyblock_core/cluster_ops.py @@ -667,6 +667,27 @@ def _wait_for_full_device_connectivity(cl_id, timeout_sec=300, poll_sec=10): 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") diff --git a/simplyblock_core/constants.py b/simplyblock_core/constants.py index 8af6c0e2c..adb6ba408 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/health_controller.py b/simplyblock_core/controllers/health_controller.py index ac571ecf6..9bbf7ffaf 100644 --- a/simplyblock_core/controllers/health_controller.py +++ b/simplyblock_core/controllers/health_controller.py @@ -510,7 +510,7 @@ def _check_node_lvstore( dev = db_controller.get_storage_device_by_id(result['UUID']) dev_node = db_controller.get_storage_node_by_id(dev.node_id) if dev.status == NVMeDevice.STATUS_ONLINE and dev_node.status in [ - StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN, StorageNode.STATUS_UNREACHABLE]: + StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN, StorageNode.STATUS_UNREACHABLE, StorageNode.STATUS_SUSPENDED]: try: remote_bdev = storage_node_ops.connect_device( f"remote_{dev.alceml_bdev}", dev, node) @@ -713,7 +713,7 @@ def check_node(node_id, with_devices=True): logger.info(f"Node remote JMs: {len(snode.remote_jm_devices)}") for remote_device in snode.remote_jm_devices: - name = f'remote_{remote_device.jm_bdev}n1' + name = remote_device.remote_bdev bdev_info = rpc_client.get_bdevs(name) logger.log(INFO if bdev_info else ERROR, f"Checking bdev: {name} ... " + ('ok' if bdev_info else 'failed')) diff --git a/simplyblock_core/controllers/tasks_controller.py b/simplyblock_core/controllers/tasks_controller.py index e903dbb95..367368074 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: @@ -174,6 +216,12 @@ def _add_task(function_name, cluster_id, node_id, device_id, logger.info(f"Task found, skip adding new task: {task_id}") return False + elif function_name == JobSchedule.FN_NODE_REMOVAL: + task_id = get_active_node_removal_task(cluster_id, node_id) + if task_id: + logger.info(f"Task found, skip adding new task: {task_id}") + return False + task_obj = JobSchedule() task_obj.uuid = str(uuid.uuid4()) task_obj.cluster_id = cluster_id @@ -551,6 +599,10 @@ def get_active_node_mig_task(cluster_id, node_id, distr_name=None): def add_device_failed_mig_task(device_id): device = db.get_storage_device_by_id(device_id) for node in db.get_storage_nodes_by_cluster_id(device.cluster_id): + # IN_REMOVAL nodes have a dead SPDK (shut down by the removal flow); + # a migration task targeting their distribs can never run and would + # stall the node-removal completion check forever. Skip them like + # already-REMOVED nodes. if node.status == StorageNode.STATUS_REMOVED: continue for bdev in node.lvstore_stack: @@ -577,6 +629,23 @@ def add_node_add_task(cluster_id, function_params): function_params=function_params, max_retry=11) +def add_node_removal_task(cluster_id, node_id, function_params=None): + # max_retry=-1: the removal runner drives a multi-step, possibly multi-hour + # orchestration (shutdown -> LVS rewire -> device fail+migrate). Migration + # waits legitimately suspend-and-retry many times; do not cap retries. + return _add_task(JobSchedule.FN_NODE_REMOVAL, cluster_id, node_id, "", + function_params=function_params or {}, max_retry=-1) + + +def get_active_node_removal_task(cluster_id, node_id): + tasks = db.get_job_tasks(cluster_id) + for task in tasks: + if task.function_name == JobSchedule.FN_NODE_REMOVAL and task.node_id == node_id: + if task.status != JobSchedule.STATUS_DONE and task.canceled is False: + return task.uuid + return False + + def add_cluster_expand_task(cluster_id, new_node_id): """Queue a single-node cluster-expansion task. The runner drives the planner/orchestrator/executor to integrate ``new_node_id`` into the @@ -627,7 +696,7 @@ def get_active_node_tasks(cluster_id, node_id): tasks = db.get_job_tasks(cluster_id) out = [] for task in tasks: - if task.function_name in [JobSchedule.FN_PORT_ALLOW, JobSchedule.FN_JC_COMP_RESUME]: + if task.function_name in [JobSchedule.FN_PORT_ALLOW, JobSchedule.FN_JC_COMP_RESUME, JobSchedule.FN_NODE_REMOVAL]: continue if task.node_id == node_id: if task.status != JobSchedule.STATUS_DONE and task.canceled is False: diff --git a/simplyblock_core/env_var b/simplyblock_core/env_var index 5abdecd44..252ed2cc4 100644 --- a/simplyblock_core/env_var +++ b/simplyblock_core/env_var @@ -1,5 +1,5 @@ SIMPLY_BLOCK_COMMAND_NAME=sbcli-dev SIMPLY_BLOCK_VERSION=19.2.34 -SIMPLY_BLOCK_DOCKER_IMAGE=public.ecr.aws/simply-block/simplyblock:main +SIMPLY_BLOCK_DOCKER_IMAGE=public.ecr.aws/simply-block/simplyblock:feature-node-removal SIMPLY_BLOCK_SPDK_ULTRA_IMAGE=public.ecr.aws/simply-block/ultra:main-latest diff --git a/simplyblock_core/models/job_schedule.py b/simplyblock_core/models/job_schedule.py index 00564f4fc..48c8065b6 100644 --- a/simplyblock_core/models/job_schedule.py +++ b/simplyblock_core/models/job_schedule.py @@ -17,6 +17,7 @@ class JobSchedule(BaseModel): FN_FAILED_DEV_MIG = "failed_device_migration" FN_NEW_DEV_MIG = "new_device_migration" FN_NODE_ADD = "node_add" + FN_NODE_REMOVAL = "node_removal" FN_PORT_ALLOW = "port_allow" FN_BALANCING_AFTER_NODE_RESTART = "balancing_on_restart" FN_BALANCING_AFTER_DEV_REMOVE = "balancing_on_dev_rem" diff --git a/simplyblock_core/models/nvme_device.py b/simplyblock_core/models/nvme_device.py index 81aed2fb8..3e0955dba 100644 --- a/simplyblock_core/models/nvme_device.py +++ b/simplyblock_core/models/nvme_device.py @@ -108,6 +108,9 @@ class JMDevice(NVMeDevice): # the per-leg member partitions. Empty for single-device (no-raid) JMs. jm_leg_bdevs: List[str] = [] jm_leg_members: List = [] + # When attaching this JM to a node, override the device name on that node. + # This is needed when a JM device is removed and needed to be replaced, but the name must be the same. + override_name_on_node: dict[str, str] = {} # node_id: new_name class RemoteDevice(BaseModel): diff --git a/simplyblock_core/scripts/docker-compose-swarm.yml b/simplyblock_core/scripts/docker-compose-swarm.yml index e7b625568..5774a13f6 100644 --- a/simplyblock_core/scripts/docker-compose-swarm.yml +++ b/simplyblock_core/scripts/docker-compose-swarm.yml @@ -484,6 +484,20 @@ services: environment: SIMPLYBLOCK_LOG_LEVEL: "$LOG_LEVEL" + TasksNodeRemovalRunner: + <<: *service-base + image: $SIMPLYBLOCK_DOCKER_IMAGE + command: "python3 simplyblock_core/services/tasks_runner_node_removal.py" + deploy: + placement: + constraints: [node.role == manager] + volumes: + - "/etc/foundationdb:/etc/foundationdb" + networks: + - hostnet + environment: + SIMPLYBLOCK_LOG_LEVEL: "$LOG_LEVEL" + TasksRunnerClusterExpand: <<: *service-base image: $SIMPLYBLOCK_DOCKER_IMAGE diff --git a/simplyblock_core/services/health_check_service.py b/simplyblock_core/services/health_check_service.py index 35d40151c..e09923895 100644 --- a/simplyblock_core/services/health_check_service.py +++ b/simplyblock_core/services/health_check_service.py @@ -102,7 +102,7 @@ def check_node(snode): # (reconnecting remote devices, repairing multipath, recreating hublvols) — # but their node/device health is reported as "not applicable" (None), # never true/false. - report_health = snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN] + report_health = snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN, StorageNode.STATUS_SUSPENDED] # 1- check node ping ping_check = health_controller._check_node_ping(snode.mgmt_ip) diff --git a/simplyblock_core/services/storage_node_monitor.py b/simplyblock_core/services/storage_node_monitor.py index 0a9915577..f1f08d8e8 100644 --- a/simplyblock_core/services/storage_node_monitor.py +++ b/simplyblock_core/services/storage_node_monitor.py @@ -615,6 +615,7 @@ def _spawn_recovery_shutdown(node): def _run(): try: + storage_node_ops.shutdown_storage_node( node_id, force=True, keep_auto_restart=True) except Exception as e: diff --git a/simplyblock_core/services/tasks_runner_migration.py b/simplyblock_core/services/tasks_runner_migration.py index b9a7cc9aa..f159544c4 100644 --- a/simplyblock_core/services/tasks_runner_migration.py +++ b/simplyblock_core/services/tasks_runner_migration.py @@ -19,7 +19,7 @@ def _cluster_unavailable_state(cluster_id): for node in db.get_storage_nodes_by_cluster_id(cluster_id): if node.status in [StorageNode.STATUS_IN_CREATION, StorageNode.STATUS_REMOVED]: continue - if node.status != StorageNode.STATUS_ONLINE: + if node.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: unavailable.append(f"node:{node.get_id()}") for dev in node.nvme_devices: if dev.status in [NVMeDevice.STATUS_REMOVED, NVMeDevice.STATUS_FAILED_AND_MIGRATED]: @@ -75,7 +75,7 @@ def task_runner(task): task.write_to_db(db.kv_store) return True - if snode.status != StorageNode.STATUS_ONLINE: + if snode.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: task.function_result = "node is not online, retrying" task.status = JobSchedule.STATUS_SUSPENDED unavailable = _cluster_unavailable_state(task.cluster_id) @@ -307,7 +307,7 @@ def _set_master_task_status(master_task, status): logger.info("no task found on same node, resuming compression") node = db.get_storage_node_by_id(task.node_id) for n in db.get_storage_nodes_by_cluster_id(node.cluster_id): - if n.status != StorageNode.STATUS_ONLINE: + if n.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: logger.warning("Not all nodes are online, can not resume JC compression") continue rpc_client = node.rpc_client(timeout=5, retry=2) diff --git a/simplyblock_core/services/tasks_runner_node_removal.py b/simplyblock_core/services/tasks_runner_node_removal.py new file mode 100644 index 000000000..a90fbc36e --- /dev/null +++ b/simplyblock_core/services/tasks_runner_node_removal.py @@ -0,0 +1,98 @@ +# coding=utf-8 +import time + + +from simplyblock_core import db_controller, storage_node_ops, utils, constants +from simplyblock_core.controllers import tasks_controller +from simplyblock_core.models.job_schedule import JobSchedule +from simplyblock_core.models.cluster import Cluster + + +logger = utils.get_logger(__name__) + +# get DB controller +db = db_controller.DBController() + + +def process_task(task): + """Advance one node-removal task by one orchestration pass. + + node_removal_orchestrate is idempotent and resumable: it returns True only + when the node is fully REMOVED, and False to mean "incomplete, retry later" + (most commonly: device failure-migration still in progress). On False we + suspend the task so the outer loop revisits it on the next tick instead of + busy-spinning here for what can be hours. + """ + if task.canceled: + task.function_result = "canceled" + task.status = JobSchedule.STATUS_DONE + task.write_to_db(db.kv_store) + return True + + cluster = db.get_cluster_by_id(task.cluster_id) + if cluster.status == Cluster.STATUS_IN_ACTIVATION: + task.function_result = "cluster is in_activation, waiting" + task.status = JobSchedule.STATUS_SUSPENDED + task.write_to_db(db.kv_store) + return False + + if task.status != JobSchedule.STATUS_RUNNING: + task.status = JobSchedule.STATUS_RUNNING + task.write_to_db(db.kv_store) + + force_remove = bool(task.function_params.get("force_remove", False)) + try: + done = storage_node_ops.node_removal_orchestrate(task.node_id, force_remove=force_remove) + except Exception as e: + logger.error(f"Node-removal task {task.uuid} raised: {e}") + logger.exception(e) + task.function_result = f"error: {e}" + task.retry += 1 + task.status = JobSchedule.STATUS_SUSPENDED + task.write_to_db(db.kv_store) + return False + + if done: + task.function_result = "Node removed" + task.status = JobSchedule.STATUS_DONE + task.write_to_db(db.kv_store) + return True + + # Incomplete: a phase asked us to retry (typically waiting on migration). + task.function_result = "removal in progress, retrying" + task.retry += 1 + task.status = JobSchedule.STATUS_SUSPENDED + task.write_to_db(db.kv_store) + return False + + +logger.info("Starting Tasks runner node removal...") + +while True: + time.sleep(constants.TASK_EXEC_INTERVAL_SEC) + try: + clusters = db.get_clusters() + except Exception as e: + logger.error(f"Failed to get clusters: {e}") + continue + if not clusters: + logger.error("No clusters found!") + continue + for cl in clusters: + tasks = db.get_job_tasks(cl.get_id(), reverse=False) + for task in tasks: + if task.function_name != JobSchedule.FN_NODE_REMOVAL: + continue + if task.status == JobSchedule.STATUS_DONE: + continue + # get a fresh object: cancel/other writers may have changed it + task = db.get_task_by_id(task.uuid) + # Lease gate: skip a task another live runner host owns. + if not tasks_controller.claim_task(task): + logger.info(f"Node-removal task {task.uuid} owned by another runner host; skipping") + continue + try: + process_task(task) + except Exception as e: + logger.error(f"Node-removal task {task.uuid} processing crashed: {e}") + logger.exception(e) diff --git a/simplyblock_core/services/tasks_runner_restart.py b/simplyblock_core/services/tasks_runner_restart.py index 4e36c5a9f..0e374a8bc 100644 --- a/simplyblock_core/services/tasks_runner_restart.py +++ b/simplyblock_core/services/tasks_runner_restart.py @@ -500,6 +500,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}") + + # Parallel restart execution for SUSPENDED clusters: during full-cluster # recovery every node is offline and no client IO flows, so node restarts # cannot violate FTT and are fanned out on this pool (~70 s each; strictly @@ -644,4 +720,9 @@ def _process_restart_task(task_uuid): # `while True` and kill recovery of every other node. _process_restart_task(task.uuid) + 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/settings.py b/simplyblock_core/settings.py index 836c4f9ce..748d27ff3 100644 --- a/simplyblock_core/settings.py +++ b/simplyblock_core/settings.py @@ -62,7 +62,7 @@ class Settings(BaseSettings): "and the response stream has no type information to mask by." ) ), - ] = False + ] = True @model_validator(mode="after") def validate_tls_files(self): diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 453d68a02..72d1846d4 100755 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -1,4 +1,5 @@ # coding=utf- 8 +import copy import datetime import json import math @@ -26,7 +27,6 @@ device_controller, tasks_controller, health_controller, tcp_ports_events, qos_controller from simplyblock_core.controllers.host_auth import _reapply_allowed_hosts from simplyblock_core.db_controller import DBController -from simplyblock_core.exceptions import PreconditionError from simplyblock_core.models.iface import IFace from simplyblock_core.models.job_schedule import JobSchedule from simplyblock_core.models.lvol_model import LVol @@ -1249,7 +1249,7 @@ def _connect_to_remote_devs( remote_devices = [] existing_remote_devices = {dev.get_id(): dev for dev in this_node.remote_devices} - allowed_node_statuses = [StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN] + allowed_node_statuses = [StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN, StorageNode.STATUS_SUSPENDED] allowed_dev_statuses = [NVMeDevice.STATUS_ONLINE, NVMeDevice.STATUS_READONLY, NVMeDevice.STATUS_CANNOT_ALLOCATE] if force_connect_restarting_nodes: @@ -1633,9 +1633,14 @@ def _connect_to_remote_jm_devs(this_node, jm_ids=None, only_node_id=None): remote_device.status = NVMeDevice.STATUS_ONLINE remote_device.nvmf_multipath = org_dev.nvmf_multipath expected_bdev = f"remote_{org_dev.jm_bdev}n1" + controller_name = f"remote_{org_dev.jm_bdev}" + if org_dev.override_name_on_node and this_node.get_id() in org_dev.override_name_on_node: + new_name = org_dev.override_name_on_node[this_node.get_id()] + expected_bdev = f"remote_{new_name}n1" + controller_name = f"remote_{new_name}" try: remote_device.remote_bdev = str(connect_device( - f"remote_{org_dev.jm_bdev}", org_dev, this_node, + controller_name, org_dev, this_node, attach_timeout=1, )) except (RuntimeError, RPCException): @@ -2588,6 +2593,32 @@ def delete_storage_node(node_id, force=False): def remove_storage_node(node_id, force_remove=False, force_migrate=False): + """Start the online removal of a storage node from its cluster. + + This is the inverse of cluster expansion (add_node). It validates the + preconditions, then queues a background FN_NODE_REMOVAL task; the + tasks_runner_node_removal service drives the multi-step, possibly + multi-hour orchestration (see ``node_removal_orchestrate``): + + shutdown -> in_removal -> rewire LVS replicas -> remove/fail/migrate + devices -> removed + + Preconditions (all enforced here, before anything is queued): + * the target node is ONLINE; + * every other (non-removed) node in the cluster is ONLINE; + * FTT headroom allows losing this node (``_check_ftt_allows_node_removal``); + * the node hosts NO LVols and NO snapshots (the operator migrates those + separately, at a higher level — see the design decision for this + feature); + * any secondary/tertiary replica this node hosts for OTHER primaries has + a valid host-disjoint relocation target. + + ``force_remove`` only bypasses the active-task guard (cancelling them). + ``force_migrate`` is accepted for signature compatibility and ignored: + LVol migration is no longer part of node removal. + + Returns the new task uuid on success, or False on a rejected precondition. + """ db_controller = DBController() try: snode = db_controller.get_storage_node_by_id(node_id) @@ -2595,81 +2626,421 @@ def remove_storage_node(node_id, force_remove=False, force_migrate=False): logger.exception("Can not find storage node") return False - if snode.status == StorageNode.STATUS_ONLINE: - logger.warning(f"Can not remove online node: {node_id}") + # Idempotent re-entry: a removal already in flight just returns its task. + existing = tasks_controller.get_active_node_removal_task(snode.cluster_id, node_id) + if existing: + logger.info(f"Node removal already in progress for {node_id} (task {existing})") + return existing + + if snode.status == StorageNode.STATUS_REMOVED: + logger.warning(f"Node already removed: {node_id}") + return False + + if snode.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: + logger.error( + f"Can not remove node {node_id}: it must be ONLINE or SUSPENDED to start removal " + f"(current status: {snode.status}). Removal shuts the node down itself.") + return False + + # All other nodes must be online — removal rewires LVS replicas and drives + # device migration across the surviving nodes, which requires every peer up. + not_online = [ + n for n in db_controller.get_storage_nodes_by_cluster_id(snode.cluster_id) + if n.get_id() != node_id + and n.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_REMOVED) + ] + if not_online: + offending = ", ".join(f"{n.get_id()}({n.status})" for n in not_online) + logger.error( + f"Can not remove node {node_id}: all nodes must be online. " + f"Not-online node(s): {offending}") + return False + + allowed, reason = _check_ftt_allows_node_removal(node_id, db_controller) + if not allowed: + logger.error(f"Can not remove node {node_id}: {reason}") + return False + + lvols = db_controller.get_lvols_by_node_id(node_id) + if lvols: + logger.error( + f"Can not remove node {node_id}: {len(lvols)} LVol(s) present. " + f"Migrate or delete them first.") + return False + + node_snaps = [ + sn for sn in db_controller.get_snapshots() + if sn.lvol.node_id == node_id and sn.deleted is False + ] + if node_snaps: + logger.error( + f"Can not remove node {node_id}: {len(node_snaps)} snapshot(s) present. " + f"Remove them first.") return False tasks = tasks_controller.get_active_node_tasks(snode.cluster_id, snode.get_id()) if tasks: - logger.warning(f"Task found: {len(tasks)}, can not remove storage node, or use --force") + logger.warning(f"Task found: {len(tasks)}, can not remove storage node, or use --force-remove") if force_remove is False: return False for task in tasks: tasks_controller.cancel_task(task.uuid) - lvols = db_controller.get_lvols_by_node_id(node_id) - if lvols: - if force_migrate: - for lvol in lvols: + # Case-B feasibility: every replica this node hosts for another primary must + # have somewhere host-disjoint to go. Catches e.g. 2-node clusters where the + # tertiary cannot be re-placed without violating anti-affinity. + feasible, reason = _check_replica_relocation_feasible(snode, db_controller) + if not feasible: + logger.error(f"Can not remove node {node_id}: {reason}") + return False + + task_id = tasks_controller.add_node_removal_task( + snode.cluster_id, node_id, {"force_remove": force_remove}) + if not task_id: + logger.error(f"Failed to queue node-removal task for {node_id}") + return False + logger.info(f"Node removal queued for {node_id}: task {task_id}") + return task_id + + +def _check_replica_relocation_feasible(removed_node, db_controller): + """Pre-flight Case-B check: a secondary/tertiary replica hosted on + ``removed_node`` for some OTHER primary must have a valid relocation target. + Returns (feasible: bool, reason: str).""" + for backref, picker in ( + ("lvstore_stack_secondary", "secondary"), + ("lvstore_stack_tertiary", "tertiary")): + primary_id = getattr(removed_node, backref) + if not primary_id: + continue + try: + primary = db_controller.get_storage_node_by_id(primary_id) + except KeyError: + # The primary is gone; nothing to relocate, just bookkeeping. + continue + if not _pick_replica_relocation_node(primary, removed_node, picker, db_controller): + return False, ( + f"no host-disjoint node available to re-host the {picker} replica " + f"of primary {primary_id} (currently on the node being removed)") + return True, "" + + +def _pick_replica_relocation_node(primary, removed_node, role, db_controller): + """Choose a node to re-host ``primary``'s ``role`` (secondary|tertiary) + replica, currently on ``removed_node``. Returns a node id or None. + Reuses the existing anti-affinity-aware placement helpers.""" + exclude_ids = [removed_node.get_id()] + if role == "secondary": + if primary.tertiary_node_id and primary.tertiary_node_id != removed_node.get_id(): + exclude_ids.append(primary.tertiary_node_id) + cands = get_secondary_nodes(primary, exclude_ids=exclude_ids, removed_node=removed_node) + else: + exclude_mgmt_ips = [] + if primary.secondary_node_id and primary.secondary_node_id != removed_node.get_id(): + exclude_ids.append(primary.secondary_node_id) + try: + sec = db_controller.get_storage_node_by_id(primary.secondary_node_id) + exclude_mgmt_ips.append(sec.mgmt_ip) + except KeyError: pass - # lvol_controller.migrate(lvol_id) - elif force_remove: - for lvol in lvols: - try: - lvol_controller.delete_lvol(lvol, force_delete=True) - except (PreconditionError, RuntimeError): - logger.warning("Failed to delete volume", exc_info=True) - else: - logger.warning("LVols found on the storage node, use --force-remove or --force-migrate") + cands = get_secondary_nodes_2( + primary, exclude_ids=exclude_ids, exclude_mgmt_ips=exclude_mgmt_ips) + return cands[0] if cands else None + + +def node_removal_orchestrate(node_id, force_remove=False): + """Idempotent, resumable orchestration driven by tasks_runner_node_removal. + + Returns True only when the node has been fully removed (status REMOVED). + Returns False to signal "incomplete, retry later" (e.g. device migration + still in progress) — the runner suspends the task and revisits it. + + Every phase is guarded so a re-entry after a crash/retry skips already + completed work. + """ + db_controller = DBController() + try: + snode = db_controller.get_storage_node_by_id(node_id) + except KeyError: + logger.error(f"node_removal_orchestrate: node {node_id} not found") + return False + + if snode.status == StorageNode.STATUS_REMOVED: + return True + + # Phase 1 — shut the node down (graceful). Skipped on re-entry. + if snode.status in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: + logger.info(f"[REMOVAL] {node_id}: phase 1 — shutdown") + ret = shutdown_storage_node(node_id, force=force_remove) + if isinstance(ret, tuple): + ret, reason = ret + if not ret: + logger.error(f"[REMOVAL] {node_id}: shutdown failed: {reason}") + return False + elif not ret: + logger.error(f"[REMOVAL] {node_id}: shutdown failed") return False + snode = db_controller.get_storage_node_by_id(node_id) - snaps = db_controller.get_snapshots() - node_snaps = [] - for sn in snaps: - if sn.lvol.node_id == node_id and sn.deleted is False: - node_snaps.append(sn) + # Phase 3a — tear down the (empty) secondary/tertiary replicas of THIS + # node's own primary LVS, on the peers that host them (Case A). + logger.info(f"[REMOVAL] {node_id}: phase 3a — tear down own replicas") + if not _teardown_replicas_of_primary(snode): + return False - if node_snaps: - if force_migrate: - logger.error("Not implemented!") + # Phase 3b — relocate replicas this node hosts for OTHER primaries (Case B). + logger.info(f"[REMOVAL] {node_id}: phase 3b — relocate hosted replicas") + if not _relocate_replicas_hosted_on(snode): + return False + + # Phase 5 — finalize (swarm leave, gpt cleanup) and flip to removed. + logger.info(f"[REMOVAL] {node_id}: phase 4 — finalize") + _finalize_node_removal(snode) + set_node_status(node_id, StorageNode.STATUS_REMOVED, caused_by="remove") + snode = db_controller.get_storage_node_by_id(node_id) + # storage_events.snode_status_change( + # snode, StorageNode.STATUS_REMOVED, StorageNode.STATUS_IN_REMOVAL, caused_by="remove") + + # Phase 4 — remove + fail devices, then wait for failure-migration to finish. + logger.info(f"[REMOVAL] {node_id}: phase 5 — devices remove/fail/migrate") + if not _decommission_node_devices(snode): + return False + + logger.info(f"[REMOVAL] {node_id}: done") + return True + + +def _teardown_replicas_of_primary(removed_node): + """Case A: the primary LVS lives on ``removed_node`` (now shut down). + Delete its secondary/tertiary replicas from the peers that host them and + clear the cross-reference bookkeeping. The node has no LVols (enforced at + entry), so the replicas hold only the empty lvstore + hublvol.""" + db_controller = DBController() + removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) + cluster = db_controller.get_cluster_by_id(removed_node.cluster_id) + + for field, backref in ( + ("secondary_node_id", "lvstore_stack_secondary"), + ("tertiary_node_id", "lvstore_stack_tertiary")): + peer_id = getattr(removed_node, field) + if not peer_id: + continue + try: + peer = db_controller.get_storage_node_by_id(peer_id) + except KeyError: + removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) + setattr(removed_node, field, "") + removed_node.write_to_db() + continue + + if peer.status == StorageNode.STATUS_ONLINE: + _delete_replica_on_peer(peer, removed_node, cluster) + + # Clear the peer's back-reference if it still points at us. + peer = db_controller.get_storage_node_by_id(peer_id) + if getattr(peer, backref) == removed_node.get_id(): + setattr(peer, backref, "") + peer.write_to_db() + + removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) + setattr(removed_node, field, "") + removed_node.write_to_db() + + return True + + +def _delete_replica_on_peer(peer, primary, cluster): + """Best-effort teardown of ``primary``'s replica lvstore (+ hublvol) on the + online ``peer``. RPC failures are logged, not fatal: the peer is healthy and + a lingering empty bdev is harmless, while blocking removal on it is not.""" + rpc_client = peer.rpc_client() + lvstore = primary.lvstore + if not lvstore: + return + try: + nqn = peer.hublvol_nqn_for_lvstore(cluster.nqn, lvstore) + if rpc_client.subsystem_list(nqn): + rpc_client.subsystem_delete(nqn) + except Exception as e: + logger.warning(f"hublvol subsystem teardown for {lvstore} on {peer.get_id()} failed: {e}") + try: + rpc_client.bdev_lvol_delete_hublvol(lvstore) + except Exception as e: + logger.warning(f"hublvol bdev teardown for {lvstore} on {peer.get_id()} failed: {e}") + try: + # deepcopy: _remove_bdev_stack stamps bdev['status']; don't mutate the + # primary's stored stack definition. + _remove_bdev_stack(copy.deepcopy(primary.lvstore_stack), rpc_client) + except Exception as e: + logger.warning(f"replica bdev-stack teardown for {lvstore} on {peer.get_id()} failed: {e}") + + +def _relocate_replicas_hosted_on(removed_node): + """Case B: ``removed_node`` holds a secondary and/or tertiary replica for + other primaries. Re-host each on a fresh, anti-affinity-valid node so the + owning primary keeps its fault tolerance after this node leaves.""" + db_controller = DBController() + + removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) + if removed_node.lvstore_stack_secondary: + if not _relocate_one_replica(removed_node, removed_node.lvstore_stack_secondary, "secondary"): return False - elif force_remove: - for sn in node_snaps: - snapshot_controller.delete(sn.get_id()) - else: - logger.error("Snapshots found on the storage node, use --force-remove or --force-migrate") + + removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) + if removed_node.lvstore_stack_tertiary: + if not _relocate_one_replica(removed_node, removed_node.lvstore_stack_tertiary, "tertiary"): return False - if snode.nvme_devices: - for dev in snode.nvme_devices: - if dev.status == NVMeDevice.STATUS_ONLINE: - distr_controller.disconnect_device(dev) + return True - if snode.jm_device and snode.jm_device.get_id() and snode.jm_device.status in [JMDevice.STATUS_ONLINE, - JMDevice.STATUS_UNAVAILABLE]: - logger.info("Removing JM") - device_controller.remove_jm_device(snode.jm_device.get_id(), force=True) - cluster = db_controller.get_cluster_by_id(snode.cluster_id) +def _relocate_one_replica(removed_node, primary_id, role): + """Re-host ``primary_id``'s ``role`` replica off ``removed_node``. + + Idempotent: the back-reference on ``removed_node`` is cleared only AFTER the + replica is successfully rebuilt on the new node, so a retry resumes cleanly. + """ + db_controller = DBController() + field = "secondary_node_id" if role == "secondary" else "tertiary_node_id" + backref = "lvstore_stack_secondary" if role == "secondary" else "lvstore_stack_tertiary" + + try: + primary = db_controller.get_storage_node_by_id(primary_id) + except KeyError: + # Primary gone — nothing to rebuild, just drop the stale back-reference. + _clear_replica_backref(removed_node, backref) + return True + + # Choose (or recover) the relocation target. If the primary's pointer still + # names the removed node we must pick a fresh one and commit the forward + # bookkeeping; on a retry it already names the new node, so reuse it. + new_id = getattr(primary, field) + if not new_id or new_id == removed_node.get_id(): + new_id = _pick_replica_relocation_node(primary, removed_node, role, db_controller) + if not new_id: + logger.error( + f"[REMOVAL] no relocation target for {role} replica of {primary_id}") + return False + primary = db_controller.get_storage_node_by_id(primary_id) + setattr(primary, field, new_id) + primary.write_to_db() + new_node = db_controller.get_storage_node_by_id(new_id) + setattr(new_node, backref, primary.get_id()) + new_node.write_to_db() + + new_node = db_controller.get_storage_node_by_id(new_id) + primary = db_controller.get_storage_node_by_id(primary_id) + + # Build the replica on the new node. The primary is online and remains the + # leader, so recreate_lvstore_on_non_leader wires distribs/raid/lvstore, + # role + ANA, and the hublvol connection exactly as the restart path does. + ret = recreate_lvstore_on_non_leader(new_node, primary, primary) + if not ret: + logger.error( + f"[REMOVAL] failed to rebuild {role} replica of {primary_id} on {new_id}, will retry") + return False + + _clear_replica_backref(removed_node, backref) + return True + + +def _clear_replica_backref(removed_node, backref): + db_controller = DBController() + removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) + if getattr(removed_node, backref): + setattr(removed_node, backref, "") + removed_node.write_to_db() + + +def _decommission_node_devices(removed_node): + """Remove, fail and migrate every data device on ``removed_node``. + + Drives each device ONLINE/UNAVAILABLE -> REMOVED -> FAILED (which queues the + failure-migration tasks on the surviving online nodes), then waits for them + all to reach FAILED_AND_MIGRATED. Returns True only once every data device + is migrated; False means "still migrating, retry later".""" + db_controller = DBController() + removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) + + if (removed_node.jm_device and removed_node.jm_device.get_id() + and removed_node.jm_device.status in (JMDevice.STATUS_ONLINE, JMDevice.STATUS_UNAVAILABLE)): + logger.info(f"[REMOVAL] {removed_node.get_id()}: removing JM device") + device_controller.remove_jm_device(removed_node.jm_device.get_id(), force=True) + # look for other nodes who use this JM and replace it + for node in db_controller.get_storage_nodes_by_cluster_id(removed_node.cluster_id): + if node.jm_ids and removed_node.jm_device.get_id() in node.jm_ids: + node.jm_ids.remove(removed_node.jm_device.get_id()) + jm_ids = get_sorted_ha_jms(node) + logger.debug(f"online_jms: {str(jm_ids)}") + new_jm_dev = "" + for jm_id in jm_ids: + if jm_id not in node.jm_ids: + new_jm_dev = jm_id + break + if new_jm_dev: + d = db_controller.get_jm_device_by_id(new_jm_dev) + jm_node = db_controller.get_storage_node_by_id(d.node_id) + jm_node.jm_device.override_name_on_node[node.get_id()]=removed_node.jm_device.jm_bdev + jm_node.write_to_db() + node.jm_ids.append(new_jm_dev) + node.remote_jm_devices = _connect_to_remote_jm_devs(node, node.jm_ids) + node.write_to_db() + else: + logger.error(f"no jm_id found for {node.get_id()}") + + removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) + for dev in removed_node.nvme_devices: + if dev.status in (NVMeDevice.STATUS_JM, NVMeDevice.STATUS_FAILED_AND_MIGRATED): + continue + if dev.status in (NVMeDevice.STATUS_ONLINE, NVMeDevice.STATUS_UNAVAILABLE): + # force=True tolerates the dead local SPDK (it was killed at shutdown); + # the meaningful work — disconnect from peers + DB state — still runs. + # device_controller.device_remove(dev.get_id(), force=True) + device_controller.device_set_state(dev.get_id(), NVMeDevice.STATUS_REMOVED) + + fresh = db_controller.get_storage_device_by_id(dev.get_id()) + if fresh.status == NVMeDevice.STATUS_REMOVED: + device_controller.device_set_failed(dev.get_id()) + + # Completion gate: every non-JM device migrated. + removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) + for dev in removed_node.nvme_devices: + if dev.status in (NVMeDevice.STATUS_JM, NVMeDevice.STATUS_FAILED_AND_MIGRATED): + continue + logger.info( + f"[REMOVAL] {removed_node.get_id()}: device {dev.get_id()} " + f"status={dev.status}, migration not complete") + # return False + + return True + + +def _finalize_node_removal(removed_node): + """Best-effort host cleanup before the node is flipped to REMOVED: leave the + docker swarm and wipe GPT partitions. Mirrors the tail of the legacy + offline-removal path.""" + db_controller = DBController() + removed_node = db_controller.get_storage_node_by_id(removed_node.get_id()) + cluster = db_controller.get_cluster_by_id(removed_node.cluster_id) if cluster.mode == "docker": logger.info("Leaving swarm...") try: - cluster_docker = utils.get_docker_client(snode.cluster_id) + cluster_docker = utils.get_docker_client(removed_node.cluster_id) for node in cluster_docker.nodes.list(): - if node.attrs["Status"] and snode.mgmt_ip in node.attrs["Status"]["Addr"]: + if node.attrs["Status"] and removed_node.mgmt_ip in node.attrs["Status"]["Addr"]: node.remove(force=True) except Exception: pass try: - if health_controller._check_node_api(snode): + if health_controller._check_node_api(removed_node): logger.info("Stopping SPDK container") - snode_api = snode.client(timeout=20) - snode_api.spdk_process_kill(snode.rpc_port, snode.cluster_id) + snode_api = removed_node.client(timeout=20) + snode_api.spdk_process_kill(removed_node.rpc_port, removed_node.cluster_id) snode_api.leave_swarm() pci_address = [] - for dev in snode.nvme_devices: + for dev in removed_node.nvme_devices: if dev.pcie_address not in pci_address: ret = snode_api.delete_dev_gpt_partitions(dev.pcie_address) logger.debug(ret) @@ -2677,13 +3048,6 @@ def remove_storage_node(node_id, force_remove=False, force_migrate=False): except Exception as e: logger.exception(e) - set_node_status(node_id, StorageNode.STATUS_REMOVED) - - for dev in snode.nvme_devices: - if dev.status in [NVMeDevice.STATUS_JM, NVMeDevice.STATUS_FAILED_AND_MIGRATED]: - continue - device_controller.device_set_failed(dev.get_id()) - logger.info("done") @@ -2722,6 +3086,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( @@ -2738,6 +3134,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" @@ -3497,6 +3894,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 @@ -3519,16 +3927,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 @@ -3573,7 +3971,7 @@ def list_storage_nodes(cluster_id=None): "Status": node.status, # Health is only meaningful for ONLINE/DOWN nodes; otherwise N/A. "Health": node.health_check if node.status in ( - StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN) else "-", + StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN, StorageNode.STATUS_SUSPENDED) else "-", "Up time": utils.strfdelta(uptime) if (uptime := node.uptime()) is not None else "", "CPU": f"{len(utils.hexa_to_cpu_list(node.spdk_cpu_mask))}", "MEM": utils.humanbytes(node.spdk_mem), @@ -3621,7 +4019,7 @@ def list_storage_devices(node_id): "IO Err": device.io_error, # Device health is only meaningful when its node is ONLINE/DOWN. "Health": device.health_check if snode.status in ( - StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN) else "-" + StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN, StorageNode.STATUS_SUSPENDED) else "-" }) for bdev in snode.lvstore_stack: @@ -3650,7 +4048,7 @@ def list_storage_devices(node_id): "Status": snode.jm_device.status, "IO Err": snode.jm_device.io_error, "Health": snode.jm_device.health_check if snode.status in ( - StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN) else "-" + StorageNode.STATUS_ONLINE, StorageNode.STATUS_DOWN, StorageNode.STATUS_SUSPENDED) else "-" }) for remote_device in snode.remote_devices: @@ -4203,44 +4601,80 @@ def shutdown_storage_node(node_id, force=False, keep_auto_restart=False): return True -def suspend_storage_node(node_id, force=False, change_node_status=True): - """Deprecated: the suspension phase is no longer a precursor to - graceful shutdown. Kept as a noop-stub so any external automation - still calling `sbctl sn suspend` doesn't hard-fail. - - Earlier revisions blocked sec/tert + own-primary lvstore ports via - iptables-`REJECT --reject-with tcp-reset` here, then transitioned - the node to STATUS_SUSPENDED. That fence cannot stop SPDK's lvol - layer from resubmitting failed-redirect IO inside the dying node - as if it were new host IO, which races the surviving sec/tert - peer's auto-promotion and produces a writer conflict (incident - 2026-05-19, jm_vuid=4818). The new shutdown_storage_node() drops - the entire suspension phase and instead relies on: - (a) device-unavailable events to peers (already part of the - old flow, retained in shutdown_storage_node), - (b) bdev_nvme_detach_controller on every peer's remote_alceml - / remote_jm controller pointing at the dying node (new - Loop 2 in shutdown_storage_node), and - (c) a hard SPDK kill — every condition we used to "drain" is - now resolved post-kill by the peers' normal recovery path. +def suspend_storage_node(node_id: str, force=False): """ - logger.warning( - "sn suspend is deprecated: the suspension phase has been removed " - "from graceful shutdown (see shutdown_storage_node docstring). " - "Treating call as no-op for node %s.", - node_id, - ) + Suspends a storage node by changing its status to suspended if the node is in an online state. + + This would exclude this node from lvol host selection, see: lvol_controller._get_next_3_nodes + + Parameters + ---------- + node_id : int + The unique identifier of the storage node to be suspended. + force : bool, optional + A flag to indicate whether to forcibly suspend the node. Currently unused, defaults to False. + + Returns + ------- + bool + Returns True if the node was successfully suspended, or False if the operation failed. + """ + db_controller = DBController() + try: + snode = db_controller.get_storage_node_by_id(node_id) + except KeyError: + logger.exception("This storage node is not part of the cluster") + return False + + logger.info("Node found: %s in state: %s", snode.hostname, snode.status) + if snode.status != StorageNode.STATUS_ONLINE: + logger.error("Node is not in online state") + return False + + logger.info("Setting node status to suspended") + set_node_status(snode.get_id(), StorageNode.STATUS_SUSPENDED) + + snode = db_controller.get_storage_node_by_id(node_id) + snode.auto_restart_disabled = True + snode.write_to_db(db_controller.kv_store) + + logger.info("Done") return True def resume_storage_node(node_id): - """Deprecated: counterpart to suspend_storage_node, which is now a - noop. There is nothing to resume.""" - logger.warning( - "sn resume is deprecated: the suspension phase has been removed " - "from graceful shutdown. Treating call as no-op for node %s.", - node_id, - ) + """ + Resumes a storage node currently in a suspended state. + + This function sets the node status to online. + + Parameters: + node_id: int + The unique identifier of the storage node to resume. + + Returns: + bool + True if the storage node was successfully resumed, False otherwise. + """ + db_controller = DBController() + try: + snode = db_controller.get_storage_node_by_id(node_id) + except KeyError: + logger.error("This storage node is not part of the cluster") + return False + + logger.info("Node found: %s in state: %s", snode.hostname, snode.status) + if snode.status != StorageNode.STATUS_SUSPENDED: + logger.error("Node is not in suspended state") + return False + + logger.info("Setting node status to online") + set_node_status(snode.get_id(), StorageNode.STATUS_ONLINE) + snode = db_controller.get_storage_node_by_id(node_id) + snode.auto_restart_disabled = False + snode.write_to_db(db_controller.kv_store) + + logger.info("Done") return True @@ -7377,12 +7811,15 @@ def get_node_jm_names(current_node, remote_node=None): continue jm_dev = DBController().get_jm_device_by_id(jm_id) - jm_list.append(f"remote_{jm_dev.jm_bdev}n1") + if jm_dev.override_name_on_node and current_node.get_id() in jm_dev.override_name_on_node: + jm_list.append(f"remote_{jm_dev.override_name_on_node[current_node.get_id()]}n1") + else: + jm_list.append(f"remote_{jm_dev.jm_bdev}n1") return jm_list[:current_node.ha_jm_count] -def get_secondary_nodes(current_node, exclude_ids=None): +def get_secondary_nodes(current_node, exclude_ids=None, removed_node=None): if exclude_ids is None: exclude_ids = [] db_controller = DBController() @@ -7417,6 +7854,9 @@ def _candidates(forbidden_fds, forbidden_labels): if nod_found: return [node.get_id()] + elif removed_node and node.get_id() == removed_node.secondary_node_id: + nodes.append(node.get_id()) + return nodes # Anti-affinity is best-effort and honored on BOTH failure domain and @@ -7701,20 +8141,19 @@ def create_lvstore(snode, ndcs, npcs, distr_bs, distr_chunk_bs, page_size_in_blo sec_node.write_to_db() - # Create hublvol on primary after all secondaries have their stacks - if secondary_ids: - cluster = db_controller.get_cluster_by_id(snode.cluster_id) - try: - snode.create_hublvol(cluster_nqn=cluster.nqn) - except RPCException as e: - logger.error("Error establishing hublvol: %s", e.message) - # return False + cluster = db_controller.get_cluster_by_id(snode.cluster_id) + try: + snode.create_hublvol(cluster_nqn=cluster.nqn) + except RPCException as e: + logger.error("Error establishing hublvol: %s", e.message) + # return False - try: - snode.create_transfer_hublvol() - except RPCException as e: - logger.error("Error creating transfer hublvol: %s", e.message) + try: + snode.create_transfer_hublvol() + except RPCException as e: + logger.error("Error creating transfer hublvol: %s", e.message) + if secondary_ids: # Create secondary hublvol on sec_1 so tertiary can multipath sec1 = db_controller.get_storage_node_by_id(secondary_ids[0]) if sec1 and sec1.status == StorageNode.STATUS_ONLINE: 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( diff --git a/tests/integration/test_activate_portblock_and_suspend_no_demote.py b/tests/integration/test_activate_portblock_and_suspend_no_demote.py index 8699aca80..ced13db26 100644 --- a/tests/integration/test_activate_portblock_and_suspend_no_demote.py +++ b/tests/integration/test_activate_portblock_and_suspend_no_demote.py @@ -339,10 +339,11 @@ class TestSuspendIsDeprecatedNoop(unittest.TestCase): def test_suspend_is_noop_and_does_not_touch_firewall(self): from simplyblock_core import storage_node_ops - snode = _node("node-A", lvstore="LVS_A") + snode = _node("node-A", lvstore="LVS_A", status=StorageNode.STATUS_ONLINE) with patch.object(storage_node_ops, "port_block") as pb, \ patch.object(storage_node_ops, "DBController") as _db: + _db.get_storage_node_by_id.return_value = snode self.assertTrue( storage_node_ops.suspend_storage_node(snode.get_id())) pb.set_port.assert_not_called() @@ -350,10 +351,11 @@ def test_suspend_is_noop_and_does_not_touch_firewall(self): def test_resume_is_noop_and_does_not_touch_firewall(self): from simplyblock_core import storage_node_ops - snode = _node("node-A", lvstore="LVS_A") + snode = _node("node-A", lvstore="LVS_A", status=StorageNode.STATUS_SUSPENDED) with patch.object(storage_node_ops, "port_block") as pb, \ patch.object(storage_node_ops, "DBController") as _db: + _db.get_storage_node_by_id.return_value = snode self.assertTrue( storage_node_ops.resume_storage_node(snode.get_id())) pb.set_port.assert_not_called() diff --git a/tests/integration/test_shutdown_no_suspension.py b/tests/integration/test_shutdown_no_suspension.py index 9a6e413b6..c613bafba 100644 --- a/tests/integration/test_shutdown_no_suspension.py +++ b/tests/integration/test_shutdown_no_suspension.py @@ -315,29 +315,6 @@ def test_returns_zero_when_no_peers(self): sno._detach_remote_controllers_from_peers(dying, db), 0) -# --------------------------------------------------------------------------- -# 3. suspend_storage_node / resume_storage_node are deprecated noops -# --------------------------------------------------------------------------- - - -class TestSuspendAndResumeAreNoops(unittest.TestCase): - """Old `sn suspend` and `sn resume` callers must still get a True - return so external automation doesn't break, but the body must do - nothing — in particular it must not touch FirewallClient.""" - - def test_suspend_is_noop(self): - from simplyblock_core import storage_node_ops as sno - with patch.object(sno, "port_block") as pb: - self.assertTrue(sno.suspend_storage_node("any-node-id")) - pb.set_port.assert_not_called() - - def test_resume_is_noop(self): - from simplyblock_core import storage_node_ops as sno - with patch.object(sno, "port_block") as pb: - self.assertTrue(sno.resume_storage_node("any-node-id")) - pb.set_port.assert_not_called() - - # --------------------------------------------------------------------------- # 4. shutdown_storage_node graceful path # --------------------------------------------------------------------------- diff --git a/tests/unit/test_node_removal.py b/tests/unit/test_node_removal.py new file mode 100644 index 000000000..720159331 --- /dev/null +++ b/tests/unit/test_node_removal.py @@ -0,0 +1,405 @@ +# coding=utf-8 +""" +Unit tests for online storage-node removal (inverse of cluster expansion). + +Covers: + * remove_storage_node() precondition gating (online-only, all-peers-online, + no LVols, no snapshots, Case-B relocation feasibility) and task queueing. + * _check_replica_relocation_feasible() / _pick_replica_relocation_node(). + * The orchestration helpers' idempotent bookkeeping: + - _teardown_replicas_of_primary() (Case A) + - _relocate_one_replica() (Case B) + - _decommission_node_devices() (remove/fail/migrate completion gate) + * The in_removal status code mapping. + +All data-plane RPCs / device-controller / DB access is mocked — these are +pure control-flow + bookkeeping tests. +""" + +import unittest +from unittest.mock import MagicMock, patch + +from simplyblock_core import storage_node_ops +from simplyblock_core.models.storage_node import StorageNode +from simplyblock_core.models.nvme_device import NVMeDevice, JMDevice +from simplyblock_core.models.cluster import Cluster + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +def _cluster(ha_type="ha", npcs=1, ndcs=2, ft=1, mode="docker"): + cl = Cluster() + cl.uuid = "cluster-1" + cl.ha_type = ha_type + cl.distr_npcs = npcs + cl.distr_ndcs = ndcs + cl.max_fault_tolerance = ft + cl.mode = mode + cl.status = Cluster.STATUS_ACTIVE + cl.nqn = "nqn.2023-01.io.simplyblock:cluster-1" + return cl + + +def _node(node_id, status=StorageNode.STATUS_ONLINE, lvstore="", + secondary_id="", tertiary_id="", + stack_secondary="", stack_tertiary="", n_devices=0, with_jm=False): + n = MagicMock(spec=StorageNode) + n.uuid = node_id + n.get_id = MagicMock(return_value=node_id) + n.status = status + n.cluster_id = "cluster-1" + n.lvstore = lvstore + n.lvstore_stack = [{"type": "bdev_distr", "name": "distrib_1"}, + {"type": "bdev_raid", "name": "raid_1"}, + {"type": "bdev_lvstore", "name": lvstore or "LVS"}] + n.secondary_node_id = secondary_id + n.tertiary_node_id = tertiary_id + n.lvstore_stack_secondary = stack_secondary + n.lvstore_stack_tertiary = stack_tertiary + n.mgmt_ip = f"10.0.0.{abs(hash(node_id)) % 250 + 1}" + n.write_to_db = MagicMock() + n.rpc_client = MagicMock(return_value=MagicMock()) + n.hublvol_nqn_for_lvstore = MagicMock(return_value=f"nqn:hub:{lvstore}") + n.client = MagicMock(return_value=MagicMock()) + n.rpc_port = 8080 + + devs = [] + for i in range(n_devices): + d = NVMeDevice() + d.uuid = f"dev-{node_id}-{i}" + d.node_id = node_id + d.status = NVMeDevice.STATUS_ONLINE + d.pcie_address = f"0000:00:0{i}.0" + d.cluster_device_order = i + devs.append(d) + n.nvme_devices = devs + + if with_jm: + jm = JMDevice() + jm.uuid = f"jm-{node_id}" + jm.node_id = node_id + jm.status = JMDevice.STATUS_ONLINE + n.jm_device = jm + else: + n.jm_device = None + return n + + +class FakeDB: + def __init__(self, cluster, nodes, lvols=None, snaps=None): + self.cluster = cluster + self.nodes = {n.get_id(): n for n in nodes} + self.lvols = lvols or {} + self.snaps = snaps or [] + self.kv_store = MagicMock() + # devices indexed by id, pulled from the nodes + self.devices = {} + for n in nodes: + for d in n.nvme_devices: + self.devices[d.get_id()] = d + + def get_cluster_by_id(self, _): + return self.cluster + + def get_storage_nodes_by_cluster_id(self, _): + return list(self.nodes.values()) + + def get_storage_node_by_id(self, nid): + if nid in self.nodes: + return self.nodes[nid] + raise KeyError(nid) + + def get_lvols_by_node_id(self, nid): + return self.lvols.get(nid, []) + + def get_snapshots(self): + return self.snaps + + def get_storage_device_by_id(self, did): + return self.devices[did] + + +# --------------------------------------------------------------------------- +# remove_storage_node — preconditions +# --------------------------------------------------------------------------- + +class TestRemovePreconditions(unittest.TestCase): + + def _run(self, db, **patches): + tc = MagicMock() + tc.get_active_node_removal_task.return_value = patches.get("active_removal", False) + tc.get_active_node_tasks.return_value = patches.get("active_tasks", []) + tc.add_node_removal_task.return_value = patches.get("task_id", "task-uuid-1") + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "tasks_controller", tc), \ + patch.object(storage_node_ops, "_check_ftt_allows_node_removal", + return_value=patches.get("ftt", (True, ""))), \ + patch.object(storage_node_ops, "_check_replica_relocation_feasible", + return_value=patches.get("feasible", (True, ""))): + ret = storage_node_ops.remove_storage_node("n1") + return ret, tc + + def test_happy_path_queues_task(self): + cl = _cluster() + nodes = [_node("n1"), _node("n2"), _node("n3")] + ret, tc = self._run(FakeDB(cl, nodes)) + self.assertEqual(ret, "task-uuid-1") + tc.add_node_removal_task.assert_called_once() + + def test_reject_target_not_online(self): + cl = _cluster() + nodes = [_node("n1", status=StorageNode.STATUS_OFFLINE), _node("n2")] + ret, tc = self._run(FakeDB(cl, nodes)) + self.assertFalse(ret) + tc.add_node_removal_task.assert_not_called() + + def test_reject_peer_not_online(self): + cl = _cluster() + nodes = [_node("n1"), _node("n2", status=StorageNode.STATUS_DOWN), _node("n3")] + ret, tc = self._run(FakeDB(cl, nodes)) + self.assertFalse(ret) + tc.add_node_removal_task.assert_not_called() + + def test_removed_peer_is_ignored(self): + cl = _cluster() + nodes = [_node("n1"), _node("n2"), + _node("n3", status=StorageNode.STATUS_REMOVED)] + ret, _ = self._run(FakeDB(cl, nodes)) + self.assertEqual(ret, "task-uuid-1") + + def test_reject_lvols_present(self): + cl = _cluster() + nodes = [_node("n1"), _node("n2")] + db = FakeDB(cl, nodes, lvols={"n1": [MagicMock()]}) + ret, tc = self._run(db) + self.assertFalse(ret) + tc.add_node_removal_task.assert_not_called() + + def test_reject_snapshots_present(self): + cl = _cluster() + nodes = [_node("n1"), _node("n2")] + snap = MagicMock() + snap.lvol.node_id = "n1" + snap.deleted = False + db = FakeDB(cl, nodes, snaps=[snap]) + ret, tc = self._run(db) + self.assertFalse(ret) + + def test_reject_relocation_infeasible(self): + cl = _cluster() + nodes = [_node("n1"), _node("n2")] + ret, tc = self._run(FakeDB(cl, nodes), + feasible=(False, "no host-disjoint node")) + self.assertFalse(ret) + tc.add_node_removal_task.assert_not_called() + + def test_reject_ftt(self): + cl = _cluster() + nodes = [_node("n1"), _node("n2")] + ret, tc = self._run(FakeDB(cl, nodes), ftt=(False, "ftt blocks")) + self.assertFalse(ret) + + def test_idempotent_returns_existing_task(self): + cl = _cluster() + nodes = [_node("n1"), _node("n2")] + ret, tc = self._run(FakeDB(cl, nodes), active_removal="existing-task") + self.assertEqual(ret, "existing-task") + tc.add_node_removal_task.assert_not_called() + + +# --------------------------------------------------------------------------- +# Replica relocation feasibility / picking +# --------------------------------------------------------------------------- + +class TestRelocationFeasibility(unittest.TestCase): + + def test_no_hosted_replicas_is_feasible(self): + cl = _cluster() + removed = _node("n1") + db = FakeDB(cl, [removed, _node("n2")]) + ok, _ = storage_node_ops._check_replica_relocation_feasible(removed, db) + self.assertTrue(ok) + + def test_infeasible_when_no_target(self): + cl = _cluster() + removed = _node("n1", stack_secondary="p1") + primary = _node("p1", secondary_id="n1") + db = FakeDB(cl, [removed, primary]) + with patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value=None): + ok, reason = storage_node_ops._check_replica_relocation_feasible(removed, db) + self.assertFalse(ok) + self.assertIn("secondary", reason) + + def test_feasible_when_target_exists(self): + cl = _cluster() + removed = _node("n1", stack_secondary="p1") + primary = _node("p1", secondary_id="n1") + db = FakeDB(cl, [removed, primary]) + with patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="n3"): + ok, _ = storage_node_ops._check_replica_relocation_feasible(removed, db) + self.assertTrue(ok) + + def test_pick_secondary_uses_get_secondary_nodes(self): + cl = _cluster() + primary = _node("p1", secondary_id="n1", tertiary_id="n9") + removed = _node("n1") + db = FakeDB(cl, [primary, removed]) + with patch.object(storage_node_ops, "get_secondary_nodes", + return_value=["n5"]) as gsn: + got = storage_node_ops._pick_replica_relocation_node( + primary, removed, "secondary", db) + self.assertEqual(got, "n5") + # the removed node and the tertiary are excluded from candidates + _, kwargs = gsn.call_args + self.assertIn("n1", kwargs["exclude_ids"]) + self.assertIn("n9", kwargs["exclude_ids"]) + + +# --------------------------------------------------------------------------- +# Case A — teardown of own primary's replicas +# --------------------------------------------------------------------------- + +class TestTeardownOwnReplicas(unittest.TestCase): + + def test_clears_bookkeeping_both_sides(self): + cl = _cluster() + removed = _node("n1", lvstore="LVS_1", + secondary_id="n2", tertiary_id="n3") + sec = _node("n2", stack_secondary="n1") + tert = _node("n3", stack_tertiary="n1") + db = FakeDB(cl, [removed, sec, tert]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_delete_replica_on_peer") as drp: + ret = storage_node_ops._teardown_replicas_of_primary(removed) + self.assertTrue(ret) + self.assertEqual(removed.secondary_node_id, "") + self.assertEqual(removed.tertiary_node_id, "") + self.assertEqual(sec.lvstore_stack_secondary, "") + self.assertEqual(tert.lvstore_stack_tertiary, "") + self.assertEqual(drp.call_count, 2) + + +# --------------------------------------------------------------------------- +# Case B — relocate a hosted replica +# --------------------------------------------------------------------------- + +class TestRelocateOneReplica(unittest.TestCase): + + def test_relocate_success_moves_bookkeeping(self): + cl = _cluster() + removed = _node("n1", stack_secondary="p1") + primary = _node("p1", secondary_id="n1", lvstore="LVS_p1") + new = _node("n3") + db = FakeDB(cl, [removed, primary, new]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="n3"), \ + patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", + return_value=True) as rec: + ret = storage_node_ops._relocate_one_replica(removed, "p1", "secondary") + self.assertTrue(ret) + self.assertEqual(primary.secondary_node_id, "n3") + self.assertEqual(new.lvstore_stack_secondary, "p1") + # back-reference cleared only after a successful rebuild + self.assertEqual(removed.lvstore_stack_secondary, "") + rec.assert_called_once() + + def test_relocate_failure_keeps_backref(self): + cl = _cluster() + removed = _node("n1", stack_secondary="p1") + primary = _node("p1", secondary_id="n1", lvstore="LVS_p1") + new = _node("n3") + db = FakeDB(cl, [removed, primary, new]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="n3"), \ + patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", + return_value=False): + ret = storage_node_ops._relocate_one_replica(removed, "p1", "secondary") + self.assertFalse(ret) + # forward bookkeeping committed, but back-ref NOT cleared -> retry resumes + self.assertEqual(primary.secondary_node_id, "n3") + self.assertEqual(removed.lvstore_stack_secondary, "p1") + + def test_relocate_resume_reuses_committed_target(self): + # Simulates a retry: primary already points at the new node, removed + # still holds the back-ref. Must NOT pick a fresh node. + cl = _cluster() + removed = _node("n1", stack_secondary="p1") + primary = _node("p1", secondary_id="n3", lvstore="LVS_p1") + new = _node("n3", stack_secondary="p1") + db = FakeDB(cl, [removed, primary, new]) + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "_pick_replica_relocation_node", + return_value="SHOULD-NOT-BE-USED") as pick, \ + patch.object(storage_node_ops, "recreate_lvstore_on_non_leader", + return_value=True): + ret = storage_node_ops._relocate_one_replica(removed, "p1", "secondary") + self.assertTrue(ret) + pick.assert_not_called() + self.assertEqual(removed.lvstore_stack_secondary, "") + + def test_relocate_missing_primary_just_clears(self): + cl = _cluster() + removed = _node("n1", stack_secondary="gone") + db = FakeDB(cl, [removed]) + with patch.object(storage_node_ops, "DBController", return_value=db): + ret = storage_node_ops._relocate_one_replica(removed, "gone", "secondary") + self.assertTrue(ret) + self.assertEqual(removed.lvstore_stack_secondary, "") + + +# --------------------------------------------------------------------------- +# Device decommission completion gate +# --------------------------------------------------------------------------- + +class TestDecommissionDevices(unittest.TestCase): + + def test_first_pass_drives_and_waits(self): + cl = _cluster() + removed = _node("n1", n_devices=2, with_jm=True) + db = FakeDB(cl, [removed]) + + dc = MagicMock() + + def _remove(dev_id, force=True): + db.get_storage_device_by_id(dev_id).status = NVMeDevice.STATUS_REMOVED + return True + + def _fail(dev_id): + db.get_storage_device_by_id(dev_id).status = NVMeDevice.STATUS_FAILED + return True + + dc.device_remove.side_effect = _remove + dc.device_set_failed.side_effect = _fail + + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "device_controller", dc): + ret = storage_node_ops._decommission_node_devices(removed) + + # devices are FAILED (migration queued) but not yet FAILED_AND_MIGRATED + self.assertFalse(ret) + dc.remove_jm_device.assert_called_once() + self.assertEqual(dc.device_remove.call_count, 2) + self.assertEqual(dc.device_set_failed.call_count, 2) + + def test_complete_when_all_migrated(self): + cl = _cluster() + removed = _node("n1", n_devices=2, with_jm=False) + for d in removed.nvme_devices: + d.status = NVMeDevice.STATUS_FAILED_AND_MIGRATED + db = FakeDB(cl, [removed]) + dc = MagicMock() + with patch.object(storage_node_ops, "DBController", return_value=db), \ + patch.object(storage_node_ops, "device_controller", dc): + ret = storage_node_ops._decommission_node_devices(removed) + self.assertTrue(ret) + dc.device_remove.assert_not_called() + + +if __name__ == "__main__": + unittest.main()