diff --git a/simplyblock_core/controllers/migration_controller.py b/simplyblock_core/controllers/migration_controller.py index 49a173f874..fd7f167797 100644 --- a/simplyblock_core/controllers/migration_controller.py +++ b/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 = [] diff --git a/simplyblock_core/models/lvol_migration.py b/simplyblock_core/models/lvol_migration.py index bff35d6845..5dbb63c99b 100644 --- a/simplyblock_core/models/lvol_migration.py +++ b/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 = "" diff --git a/simplyblock_core/models/lvol_migration_group.py b/simplyblock_core/models/lvol_migration_group.py index a21c9a3a34..4024f45359 100644 --- a/simplyblock_core/models/lvol_migration_group.py +++ b/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 = "" diff --git a/simplyblock_core/services/tasks_runner_batch_migration.py b/simplyblock_core/services/tasks_runner_batch_migration.py index 0b046fd6e9..5ced144d4b 100644 --- a/simplyblock_core/services/tasks_runner_batch_migration.py +++ b/simplyblock_core/services/tasks_runner_batch_migration.py @@ -269,12 +269,17 @@ def _handle_snap_copy_barrier(group, member_migrations, tgt_node, tgt_rpc): return True, None -def _build_batch_final_args(group, member_migrations, src_node, tgt_node, tgt_rpc): +def _build_batch_final_args(group, member_migrations, src_node, tgt_node, tgt_rpc, + primary_src_node=None): """ Build the argument lists for bdev_lvol_batch_final_step, ordered by ns_id. Returns (lvol_names, lvol_ids, snapshot_names) or raises ValueError. """ + # The lvstore NAME is always the true primary's own lvstore, regardless + # of which node is actually driving the transfer as src_node/src_rpc — + # see tasks_runner_lvol_migration._build_paths' matching comment. + src_lvstore = (primary_src_node or src_node).lvstore mid_to_migration = {m.uuid: m for m in member_migrations} ordered_ids = group.ordered_migration_ids() @@ -295,7 +300,7 @@ def _build_batch_final_args(group, member_migrations, src_node, tgt_node, tgt_rp raise ValueError(f"migration {migration_id} not found in member_migrations") lvol = db.get_lvol_by_id(m.lvol_id) - src_composite = f"{src_node.lvstore}/{lvol.lvol_bdev}" + src_composite = f"{src_lvstore}/{lvol.lvol_bdev}" lvol_names.append(src_composite) tgt_bdev_short = _lvol_tgt_bdev_name(lvol.lvol_bdev) @@ -464,7 +469,8 @@ def _commit_intermediate_snapshot_chain(group, member_migrations, tgt_node, tgt_ return None -def _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node, tgt_rpc): +def _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=None): """ After a successful bdev_lvol_batch_final_step, drive clients to the new target. @@ -487,7 +493,8 @@ def _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node 4. Remove old SRC-port listener from overlap TGT nodes if port changed """ nqn = group.target_nqn - src_paths, tgt_paths, overlap_ids = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) + src_paths, tgt_paths, overlap_ids = _build_paths( + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) src_port_by_id = {p['node_id']: p['port'] for p in src_paths} # Detect and repair a target-side node restart that wiped the migration's @@ -674,7 +681,8 @@ def _flip_all_required(rpc, ips, port, trtype, state, label, attempts=3): f"{tgt['node_id'][:8]} (non-fatal): {e}") -def _handle_intermediate_barrier(group, member_migrations, src_node, tgt_node, src_rpc, tgt_rpc): +def _handle_intermediate_barrier(group, member_migrations, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=None): """ Wait for all workers to reach intermediates_done, then call bdev_lvol_batch_final_step. Returns (batch_ok, error). @@ -714,7 +722,8 @@ def _handle_intermediate_barrier(group, member_migrations, src_node, tgt_node, s try: lvol_names, lvol_ids, snapshot_names = _build_batch_final_args( - group, member_migrations, src_node, tgt_node, tgt_rpc) + group, member_migrations, src_node, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) except (ValueError, KeyError) as e: # Hub controller left attached — hub_manager owns its lifecycle # entirely via its own idle timeout. @@ -729,8 +738,11 @@ def _handle_intermediate_barrier(group, member_migrations, src_node, tgt_node, s # synchronous final-step transfer below (see the diagnostic block further # down for the current, temporarily-widened version of this). nqn = group.target_nqn - src_paths, tgt_paths, _ = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) - src_replica_paths = src_paths[1:] # secondary/tertiary only; used for the failure-path revert below + src_paths, tgt_paths, _ = _build_paths( + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) + # secondary/tertiary only -- the active source is frozen internally by the + # RPC below; used for the failure-path revert further down. + src_replica_paths = src_paths[1:] def _flip(rpc, ip, port, trtype, state, label): try: @@ -910,7 +922,8 @@ def _revert_src_replicas(reason): logger.warning( f"Group {group.uuid[:8]}: add_clone for member {m.uuid[:8]} (non-fatal): {e}") - _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node, tgt_rpc) + _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) # Hub controller left attached on both success and failure — hub_manager # owns its lifecycle entirely via its own idle timeout. Detaching it here @@ -954,14 +967,23 @@ def _handle_cleanup_source_barrier(group): return expected.issubset(done_set) -def _delete_source_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc): +def _delete_source_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc, primary_src_node=None): """ - Delete the source NVMe-oF subsystem on all SRC replicas (primary, secondary, - tertiary). Overlap nodes (which also host TGT replicas) are skipped because - the subsystem is still in use on those nodes. Best-effort. + Delete the source NVMe-oF subsystem on all SRC replicas (active source, + plus whichever of the primary's secondary/tertiary are still online and + aren't the active source itself). Overlap nodes (which also host TGT + replicas) are skipped because the subsystem is still in use on those + nodes. Best-effort. + + *src_node* is the active source (primary, or its fallback replica when + the primary was offline at create time). *primary_src_node* — the true + primary — is only consulted for its own secondary_node_id/tertiary_node_id + fields, since a replica's own such fields describe an unrelated pairing. """ nqn = group.target_nqn - _, _, overlap_ids = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) + primary_src_node = primary_src_node or src_node + _, _, overlap_ids = _build_paths( + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) def _try_delete(rpc, node_id, label): if node_id in overlap_ids: @@ -973,24 +995,25 @@ def _try_delete(rpc, node_id, label): except Exception as e: logger.warning(f"Group {group.uuid[:8]}: {label} source subsystem delete (non-fatal): {e}") - _try_delete(src_rpc, src_node.get_id(), "primary") + _active_label = "active-source" if src_node.get_id() != primary_src_node.get_id() else "primary" + _try_delete(src_rpc, src_node.get_id(), _active_label) - if src_node.secondary_node_id: + if primary_src_node.secondary_node_id and primary_src_node.secondary_node_id != src_node.get_id(): try: - sec_node = db.get_storage_node_by_id(src_node.secondary_node_id) + sec_node = db.get_storage_node_by_id(primary_src_node.secondary_node_id) sec_rpc = _make_rpc(sec_node) _try_delete(sec_rpc, sec_node.get_id(), "secondary") except Exception as e: logger.warning( f"Group {group.uuid[:8]}: secondary src node lookup (non-fatal): {e}") - tert_node = _get_source_tertiary_node(src_node) - if tert_node: + tert_node = _get_source_tertiary_node(primary_src_node) + if tert_node and tert_node.get_id() != src_node.get_id(): tert_rpc = _make_rpc(tert_node) _try_delete(tert_rpc, tert_node.get_id(), "tertiary") -def _delete_target_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc): +def _delete_target_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc, primary_src_node=None): """Delete the target NVMe-oF subsystem on all TGT replicas. Best-effort. Nodes that appear in both SRC and TGT replica sets (overlap nodes) share the @@ -1000,7 +1023,8 @@ def _delete_target_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc): nqn = group.target_nqn try: - _, _, overlap_ids = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) + _, _, overlap_ids = _build_paths( + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) except Exception as e: logger.warning( f"Group {group.uuid[:8]}: _build_paths in _delete_target_subsystem (non-fatal): {e}") @@ -1103,12 +1127,25 @@ def task_runner(task): task.write_to_db(db.kv_store) return True + # primary_src_node is the true primary (kept only for HA topology lookups + # via its own secondary_node_id/tertiary_node_id). src_node is the node + # this group actually issues source-side RPCs against — the primary when + # reachable, otherwise the replica pinned once at create_batch_migration() + # time as group.active_source_node_id. All data-plane calls below must use + # src_node/src_rpc, never primary_src_node. try: - src_node = db.get_storage_node_by_id(group.source_node_id) + primary_src_node = db.get_storage_node_by_id(group.source_node_id) except KeyError: return _batch_budget_suspend( task, group, group_id, f"source node {group.source_node_id} not found") + try: + src_node = db.get_storage_node_by_id( + group.active_source_node_id or group.source_node_id) + except KeyError: + return _batch_budget_suspend( + task, group, group_id, "active source node not found") + try: tgt_node = db.get_storage_node_by_id(group.target_node_id) except KeyError: @@ -1171,7 +1208,8 @@ def task_runner(task): task.write_to_db(db.kv_store) return False - fresh_src = db.get_storage_node_by_id(group.source_node_id) + fresh_src = db.get_storage_node_by_id( + group.active_source_node_id or group.source_node_id) if fresh_src.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): logger.warning( f"Group {group_id[:8]}: source node unavailable " @@ -1213,7 +1251,8 @@ def task_runner(task): # ── PHASE_INTERMEDIATE: wait for intermediates, then batch_final_step ──── if phase == LVolMigrationGroup.PHASE_INTERMEDIATE: batch_ok, err = _handle_intermediate_barrier( - group, member_migrations, src_node, tgt_node, src_rpc, tgt_rpc) + group, member_migrations, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=primary_src_node) if err: logger.error(f"Group {group_id[:8]}: intermediate barrier error: {err}") @@ -1262,7 +1301,8 @@ def task_runner(task): task.write_to_db(db.kv_store) return False - _delete_source_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc) + _delete_source_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) group.phase = LVolMigrationGroup.PHASE_COMPLETED group.status = LVolMigrationGroup.STATUS_DONE @@ -1282,7 +1322,8 @@ def task_runner(task): task.write_to_db(db.kv_store) return False - _delete_target_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc) + _delete_target_subsystem(group, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) group.status = LVolMigrationGroup.STATUS_FAILED group.write_to_db(db.kv_store) diff --git a/simplyblock_core/services/tasks_runner_lvol_migration.py b/simplyblock_core/services/tasks_runner_lvol_migration.py index 2653514e9d..2c7699c462 100644 --- a/simplyblock_core/services/tasks_runner_lvol_migration.py +++ b/simplyblock_core/services/tasks_runner_lvol_migration.py @@ -581,7 +581,7 @@ def _get_source_tertiary_node(src_node): -def _build_paths(src_node, tgt_node, src_rpc, tgt_rpc): +def _build_paths(src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=None): """Build ordered path lists for source and target nodes and compute overlap. Returns (src_paths, tgt_paths, overlap_ids) where each path entry is: @@ -593,7 +593,20 @@ def _build_paths(src_node, tgt_node, src_rpc, tgt_rpc): Port is role-specific: SRC entries use src_node.lvstore; TGT entries use tgt_node.lvstore. Adding tertiary support = append one more entry to each list; all callers automatically handle it via loop/set operations. + + *src_node* is the node actually driving the transfer (position 0 in + src_paths) — normally the source primary, but the online secondary/ + tertiary when the primary was offline at migration create time + (migration.active_source_node_id). *primary_src_node* — defaulting to + src_node when the fallback isn't in play — is only consulted for its own + secondary_node_id/tertiary_node_id fields, which describe the primary's + true HA replicas; a replica node's own such fields describe an unrelated + pairing and must never be used for this lookup. Whichever replica was + chosen as src_node is excluded from the discovered peer set so it isn't + listed twice. """ + primary_src_node = primary_src_node or src_node + def _entry(node, rpc, lvstore): trtype, ip = _get_migration_nic(node) fabric = trtype.lower() @@ -610,19 +623,26 @@ def _entry(node, rpc, lvstore): 'node_id': node.get_id(), } - src_paths = [_entry(src_node, src_rpc, src_node.lvstore)] - if src_node.secondary_node_id: + # The lvstore NAME is always the true primary's own lvstore — a replica + # node hosts this data under the primary's lvstore name, not its own (a + # node's own .lvstore is whatever IT owns as a primary elsewhere in the + # HA ring, which is unrelated). Only the node/rpc/IP differ per entry. + src_lvstore = primary_src_node.lvstore + src_paths = [_entry(src_node, src_rpc, src_lvstore)] + _src_seen_ids = {src_node.get_id()} + if primary_src_node.secondary_node_id and primary_src_node.secondary_node_id not in _src_seen_ids: try: - ss = db.get_storage_node_by_id(src_node.secondary_node_id) + ss = db.get_storage_node_by_id(primary_src_node.secondary_node_id) if ss.status == StorageNode.STATUS_ONLINE: - src_paths.append(_entry(ss, _make_rpc(ss), src_node.lvstore)) + src_paths.append(_entry(ss, _make_rpc(ss), src_lvstore)) + _src_seen_ids.add(ss.get_id()) except KeyError: pass - if src_node.tertiary_node_id: + if primary_src_node.tertiary_node_id and primary_src_node.tertiary_node_id not in _src_seen_ids: try: - ts = db.get_storage_node_by_id(src_node.tertiary_node_id) + ts = db.get_storage_node_by_id(primary_src_node.tertiary_node_id) if ts.status == StorageNode.STATUS_ONLINE: - src_paths.append(_entry(ts, _make_rpc(ts), src_node.lvstore)) + src_paths.append(_entry(ts, _make_rpc(ts), src_lvstore)) except KeyError: pass @@ -869,7 +889,7 @@ def _setup_snap_transfer(snap, snap_index, src_node, tgt_node, src_rpc, tgt_rpc, trtype, tgt_sec=None, sec_rpc=None, tgt_ter=None, ter_rpc=None, lvol_size_mib=None, migration=None, - existing_bdev_info=_BDEV_INFO_UNSET): + existing_bdev_info=_BDEV_INFO_UNSET, primary_src_node=None): """ Prepare a single snapshot for async transfer: 1. Create writable lvol on target primary @@ -892,7 +912,7 @@ def _setup_snap_transfer(snap, snap_index, src_node, tgt_node, """ snap_uuid = snap.uuid snap_short = _snap_tgt_short_name(snap) - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite((primary_src_node or src_node).lvstore, snap) tgt_composite = f"{tgt_node.lvstore}/{snap_short}" # Step 1: create target lvol on primary. @@ -1169,7 +1189,7 @@ def _post_process_snap(snap: SnapShot, tgt_node: StorageNode, tgt_rpc: RPCClient return True, None -def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): +def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=None): """ Drive the SNAP_COPY phase. @@ -1210,6 +1230,10 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): plan = migration.snap_migration_plan trtype, _ = _get_migration_nic(tgt_node) ctx = migration.transfer_context or {} + # The lvstore NAME is always the true primary's own lvstore, regardless + # of which node (primary, secondary, or tertiary) is actually driving the + # transfer as src_node/src_rpc — see _build_paths' matching comment. + src_lvstore = (primary_src_node or src_node).lvstore # Snap bdevs on TGT must cover the full logical address range of the lvol, # not just each snap's own allocated clusters. @@ -1324,7 +1348,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): return False, True, f"Snapshot {snap_uuid} not found in DB" snap_short_tgt = _snap_tgt_short_name(snap) - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) tgt_composite = f"{tgt_node.lvstore}/{snap_short_tgt}" # Idempotency: transfer already running from a previous crashed run @@ -1377,7 +1401,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): tgt_ter=tgt_ter, ter_rpc=ter_rpc, lvol_size_mib=_snap_lvol_size_mib, migration=migration, - existing_bdev_info=_existing_bdev) + existing_bdev_info=_existing_bdev, primary_src_node=primary_src_node) if t is None: return False, True, err @@ -1436,7 +1460,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): migration.write_to_db(db.kv_store) return False, True, f"Snapshot {snap_uuid} disappeared during transfer" - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) # Update transfer-done status for this entry if not t['transfer_done']: @@ -1508,7 +1532,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): # additional shrink pass is worth the overhead. while migration.intermediate_snap_rounds < migration.max_intermediate_snap_rounds: _lvol = db.get_lvol_by_id(migration.lvol_id) - _src_composite = f"{src_node.lvstore}/{_lvol.lvol_bdev}" + _src_composite = f"{src_lvstore}/{_lvol.lvol_bdev}" _delta = _get_lvol_delta_bytes(src_rpc, _src_composite) _threshold = constants.LVOL_MIG_INTERMEDIATE_SNAP_THRESHOLD_BYTES if migration.intermediate_snap_rounds > 0 and _delta is not None and _delta <= _threshold: @@ -1564,7 +1588,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): ter_rpc = _make_rpc(tgt_ter) snap_short_tgt = _snap_tgt_short_name(snap) - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) tgt_composite = f"{tgt_node.lvstore}/{snap_short_tgt}" # Pre-cleanup: if a bdev exists on the target it is a writable leftover @@ -1600,7 +1624,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): tgt_ter=tgt_ter, ter_rpc=ter_rpc, lvol_size_mib=_snap_lvol_size_mib, migration=migration, - existing_bdev_info=_existing_bdev) + existing_bdev_info=_existing_bdev, primary_src_node=primary_src_node) if t is None: return False, True, err @@ -1719,7 +1743,7 @@ def _take_intermediate_snapshot(migration): ) -def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): +def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=None): """ Drive the LVOL_MIGRATE phase. @@ -1740,7 +1764,7 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): return False, True, str(e) trtype, _ = _get_migration_nic(tgt_node) - src_lvol_composite = f"{src_node.lvstore}/{lvol.lvol_bdev}" + src_lvol_composite = f"{(primary_src_node or src_node).lvstore}/{lvol.lvol_bdev}" tgt_lvol_bdev = _lvol_tgt_bdev_name(lvol.lvol_bdev) tgt_lvol_composite = f"{tgt_node.lvstore}/{tgt_lvol_bdev}" ctx = migration.transfer_context or {} @@ -1756,7 +1780,8 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): # overlap_ids: nodes that appear in BOTH source and target paths — they # already have a subsystem (from SRC role); their namespace is swapped in # the Done handler's step 4. - src_paths, tgt_paths, overlap_ids = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) + src_paths, tgt_paths, overlap_ids = _build_paths( + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) src_replica_paths = src_paths[1:] # secondary/tertiary only; primary stays live until cutover # Detect and repair a target-side node restart that wiped the migration's @@ -2493,7 +2518,7 @@ def _rename_with_fallback(current_short, label): lvol.write_to_db(db.kv_store) -def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): +def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc, primary_src_node=None): """ Best-effort source cleanup after a successful migration. The lvol is already live on the target — this phase only removes source-side artifacts @@ -2565,9 +2590,19 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): migration.transfer_context = ctx migration.write_to_db(db.kv_store) - src_sec = _get_source_secondary_node(src_node) + # Peer discovery must key off the true primary's own secondary_node_id/ + # tertiary_node_id fields (a replica node's own such fields describe an + # unrelated pairing — see _build_paths). Whichever replica is already + # src_node (the active fallback source) is excluded so it isn't cleaned + # up twice. + _primary_src_node = primary_src_node or src_node + src_sec = _get_source_secondary_node(_primary_src_node) + if src_sec is not None and src_sec.get_id() == src_node.get_id(): + src_sec = None src_sec_rpc = _make_rpc(src_sec) if src_sec else None - src_ter = _get_source_tertiary_node(src_node) + src_ter = _get_source_tertiary_node(_primary_src_node) + if src_ter is not None and src_ter.get_id() == src_node.get_id(): + src_ter = None src_ter_rpc = _make_rpc(src_ter) if src_ter else None # --- Delete source snapshots (best-effort, leader-routed) --- @@ -2578,12 +2613,12 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): try: snap = db.get_snapshot_by_id(snap_uuid) bdev_name = (source_snap_bdevs.get(snap_uuid) - or f"{src_node.lvstore}/{_snap_short_name(snap)}") + or f"{_primary_src_node.lvstore}/{_snap_short_name(snap)}") try: _delete_bdev_blocking(bdev_name, src_rpc, secondary_rpc=src_sec_rpc, tertiary_rpc=src_ter_rpc, all_nodes=[n for n in [src_node, src_sec, src_ter] if n], - lvs_name=src_node.lvstore) + lvs_name=_primary_src_node.lvstore) logger.info(f"Deleted source bdev {bdev_name}") except Exception as e: logger.warning(f"delete source bdev {bdev_name}: {e}") @@ -2606,7 +2641,7 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): else: logger.info(f"Step 8: removing source NVMe-oF subsystem {lvol.nqn}") _src_paths_cu, _, _overlap_ids_cu = _build_paths( - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=_primary_src_node) for _sp in _src_paths_cu: if _sp['node_id'] in _overlap_ids_cu: logger.info( @@ -2622,13 +2657,13 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): # lvol.lvol_bdev in the DB to the target name, so we must not use lvol.lvol_bdev. src_bdev_short = ctx.get('source_lvol_bdev') if lvol is not None and src_bdev_short: - src_lvol_composite = f"{src_node.lvstore}/{src_bdev_short}" + src_lvol_composite = f"{_primary_src_node.lvstore}/{src_bdev_short}" try: _delete_bdev_blocking( src_lvol_composite, src_rpc, secondary_rpc=src_sec_rpc, tertiary_rpc=src_ter_rpc, all_nodes=[n for n in [src_node, src_sec, src_ter] if n], - lvs_name=src_node.lvstore) + lvs_name=_primary_src_node.lvstore) logger.info(f"Deleted source lvol bdev {src_lvol_composite}") except Exception as e: logger.warning(f"Source lvol delete failed: {e}") @@ -2967,15 +3002,43 @@ def task_runner(task): migration_events.migration_phase_changed(migration) # --- Load nodes --- + # primary_src_node is the true primary (lvol.node_id) — kept only for HA + # topology lookups (its own secondary_node_id/tertiary_node_id fields). + # src_node is the node this migration actually issues source-side RPCs + # against: the primary when reachable, otherwise the online replica + # pinned once at create_migration() time as active_source_node_id. Every + # data-plane call below must use src_node/src_rpc, never primary_src_node. + # Group workers must never give up through the plain (non-group-aware) + # _budget_suspend below: it only marks this one migration record + # CLEANUP_TARGET, never the shared group.phase, so the orchestrator's + # barrier (waiting for every member's snap_copy_done/intermediates_done) + # never learns this worker is gone and waits for it forever. Route through + # _group_worker_budget_suspend instead whenever this migration belongs to + # a batch group, so one member failing here fails the whole group instead + # of hanging it (discovered 2026-09-04: a fallback migration whose active + # source node went non-online mid-INTERMEDIATE left all 3 workers silently + # CLEANUP_TARGET'd while the group orchestrator polled "waiting for 3 + # workers" forever, past the test's own 10-minute timeout). + def _node_lookup_suspend(error_msg): + if migration.migration_group_id: + return _group_worker_budget_suspend(task, migration, migration.migration_group_id, error_msg) + return _budget_suspend(task, migration, migration_id, error_msg) + try: - src_node = db.get_storage_node_by_id(migration.source_node_id) + primary_src_node = db.get_storage_node_by_id(migration.source_node_id) except KeyError: - return _budget_suspend(task, migration, migration_id, "source node not found") + return _node_lookup_suspend("source node not found") + + try: + src_node = db.get_storage_node_by_id( + migration.active_source_node_id or migration.source_node_id) + except KeyError: + return _node_lookup_suspend("active source node not found") try: tgt_node = db.get_storage_node_by_id(migration.target_node_id) except KeyError: - return _budget_suspend(task, migration, migration_id, "target node not found") + return _node_lookup_suspend("target node not found") # Cleanup phases proceed regardless of node status: deletes go through LVS # leadership, so a downed node doesn't block the cleanup path. @@ -2983,9 +3046,7 @@ def task_runner(task): LVolMigration.PHASE_CLEANUP_TARGET, LVolMigration.PHASE_CLEANUP_SOURCE) if not _is_cleanup_phase: if src_node.status not in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): - return _budget_suspend( - task, migration, migration_id, - f"source node not online (status={src_node.status})") + return _node_lookup_suspend(f"source node not online (status={src_node.status})") if tgt_node.status != StorageNode.STATUS_ONLINE: if (migration.phase in (LVolMigration.PHASE_SNAP_COPY, @@ -3006,6 +3067,8 @@ def task_runner(task): migration.write_to_db(db.kv_store) task.write_to_db(db.kv_store) migration_events.migration_phase_changed(migration) + if migration.migration_group_id: + _fail_group_from_worker(migration, migration.migration_group_id, migration.error_message) return False if not _is_cleanup_phase: # cleanup phases are exempt: deletes go through LVS leadership; @@ -3053,20 +3116,25 @@ def task_runner(task): try: if migration.migration_group_id: return _group_worker_phase_dispatch( - task, migration, phase, src_node, tgt_node, src_rpc, tgt_rpc) + task, migration, phase, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=primary_src_node) if phase == LVolMigration.PHASE_SNAP_COPY: done, suspend, error = _handle_snap_copy( - migration, src_node, tgt_node, src_rpc, tgt_rpc) + migration, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=primary_src_node) next_phase = LVolMigration.PHASE_LVOL_MIGRATE elif phase == LVolMigration.PHASE_LVOL_MIGRATE: done, suspend, error = _handle_lvol_migrate( - migration, src_node, tgt_node, src_rpc, tgt_rpc) + migration, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=primary_src_node) next_phase = LVolMigration.PHASE_CLEANUP_SOURCE elif phase == LVolMigration.PHASE_CLEANUP_SOURCE: - done, suspend, error = _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc) + done, suspend, error = _handle_cleanup_source( + migration, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) next_phase = LVolMigration.PHASE_COMPLETED elif phase == LVolMigration.PHASE_CLEANUP_TARGET: @@ -3221,7 +3289,7 @@ def _post_process_snap_group(snap, migration): return True, None -def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): +def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=None): """ SNAP_COPY phase for a group worker. @@ -3235,6 +3303,10 @@ def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): plan = migration.snap_migration_plan trtype, _ = _get_migration_nic(tgt_node) ctx = migration.transfer_context or {} + # See _build_paths' matching comment: the lvstore NAME is always the true + # primary's own lvstore, regardless of which node is actually driving the + # transfer as src_node/src_rpc. + src_lvstore = (primary_src_node or src_node).lvstore try: _lvol_for_size = db.get_lvol_by_id(migration.lvol_id) @@ -3258,7 +3330,7 @@ def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): return False, True, f"Snapshot {snap_uuid} not found in DB" snap_short_tgt = _snap_tgt_short_name(snap) - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) tgt_composite = f"{tgt_node.lvstore}/{snap_short_tgt}" existing_stat = src_rpc.bdev_lvol_transfer_stat(src_composite) @@ -3311,7 +3383,7 @@ def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): tgt_ter=_g_tgt_ter, ter_rpc=_g_ter_rpc, lvol_size_mib=_snap_lvol_size_mib, migration=migration, - existing_bdev_info=_existing_bdev) + existing_bdev_info=_existing_bdev, primary_src_node=primary_src_node) if t is None: return False, True, err @@ -3336,7 +3408,7 @@ def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): migration.write_to_db(db.kv_store) return False, True, f"Snapshot {snap_uuid} disappeared during transfer" - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) if not t['transfer_done']: result = src_rpc.bdev_lvol_transfer_stat(src_composite) if result is None: @@ -3374,7 +3446,8 @@ def _handle_group_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): return True, False, None -def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, target_round=0): +def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, + target_round=0, primary_src_node=None): """ INTERMEDIATE phase for a group worker. @@ -3393,6 +3466,10 @@ def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, """ trtype, _ = _get_migration_nic(tgt_node) ctx = migration.transfer_context or {} + # See _build_paths' matching comment: the lvstore NAME is always the true + # primary's own lvstore, regardless of which node is actually driving the + # transfer as src_node/src_rpc. + src_lvstore = (primary_src_node or src_node).lvstore # If we already took and transferred the intermediate snap for the round # the group is currently on, we're done. Otherwise the group has asked @@ -3463,7 +3540,7 @@ def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, tgt_ter=_g_tgt_ter, ter_rpc=_g_ter_rpc, lvol_size_mib=_snap_lvol_size_mib, migration=migration, - existing_bdev_info=_existing_bdev) + existing_bdev_info=_existing_bdev, primary_src_node=primary_src_node) if t is None: return False, True, err @@ -3484,7 +3561,7 @@ def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, migration.write_to_db(db.kv_store) return False, True, f"Intermediate snap {snap_uuid} disappeared" - src_composite = _snap_composite(src_node.lvstore, snap) + src_composite = _snap_composite(src_lvstore, snap) if not t.get('transfer_done'): result = src_rpc.bdev_lvol_transfer_stat(src_composite) if result is None: @@ -3511,6 +3588,38 @@ def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, return True, False, None +def _fail_group_from_worker(migration, group_id, error_msg): + """Force the whole group into CLEANUP_TARGET because one member just gave + up (retry budget exhausted, or an unconditional hard-fail like the target + going offline mid-transfer). + + A worker that stops here without this will never signal its barrier + (snap_copy_done / intermediates_done / cleanup_source_done) again -- the + orchestrator's barrier check has no way to distinguish "still working" from + "silently gone", so it waits forever instead of failing the group. Every + place a group worker can reach a terminal state outside its own normal + phase progression must call this. + """ + try: + group = db.get_migration_group_by_id(group_id) + if group.phase not in (LVolMigrationGroup.PHASE_CLEANUP_TARGET, + LVolMigrationGroup.PHASE_CLEANUP_SOURCE, + LVolMigrationGroup.PHASE_COMPLETED): + group.phase = LVolMigrationGroup.PHASE_CLEANUP_TARGET + group.error_message = ( + f"worker {migration.uuid[:8]} (lvol={migration.lvol_id}) failed: {error_msg}") + group.write_to_db(db.kv_store) + logger.error( + f"Group {group_id[:8]}: failing whole group — worker " + f"{migration.uuid[:8]} failed: {error_msg}") + except KeyError: + # Group may already be removed/cleaned up by another workflow. + # We keep worker cleanup flow idempotent by not re-raising. + logger.warning( + f"Group {group_id[:8]} not found while propagating worker " + f"{migration.uuid[:8]} failure; continuing.") + + def _group_worker_budget_suspend(task, migration, group_id, error_msg): """Charge retry budget for a group worker; fail the WHOLE GROUP when this worker's own budget is exhausted. @@ -3541,30 +3650,13 @@ def _group_worker_budget_suspend(task, migration, group_id, error_msg): # This worker will never signal done to its barrier now -- fail the # whole group rather than let siblings (and the orchestrator) wait # on it forever. - try: - group = db.get_migration_group_by_id(group_id) - if group.phase not in (LVolMigrationGroup.PHASE_CLEANUP_TARGET, - LVolMigrationGroup.PHASE_CLEANUP_SOURCE, - LVolMigrationGroup.PHASE_COMPLETED): - group.phase = LVolMigrationGroup.PHASE_CLEANUP_TARGET - group.error_message = ( - f"worker {migration.uuid[:8]} (lvol={migration.lvol_id}) " - f"exceeded max retries: {error_msg}") - group.write_to_db(db.kv_store) - logger.error( - f"Group {group_id[:8]}: failing whole group — worker " - f"{migration.uuid[:8]} exhausted its retry budget") - except KeyError: - # Group may already be removed/cleaned up by another workflow. - # We keep worker cleanup flow idempotent by not re-raising. - logger.warning( - f"Group {group_id[:8]} not found while propagating worker " - f"{migration.uuid[:8]} retry-budget exhaustion; continuing.") + _fail_group_from_worker(migration, group_id, error_msg) return False return _suspend_task(task, migration, error_msg) -def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src_rpc, tgt_rpc): +def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=None): """ Complete phase dispatcher for FN_LVOL_MIG tasks that belong to a batch migration group (``migration.migration_group_id`` is set). @@ -3595,7 +3687,8 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src # Still transferring owned snaps. try: done, suspend, error = _handle_group_snap_copy( - migration, src_node, tgt_node, src_rpc, tgt_rpc) + migration, src_node, tgt_node, src_rpc, tgt_rpc, + primary_src_node=primary_src_node) except RPCException as exc: # Charge this worker's own retry budget and report failure to # the group -- never decide/roll back unilaterally (see @@ -3628,13 +3721,13 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src migration_events.migration_phase_changed(migration) return _group_worker_phase_dispatch( task, migration, LVolMigration.PHASE_LVOL_MIGRATE, - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) if group.phase == LVolMigrationGroup.PHASE_CLEANUP_TARGET: migration.phase = LVolMigration.PHASE_CLEANUP_TARGET migration.write_to_db(db.kv_store) return _group_worker_phase_dispatch( task, migration, LVolMigration.PHASE_CLEANUP_TARGET, - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) # Still waiting for other workers. task.write_to_db(db.kv_store) return False @@ -3651,12 +3744,12 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src migration.write_to_db(db.kv_store) return _group_worker_phase_dispatch( task, migration, LVolMigration.PHASE_CLEANUP_TARGET, - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) try: done, suspend, error = _handle_group_intermediate( migration, src_node, tgt_node, src_rpc, tgt_rpc, - target_round=group.intermediate_round) + target_round=group.intermediate_round, primary_src_node=primary_src_node) except RPCException as exc: # Charge this worker's own retry budget and report failure to # the group -- never decide/roll back unilaterally (see @@ -3702,6 +3795,18 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src # intermediates_done signalled — wait for batch_result. group = db.get_migration_group_by_id(group_id) + if group.batch_result is None and group.phase == LVolMigrationGroup.PHASE_CLEANUP_TARGET: + # A sibling exhausted its retry budget and forced the group into + # cleanup without ever setting batch_result (only the normal + # INTERMEDIATE barrier path sets it) -- notice the forced phase + # directly instead of polling batch_result forever. + migration.phase = LVolMigration.PHASE_CLEANUP_TARGET + migration.transfer_context = {} + migration.write_to_db(db.kv_store) + migration_events.migration_phase_changed(migration) + return _group_worker_phase_dispatch( + task, migration, LVolMigration.PHASE_CLEANUP_TARGET, + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) if group.batch_result is True: lvol = db.get_lvol_by_id(migration.lvol_id) migration.phase = LVolMigration.PHASE_CLEANUP_SOURCE @@ -3713,7 +3818,7 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src migration_events.migration_phase_changed(migration) return _group_worker_phase_dispatch( task, migration, LVolMigration.PHASE_CLEANUP_SOURCE, - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) if group.batch_result is False: migration.phase = LVolMigration.PHASE_CLEANUP_TARGET migration.transfer_context = {} @@ -3721,7 +3826,7 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src migration_events.migration_phase_changed(migration) return _group_worker_phase_dispatch( task, migration, LVolMigration.PHASE_CLEANUP_TARGET, - src_node, tgt_node, src_rpc, tgt_rpc) + src_node, tgt_node, src_rpc, tgt_rpc, primary_src_node=primary_src_node) task.write_to_db(db.kv_store) return False @@ -3729,7 +3834,8 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src if phase == LVolMigration.PHASE_CLEANUP_SOURCE: try: done, suspend, error = _handle_cleanup_source( - migration, src_node, src_rpc, tgt_node, tgt_rpc) + migration, src_node, src_rpc, tgt_node, tgt_rpc, + primary_src_node=primary_src_node) except RPCException as exc: return _suspend_task(task, migration, str(exc)) diff --git a/simplyblock_web/api/v2/_dtos.py b/simplyblock_web/api/v2/_dtos.py index 172e89e24c..b3242cbfd4 100644 --- a/simplyblock_web/api/v2/_dtos.py +++ b/simplyblock_web/api/v2/_dtos.py @@ -648,6 +648,7 @@ class MigrationDTO(BaseModel): id: UUID lvol_id: str source_node_id: str + active_source_node_id: str target_node_id: str phase: str status: str @@ -668,6 +669,7 @@ def from_model(model: LVolMigration, connect_strings: Optional[List[NvmeConnectE id=UUID(model.uuid), lvol_id=model.lvol_id, source_node_id=model.source_node_id, + active_source_node_id=model.active_source_node_id or model.source_node_id, target_node_id=model.target_node_id, phase=model.phase, status=model.status, @@ -688,6 +690,7 @@ class BatchMigrationDTO(BaseModel): id: UUID cluster_id: str source_node_id: str + active_source_node_id: str target_node_id: str target_nqn: str phase: str @@ -702,6 +705,7 @@ def from_model(model: LVolMigrationGroup, connect_strings: Optional[List[NvmeCon id=UUID(model.uuid), cluster_id=model.cluster_id, source_node_id=model.source_node_id, + active_source_node_id=model.active_source_node_id or model.source_node_id, target_node_id=model.target_node_id, target_nqn=model.target_nqn, phase=model.phase, diff --git a/simplyblock_web/api/v2/cluster/subsystem/migration.py b/simplyblock_web/api/v2/cluster/subsystem/migration.py index a767d3b46c..85b5bcd75f 100644 --- a/simplyblock_web/api/v2/cluster/subsystem/migration.py +++ b/simplyblock_web/api/v2/cluster/subsystem/migration.py @@ -96,7 +96,15 @@ def create_migration( ctrl_loss_tmo=parameters.ctrl_loss_tmo, host_nqn=parameters.host_nqn, ) - except (ValueError, MigrationConflictError, PreconditionError, RuntimeError) as e: + except (MigrationConflictError, PreconditionError) as e: + # Conflicting/not-yet-satisfiable state (e.g. a migration already + # active for this subsystem, or -- for a fallback-source migration -- + # the chosen target is the node currently serving as the fallback + # source itself) -- matches the 409 convention used for the same + # shape of error elsewhere in v2 (storage_node shutdown, pool/volume + # already-exists, in-flight replication cutover). + raise HTTPException(409, str(e)) + except (ValueError, RuntimeError) as e: raise HTTPException(400, str(e)) def get_full(id): @@ -149,7 +157,15 @@ def continue_migration(migration: SubsystemMigration, parameters: _ContinueParam max_retries=parameters.max_retries, deadline_seconds=parameters.deadline_seconds, ) - except (ValueError, MigrationConflictError, PreconditionError, RuntimeError) as e: + except (MigrationConflictError, PreconditionError) as e: + # Conflicting/not-yet-satisfiable state (e.g. a migration already + # active for this subsystem, or -- for a fallback-source migration -- + # the chosen target is the node currently serving as the fallback + # source itself) -- matches the 409 convention used for the same + # shape of error elsewhere in v2 (storage_node shutdown, pool/volume + # already-exists, in-flight replication cutover). + raise HTTPException(409, str(e)) + except (ValueError, RuntimeError) as e: raise HTTPException(400, str(e)) return {"migration_id": result_id} diff --git a/tests/integration/migration/test_unit_controller.py b/tests/integration/migration/test_unit_controller.py index 5865159c86..362d7ba247 100644 --- a/tests/integration/migration/test_unit_controller.py +++ b/tests/integration/migration/test_unit_controller.py @@ -12,10 +12,12 @@ import pytest from simplyblock_core.models.lvol_migration import LVolMigration +from simplyblock_core.models.lvol_migration_group import LVolMigrationGroup from simplyblock_core.models.lvol_model import LVol from simplyblock_core.models.snapshot import SnapShot from simplyblock_core.models.storage_node import StorageNode from simplyblock_core.models.cluster import Cluster +from simplyblock_core.exceptions import PreconditionError # Module under test (import after patching, but top-level import is fine since # we patch the db attribute before each individual call). @@ -692,6 +694,55 @@ def test_reject_target_node_offline(self): with pytest.raises(ValueError, match="Target node is not online"): ctl.start_migration("mig-uuid") + def test_reject_degraded_cluster_for_non_fallback_source(self): + # Primary is the active source (no fallback) -- a DEGRADED cluster + # must still be rejected here, matching the pre-existing strict + # ACTIVE-only gate for ordinary migrations. + mig = self._pre_created() + lvol = _lvol("lvol-1", "node-src") + src = _node("node-src") + tgt = _node("node-tgt") + mock_db = self._base_db(mig, lvol, src, tgt) + mock_db.get_cluster_by_id.return_value.status = Cluster.STATUS_DEGRADED + with patch.object(ctl, 'db', mock_db): + with pytest.raises(PreconditionError, match="not active"): + ctl.start_migration("mig-uuid") + + def test_allow_degraded_cluster_for_fallback_source(self): + # Primary is offline; active_source_node_id was pinned to the + # secondary at create_migration() time. The primary being down is + # exactly what drives the cluster to DEGRADED, so DEGRADED must be + # allowed here or the fallback feature could never actually run. + mig = self._pre_created() + mig.active_source_node_id = "node-sec" + lvol = _lvol("lvol-1", "node-src") + src = _node("node-src", status=StorageNode.STATUS_OFFLINE) + sec = _node("node-sec") + tgt = _node("node-tgt") + + mock_db = MagicMock() + mock_db.get_migration_by_id.return_value = mig + mock_db.get_lvol_by_id.return_value = lvol + nodes_by_id = {"node-src": src, "node-sec": sec, "node-tgt": tgt} + mock_db.get_storage_node_by_id.side_effect = lambda nid: nodes_by_id[nid] + mock_db.get_snapshots_by_node_id.return_value = [ + _snap("s1", "lvol-1", "node-src")] + cluster = Cluster() + cluster.uuid = "cluster-1" + cluster.status = Cluster.STATUS_DEGRADED + mock_db.get_cluster_by_id.return_value = cluster + mock_db.get_job_tasks.return_value = [] + mock_db.kv_store = MagicMock() + + with patch.object(ctl, 'db', mock_db), \ + patch('simplyblock_core.controllers.migration_controller.tasks_controller') as tc, \ + patch('simplyblock_core.controllers.migration_controller.migration_events'): + tc.add_lvol_mig_task.return_value = "task-uuid" + tc.get_active_node_mig_task.return_value = None + result = ctl.start_migration("mig-uuid") + + assert result == "mig-uuid" + def test_success_creates_task(self): mig = self._pre_created() lvol = _lvol("lvol-1", "node-src") @@ -714,3 +765,59 @@ def test_success_creates_task(self): assert result == "mig-uuid" tc.add_lvol_mig_task.assert_called_once() + + +# --------------------------------------------------------------------------- +# start_batch_migration – precondition validation (same DEGRADED/fallback +# gate as start_migration, applied to LVolMigrationGroup) +# --------------------------------------------------------------------------- + +class TestStartBatchMigrationPreconditions(unittest.TestCase): + + def _group(self, active_source_node_id=""): + g = LVolMigrationGroup() + g.uuid = "group-uuid" + g.cluster_id = "cluster-1" + g.source_node_id = "node-src" + g.target_node_id = "node-tgt" + g.active_source_node_id = active_source_node_id + g.members = [] + g.snap_owners = {} + g.phase = LVolMigrationGroup.PHASE_PRE_CREATED + return g + + def test_reject_degraded_cluster_for_non_fallback_source(self): + group = self._group() # active_source_node_id empty -> falls back to source_node_id + src = _node("node-src") + mock_db = MagicMock() + mock_db.get_migration_group_by_id.return_value = group + mock_db.get_storage_node_by_id.return_value = src + cluster = Cluster() + cluster.uuid = "cluster-1" + cluster.status = Cluster.STATUS_DEGRADED + mock_db.get_cluster_by_id.return_value = cluster + with patch.object(ctl, 'db', mock_db): + with pytest.raises(PreconditionError, match="not active"): + ctl.start_batch_migration("group-uuid") + + def test_allow_degraded_cluster_for_fallback_source(self): + group = self._group(active_source_node_id="node-sec") + sec = _node("node-sec") + mock_db = MagicMock() + mock_db.get_migration_group_by_id.return_value = group + mock_db.get_storage_node_by_id.return_value = sec + cluster = Cluster() + cluster.uuid = "cluster-1" + cluster.status = Cluster.STATUS_DEGRADED + mock_db.get_cluster_by_id.return_value = cluster + mock_db.get_job_tasks.return_value = [] + mock_db.kv_store = MagicMock() + + with patch.object(ctl, 'db', mock_db), \ + patch('simplyblock_core.controllers.migration_controller.tasks_controller') as tc: + tc.get_active_node_mig_task.return_value = None + tc.add_batch_mig_task.return_value = "task-uuid" + result = ctl.start_batch_migration("group-uuid") + + assert result == "group-uuid" + tc.add_batch_mig_task.assert_called_once() diff --git a/tests/unit/tasks/test_retry_ceiling.py b/tests/unit/tasks/test_retry_ceiling.py index c3769966dc..b5d514cd61 100644 --- a/tests/unit/tasks/test_retry_ceiling.py +++ b/tests/unit/tasks/test_retry_ceiling.py @@ -442,6 +442,12 @@ def _spec_batch_migration(runner, monkeypatch): group = MagicMock() group.phase = LVolMigrationGroup.PHASE_SNAP_COPY group.source_node_id = "src-1" + # MagicMock auto-creates a truthy attribute for anything unset; without + # this, `group.active_source_node_id or group.source_node_id` in the + # runner picks the mock object instead of "src-1", silently routing node + # lookups to the tgt_node branch below and masking the source-offline + # retry path this test exercises. + group.active_source_node_id = "src-1" group.target_node_id = "tgt-1" group.cluster_id = "cl-1" group.members = [{"migration_id": "mig-1"}]