Fix node-removal state handling and pause/resume orchestration

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

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,
     }

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

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,19 +3677,16 @@ 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")
         if not _decommission_node_devices(snode):
             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):

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()