diff --git a/simplyblock_core/distr_controller.py b/simplyblock_core/distr_controller.py index 6cca30acf..cddd8fd24 100644 --- a/simplyblock_core/distr_controller.py +++ b/simplyblock_core/distr_controller.py @@ -63,7 +63,7 @@ def _mutate(node): def send_node_status_event(node, node_status, target_node=None): db_controller = DBController() node_id = node.get_id() - if node_status == StorageNode.STATUS_SCHEDULABLE: + if node_status in [StorageNode.STATUS_SCHEDULABLE, StorageNode.STATUS_IN_REMOVAL, StorageNode.STATUS_PENDING_REMOVAL]: node_status = StorageNode.STATUS_UNREACHABLE logger.info(f"Sending event updates, node: {node_id}, status: {node_status}") node_status_event = { @@ -251,7 +251,7 @@ def get_distr_cluster_map(snodes, target_node, distr_name=""): node_w += dev_w_gib node_status = snode.status - if node_status == StorageNode.STATUS_SCHEDULABLE: + if node_status in [StorageNode.STATUS_SCHEDULABLE, StorageNode.STATUS_IN_REMOVAL, StorageNode.STATUS_PENDING_REMOVAL]: node_status = StorageNode.STATUS_UNREACHABLE map_cluster[snode.get_id()] = { "status": node_status, @@ -348,6 +348,8 @@ def parse_distr_cluster_map(map_string, nodes=None, devices=None): StorageNode.STATUS_SCHEDULABLE, StorageNode.STATUS_RESTARTING, StorageNode.STATUS_IN_SHUTDOWN, + StorageNode.STATUS_IN_REMOVAL, + StorageNode.STATUS_PENDING_REMOVAL ): node_status = StorageNode.STATUS_UNREACHABLE data["Desired Status"] = node_status diff --git a/simplyblock_core/models/base_model.py b/simplyblock_core/models/base_model.py index c0a2cadc4..505fc9b69 100644 --- a/simplyblock_core/models/base_model.py +++ b/simplyblock_core/models/base_model.py @@ -374,6 +374,8 @@ class BaseNodeObject(BaseModel): STATUS_UNREACHABLE = 'unreachable' STATUS_SCHEDULABLE = 'schedulable' STATUS_DOWN = 'down' + STATUS_IN_REMOVAL = 'in_removal' + STATUS_PENDING_REMOVAL = 'pending_removal' _STATUS_CODE_MAP = { STATUS_ONLINE: 0, @@ -386,4 +388,6 @@ class BaseNodeObject(BaseModel): STATUS_UNREACHABLE: 20, STATUS_SCHEDULABLE: 30, STATUS_DOWN: 40, + STATUS_IN_REMOVAL: 41, + STATUS_PENDING_REMOVAL: 42, } diff --git a/simplyblock_core/services/tasks_runner_node_removal.py b/simplyblock_core/services/tasks_runner_node_removal.py index a90fbc36e..b26e1eafe 100644 --- a/simplyblock_core/services/tasks_runner_node_removal.py +++ b/simplyblock_core/services/tasks_runner_node_removal.py @@ -6,7 +6,7 @@ from simplyblock_core.controllers import tasks_controller from simplyblock_core.models.job_schedule import JobSchedule from simplyblock_core.models.cluster import Cluster - +from simplyblock_core.models.storage_node import StorageNode logger = utils.get_logger(__name__) @@ -34,6 +34,7 @@ def process_task(task): task.function_result = "cluster is in_activation, waiting" task.status = JobSchedule.STATUS_SUSPENDED task.write_to_db(db.kv_store) + storage_node_ops.set_node_status(task.node_id, StorageNode.STATUS_PENDING_REMOVAL, caused_by="remove") return False if task.status != JobSchedule.STATUS_RUNNING: @@ -57,13 +58,13 @@ def process_task(task): 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 + else: + # 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...") diff --git a/simplyblock_core/storage_node_ops.py b/simplyblock_core/storage_node_ops.py index 6c9657b52..c0557386e 100644 --- a/simplyblock_core/storage_node_ops.py +++ b/simplyblock_core/storage_node_ops.py @@ -3451,24 +3451,11 @@ def remove_storage_node(node_id, force_remove=False, force_migrate=False): logger.warning(f"Node already removed: {node_id}") return False - if snode.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED]: + if snode.status not in [StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED, + StorageNode.STATUS_PENDING_REMOVAL, StorageNode.STATUS_IN_REMOVAL, + StorageNode.STATUS_OFFLINE, StorageNode.STATUS_UNREACHABLE]: 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}") + f"Can not remove node {node_id}: (current status: {snode.status}).") return False allowed, reason = _check_ftt_allows_node_removal(node_id, db_controller) @@ -3520,6 +3507,23 @@ def remove_storage_node(node_id, force_remove=False, force_migrate=False): logger.error(f"Can not remove node {node_id}: {reason}") return False + if snode.status not in [StorageNode.STATUS_PENDING_REMOVAL, StorageNode.STATUS_IN_REMOVAL, + StorageNode.STATUS_OFFLINE, StorageNode.STATUS_REMOVED]: + 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) + + if snode.status != StorageNode.STATUS_PENDING_REMOVAL: + set_node_status(node_id, StorageNode.STATUS_PENDING_REMOVAL, caused_by="remove") + task_id = tasks_controller.add_node_removal_task( snode.cluster_id, node_id, {"force_remove": force_remove}) if not task_id: @@ -3655,6 +3659,9 @@ def node_removal_orchestrate(node_id, force_remove=False): return False snode = db_controller.get_storage_node_by_id(node_id) + if snode.status != StorageNode.STATUS_IN_REMOVAL: + set_node_status(node_id, StorageNode.STATUS_IN_REMOVAL, caused_by="remove") + # 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") @@ -3670,9 +3677,6 @@ def node_removal_orchestrate(node_id, force_remove=False): 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") @@ -3680,9 +3684,9 @@ def node_removal_orchestrate(node_id, force_remove=False): return False logger.info(f"[REMOVAL] {node_id}: done") - return True finally: cluster_ops.set_cluster_status(cluster.get_id(), prev_cluster_status) + return True def _teardown_replicas_of_primary(removed_node: StorageNode): diff --git a/tests/unit/test_node_removal.py b/tests/unit/test_node_removal.py index 3a0b78840..60f5923e9 100644 --- a/tests/unit/test_node_removal.py +++ b/tests/unit/test_node_removal.py @@ -132,8 +132,12 @@ 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.get_active_node_restart_task.return_value = [] + tc.get_active_lvol_migration.return_value = [] 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, "shutdown_storage_node", return_value=True), \ + patch.object(storage_node_ops, "set_node_status", return_value=True), \ patch.object(storage_node_ops, "tasks_controller", tc), \ patch.object(storage_node_ops, "_check_ftt_allows_node_removal", return_value=patches.get("ftt", (True, ""))), \ @@ -149,19 +153,6 @@ def test_happy_path_queues_task(self): 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()