Lvol migration from secondary during node removal

simplyblock_core/controllers/migration_controller.py

@@ -103,10 +103,20 @@ def start_migration(migration_id,
     if lvol.status != LVol.STATUS_ONLINE:
         raise ValueError(f"Volume is not online (status={lvol.status})")
 
-    source_node_id = lvol.node_id
+    # source_node_id / active_source_node_id are read from the migration record
+    # (set once by create_migration()), never re-derived from lvol.node_id here —
+    # re-deriving could pick a different fallback than create_migration did if
+    # node health changed in between.
+    source_node_id = migration.source_node_id
 
     try:
-        source_node = db.get_storage_node_by_id(source_node_id)
+        db.get_storage_node_by_id(source_node_id)
+    except KeyError as e:
+        raise ValueError(str(e))
+
+    active_source_node_id = migration.active_source_node_id or source_node_id
+    try:
+        active_source_node = db.get_storage_node_by_id(active_source_node_id)
     except KeyError as e:
         raise ValueError(str(e))
 
@@ -118,19 +128,33 @@ def start_migration(migration_id,
     if source_node_id == target_node_id:
         raise ValueError("Source and target nodes must be different")
 
-    if source_node.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED):
-        raise ValueError(f"Source node is not online (status={source_node.status})")
+    if active_source_node.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED):
+        raise ValueError(f"Source node is not online (status={active_source_node.status})")
 
     if target_node.status != StorageNode.STATUS_ONLINE:
         raise ValueError(f"Target node is not online (status={target_node.status})")
 
+    is_fallback_source = active_source_node_id != source_node_id
+    if is_fallback_source:
+        logger.info(
+            f"start_migration {migration.uuid}: source primary {source_node_id} is offline; "
+            f"continuing with pre-selected fallback source {active_source_node_id}")
+
     cluster = db.get_cluster_by_id(migration.cluster_id)
-    if cluster.status != Cluster.STATUS_ACTIVE:
+    # A fallback migration exists precisely because its primary source node is
+    # down, which is what drives the cluster to DEGRADED in the first place
+    # (storage_node_monitor's one-node-down verdict) — requiring strict ACTIVE
+    # here would make the feature unusable in the scenario it exists for.
+    # Ordinary (non-fallback) migrations keep the stricter ACTIVE-only gate.
+    allowed_statuses = (
+        (Cluster.STATUS_ACTIVE, Cluster.STATUS_DEGRADED) if is_fallback_source
+        else (Cluster.STATUS_ACTIVE,))
+    if cluster.status not in allowed_statuses:
         raise PreconditionError(f"Cluster {cluster.get_id()} is not active (status={cluster.status})")
     if not _can_add_lvol_migration(cluster.get_id()):
         raise PreconditionError(f"Cluster {cluster.get_id()} is rebalancing; wait for it to finish before migrating")
 
-    for node_id in (source_node_id, target_node_id):
+    for node_id in {source_node_id, active_source_node_id, target_node_id}:
         if tasks_controller.get_active_node_mig_task(migration.cluster_id, node_id):
             raise PreconditionError(f"Node {node_id} has a data migration in progress; wait for it to finish")
 
@@ -474,6 +498,50 @@ def _add(uid):
     return result
 
 
+def _resolve_active_source_node(primary_node, target_node_id):
+    """
+    Decide which node the migration will actually issue source-side RPCs
+    against: *primary_node* itself when reachable, otherwise its online
+    secondary, otherwise its online tertiary.
+
+    This is called exactly once, at create time (create_migration /
+    create_batch_migration). The result is persisted as
+    migration.active_source_node_id / group.active_source_node_id and must
+    never be re-derived afterward — start_migration/start_batch_migration and
+    the task runners only ever read it.
+
+    Raises ValueError if the primary is unreachable and no replica is
+    online either. Raises PreconditionError if the resolved node is the
+    same as target_node_id (can't migrate a replica onto itself).
+    """
+    if primary_node.status in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED):
+        active_node = primary_node
+    else:
+        active_node = None
+        for replica_id in (primary_node.secondary_node_id, primary_node.tertiary_node_id):
+            if not replica_id:
+                continue
+            try:
+                replica = db.get_storage_node_by_id(replica_id)
+            except KeyError:
+                continue
+            if replica.status == StorageNode.STATUS_ONLINE:
+                active_node = replica
+                break
+        if active_node is None:
+            raise ValueError(
+                f"Source node is not online (status={primary_node.status}) "
+                f"and no online secondary/tertiary replica is available")
+
+    if active_node.get_id() == target_node_id:
+        raise PreconditionError(
+            f"Cannot migrate to node {target_node_id}: source primary "
+            f"{primary_node.get_id()} is offline and {target_node_id} is "
+            f"currently serving as the fallback source for this volume")
+
+    return active_node
+
+
 def _is_snap_on_node(snap_id, node_id):
     """Return True if *snap_id* already has a copy on *node_id*.
 
@@ -1003,8 +1071,22 @@ def create_migration(lvol_id, target_node_id,
     except KeyError:
         raise ValueError(f"Source node {src_node_id} not found")
 
+    active_src_node = _resolve_active_source_node(src_node, target_node_id)
+    is_fallback_source = active_src_node.get_id() != src_node_id
+    if is_fallback_source:
+        logger.warning(
+            f"create_migration: source primary {src_node_id} is offline; "
+            f"using {active_src_node.get_id()} as the effective source for lvol={lvol_id}")
+
     cluster = db.get_cluster_by_id(tgt_node.cluster_id)
-    if cluster.status != Cluster.STATUS_ACTIVE:
+    # See the matching comment in start_migration(): a fallback migration's
+    # primary is down, which is what drives the cluster to DEGRADED, so the
+    # strict ACTIVE-only gate would make the feature unusable for the
+    # scenario it exists for. Non-fallback migrations keep the stricter gate.
+    allowed_statuses = (
+        (Cluster.STATUS_ACTIVE, Cluster.STATUS_DEGRADED) if is_fallback_source
+        else (Cluster.STATUS_ACTIVE,))
+    if cluster.status not in allowed_statuses:
         raise PreconditionError(f"Cluster {cluster.get_id()} is not active (status={cluster.status})")
     if not _can_add_lvol_migration(cluster.get_id()):
         raise PreconditionError(f"Cluster {cluster.get_id()} is rebalancing; wait for it to finish before migrating")
@@ -1315,6 +1397,7 @@ def create_migration(lvol_id, target_node_id,
     migration.cluster_id = tgt_node.cluster_id
     migration.lvol_id = lvol_id
     migration.source_node_id = lvol.node_id
+    migration.active_source_node_id = active_src_node.get_id()
     migration.target_node_id = target_node_id
     migration.phase = LVolMigration.PHASE_PRE_CREATED
     migration.status = LVolMigration.STATUS_NEW
@@ -1401,6 +1484,10 @@ def create_batch_migration(lvol_id, target_node_id,
 
     # Pre-create individual migration records for each member.
     # connect_strings come from the master (ns_id=1) since the NQN is shared.
+    # Each create_migration() call independently resolves the same active
+    # source node (all members share the same primary/lvstore), so the
+    # group's own active_source_node_id below is read from the first member's
+    # already-resolved record rather than re-resolved here.
     member_records = []   # list of (ns_id, migration_id)
     master_connect_strings = []
     for member in members:
@@ -1414,6 +1501,14 @@ def create_batch_migration(lvol_id, target_node_id,
         if member.ns_id == 1:
             master_connect_strings = connect_strings
 
+    active_source_node_id = source_node_id
+    if member_records:
+        try:
+            active_source_node_id = db.get_migration_by_id(
+                member_records[0]["migration_id"]).active_source_node_id or source_node_id
+        except KeyError:
+            pass
+
     # Compute snap ownership: snap_uuid → lvol_uuid, then remap to migration_id.
     lvol_uuid_to_migration_id = {
         member.uuid: rec["migration_id"]
@@ -1438,6 +1533,11 @@ def create_batch_migration(lvol_id, target_node_id,
     group.uuid = str(uuid.uuid4())
     group.cluster_id = tgt_node.cluster_id
     group.source_node_id = source_node_id
+    group.active_source_node_id = active_source_node_id
+    if active_source_node_id != source_node_id:
+        logger.warning(
+            f"create_batch_migration: source primary {source_node_id} is offline; "
+            f"using {active_source_node_id} as the effective source for group NQN={lvol.nqn}")
     group.target_node_id = target_node_id
     group.target_nqn = lvol.nqn
     group.members = member_records
@@ -1482,16 +1582,39 @@ def start_batch_migration(group_id,
             f"Group {group_id} is not in PHASE_PRE_CREATED (phase={group.phase})"
         )
 
+    # active_source_node_id is read-only here — it was resolved once, at
+    # create_batch_migration() time, and must never be re-derived.
+    active_source_node_id = group.active_source_node_id or group.source_node_id
+    is_fallback_source = active_source_node_id != group.source_node_id
+
     # Same preconditions as start_migration's single-lvol path — these are
     # only checked at create_batch_migration (precreate) time today, so a
     # cluster rebalance / conflicting node migration starting in the gap
     # before migrate-continue --batch would otherwise go unnoticed here.
+    # A fallback group's primary is down, which is what drives the cluster to
+    # DEGRADED in the first place, so the strict ACTIVE-only gate would make
+    # the feature unusable for the scenario it exists for (see the matching
+    # comment in start_migration()). Non-fallback groups keep the stricter gate.
     cluster = db.get_cluster_by_id(group.cluster_id)
-    if cluster.status != Cluster.STATUS_ACTIVE:
+    allowed_statuses = (
+        (Cluster.STATUS_ACTIVE, Cluster.STATUS_DEGRADED) if is_fallback_source
+        else (Cluster.STATUS_ACTIVE,))
+    if cluster.status not in allowed_statuses:
         raise PreconditionError(f"Cluster {cluster.get_id()} is not active (status={cluster.status})")
     if not _can_add_lvol_migration(cluster.get_id()):
         raise PreconditionError(f"Cluster {cluster.get_id()} is rebalancing; wait for it to finish before migrating")
-    for node_id in (group.source_node_id, group.target_node_id):
+    try:
+        active_source_node = db.get_storage_node_by_id(active_source_node_id)
+    except KeyError as e:
+        raise ValueError(str(e))
+    if active_source_node.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED):
+        raise ValueError(f"Source node is not online (status={active_source_node.status})")
+    if is_fallback_source:
+        logger.info(
+            f"start_batch_migration {group_id}: source primary {group.source_node_id} is offline; "
+            f"continuing with pre-selected fallback source {active_source_node_id}")
+
+    for node_id in {group.source_node_id, active_source_node_id, group.target_node_id}:
         if tasks_controller.get_active_node_mig_task(group.cluster_id, node_id):
             raise PreconditionError(f"Node {node_id} has a data migration in progress; wait for it to finish")
 
@@ -1524,7 +1647,6 @@ def start_batch_migration(group_id,
                                  if s not in snaps_on_target
                                  and group.snap_owners.get(s) != migration_id]
 
-        migration.source_node_id = lvol.node_id
         migration.phase = LVolMigration.PHASE_SNAP_COPY
         migration.snap_migration_plan = owned_snaps
         migration.snaps_migrated = []

simplyblock_core/models/lvol_migration.py

@@ -56,6 +56,13 @@ class LVolMigration(BaseModel):
     source_node_id: str = ""
     target_node_id: str = ""
 
+    # Node to actually issue source-side RPCs against. Equals source_node_id
+    # (the primary) unless the primary was offline at create_migration() time,
+    # in which case this is the online secondary/tertiary replica that was
+    # chosen as the effective source. Resolved once at create time and never
+    # re-derived afterward — the runner only ever reads it.
+    active_source_node_id: str = ""
+
     # --- Phase tracking ---
     phase: str = ""
 

simplyblock_core/models/lvol_migration_group.py

@@ -68,6 +68,13 @@ class LVolMigrationGroup(BaseModel):
     source_node_id: str = ""
     target_node_id: str = ""
 
+    # Node to actually issue source-side RPCs against. Equals source_node_id
+    # (the primary) unless the primary was offline at create_batch_migration()
+    # time, in which case this is the online secondary/tertiary replica chosen
+    # as the effective source. Resolved once at create time and never
+    # re-derived afterward — the runner only ever reads it.
+    active_source_node_id: str = ""
+
     # Shared NVMe-oF NQN created on the target during PRECREATE.
     target_nqn: str = ""