diff --git a/simplyblock_core/controllers/migration_controller.py b/simplyblock_core/controllers/migration_controller.py index b723e2575..bfeb78171 100644 --- a/simplyblock_core/controllers/migration_controller.py +++ b/simplyblock_core/controllers/migration_controller.py @@ -1174,7 +1174,8 @@ def create_migration(lvol_id, target_node_id, f"create_migration: listener on overlap {_node_id[:8]} " f"(non-fatal): {_e}") else: - if not _rpc.subsystem_get(nqn): + _existing_subsys = _rpc.subsystem_get(nqn) + if not _existing_subsys: if _min_cntlid in subsys_min_cntlid_used: _min_cntlid = _min_cntlid + 10000 _rpc.subsystem_create( @@ -1190,22 +1191,51 @@ def create_migration(lvol_id, target_node_id, f"create_migration: allowed_hosts reapply on " f"{_node_id[:8]} (non-fatal): {_e}") + # For a shared-namespace batch group, create_migration() runs once + # per member against the SAME nqn/listener -- guard against + # re-adding a listener that a prior member's precreate already + # established (listeners_create had no existence check and its + # result was never inspected, so a real failure on a later + # member's redundant add would previously have been silent). + _existing_listeners = { + (_l.get('trtype', '').lower(), _l.get('traddr'), str(_l.get('trsvcid'))) + for _l in ((_existing_subsys or {}).get('listen_addresses') or []) + } for nic in _node.data_nics: if not nic.ip4_address or nic.trtype.lower() != lvol.fabric: continue + _listener_key = (nic.trtype.lower(), nic.ip4_address, str(_port)) + if _listener_key in _existing_listeners: + continue try: - _rpc.listeners_create(nqn, nic.trtype.lower(), nic.ip4_address, - _port, ana_state="inaccessible") + _ret_listener = _rpc.listeners_create( + nqn, nic.trtype.lower(), nic.ip4_address, + _port, ana_state="inaccessible") + if not _ret_listener: + logger.warning( + f"create_migration: listener add for {_node_id[:8]} " + f"{nic.ip4_address}:{_port} returned falsy") except Exception as _e: logger.warning( f"create_migration: listener on {_node_id[:8]} " f"(non-fatal): {_e}") - _ns = _rpc.nvmf_subsystem_add_ns(nqn, _ns_bdev, lvol.uuid, lvol.guid) + # Pin the target namespace to the SAME nsid the source already + # uses, rather than letting SPDK auto-assign on the target + # subsystem. Auto-assignment just happens to reproduce the + # source's numbering when adds land in the same order on an + # empty subsystem -- it isn't enforced, and any stale/leftover + # namespace occupying a low nsid on the target (or add_ns calls + # racing/reordering across nodes) would silently diverge the + # source and target nsid maps for this lvol. + _ns = _rpc.nvmf_subsystem_add_ns( + nqn, _ns_bdev, lvol.uuid, lvol.guid, + nsid=lvol.ns_id if lvol.ns_id else None) if _ns: logger.info( f"create_migration: namespace {_ns_bdev} added on " - f"{_tgt_label} {_node_id[:8]} nsid={_ns}") + f"{_tgt_label} {_node_id[:8]} nsid={_ns} " + f"(source nsid={lvol.ns_id})") else: logger.warning( f"create_migration: nvmf_subsystem_add_ns failed on " diff --git a/simplyblock_core/models/lvol_migration_group.py b/simplyblock_core/models/lvol_migration_group.py index 2641cf452..09d29dab2 100644 --- a/simplyblock_core/models/lvol_migration_group.py +++ b/simplyblock_core/models/lvol_migration_group.py @@ -14,8 +14,14 @@ N worker tasks copy their owned snapshot chains in parallel. snap_copy_done tracks which workers have finished. INTERMEDIATE - All workers take exactly one intermediate ('shrink') snapshot each. - intermediates_done tracks which workers have finished. + All workers take one intermediate ('shrink') snapshot each, in lockstep + rounds. intermediates_done tracks which workers have finished the + current round (intermediate_round). If any worker's dirty delta is + still above the threshold after a round, it flags itself in + intermediate_more_needed; once every worker has finished the round, the + orchestrator starts another synchronized round (all members retake a + snapshot together, even ones whose own delta was already low) if + intermediate_more_needed is non-empty and the round cap hasn't been hit. BATCH_MIGRATE Main calls bdev_lvol_batch_final_step with all lvols ordered by ns_id. batch_result is set to True on success, False on failure. @@ -80,10 +86,22 @@ class LVolMigrationGroup(BaseModel): # waiting for the INTERMEDIATE phase signal from the main orchestrator. snap_copy_done: List[str] = [] - # migration_ids that have taken and transferred their single intermediate - # snapshot and are waiting for batch_result. + # migration_ids that have taken and transferred their intermediate + # snapshot for the CURRENT intermediate_round and are waiting for either + # another round or batch_result. Cleared when a new round starts. intermediates_done: List[str] = [] + # Which intermediate round is currently in flight (0-indexed; round 0 is + # always taken unconditionally). Incremented when the orchestrator starts + # another synchronized round. + intermediate_round: int = 0 + + # migration_ids that reported their dirty delta still exceeded the + # threshold after finishing intermediate_round. Cleared when a new round + # starts. Non-empty at the end of a round (and under the round cap) + # triggers another synchronized round for every member. + intermediate_more_needed: List[str] = [] + # migration_ids that have completed CLEANUP_SOURCE. cleanup_source_done: List[str] = [] diff --git a/simplyblock_core/services/hub_controller_manager.py b/simplyblock_core/services/hub_controller_manager.py index 9e5b9a0a3..092b89558 100644 --- a/simplyblock_core/services/hub_controller_manager.py +++ b/simplyblock_core/services/hub_controller_manager.py @@ -93,7 +93,7 @@ class HubControllerManager: # Seconds since the last acquire() before the GC triggers a detach. # Refreshed on every acquire() so concurrent migrations naturally keep # the controller alive without any reference counting. - IDLE_TIMEOUT = 300 # 5 minutes + IDLE_TIMEOUT = 1200 # 20 minutes # GC sweep period. GC_INTERVAL = 30 diff --git a/simplyblock_core/services/tasks_runner_batch_migration.py b/simplyblock_core/services/tasks_runner_batch_migration.py index 5e45b6f72..f237878bc 100644 --- a/simplyblock_core/services/tasks_runner_batch_migration.py +++ b/simplyblock_core/services/tasks_runner_batch_migration.py @@ -17,8 +17,11 @@ Advance group to PHASE_INTERMEDIATE. PHASE_INTERMEDIATE (orchestrator: wait + batch-final) - Wait for all workers to signal intermediates_done. - Build the batch-final-step argument lists (one entry per member, ordered + Wait for all workers to signal intermediates_done for the current round. + If any worker's dirty delta is still above the threshold, start another + synchronized round (every member retakes a snapshot together) up to + LVOL_MIG_MAX_INTERMEDIATE_SNAPS rounds. Once no more rounds are needed, + build the batch-final-step argument lists (one entry per member, ordered by ns_id), acquire a shared hub connection via hub_manager, and call bdev_lvol_batch_final_step on the source node. Set group.batch_result = True/False. @@ -98,13 +101,29 @@ def _reconstruct_snap_tree(group, member_migrations, tgt_node, tgt_rpc) -> Optio tgt_ter, _ = _get_target_tertiary_node(tgt_node, "") ter_rpc = _make_rpc(tgt_ter) if tgt_ter else None + # A member's snaps_preexisting_on_target (set at create_batch_migration_continue + # time) conflates two very different things: snaps truly already on the + # target from OUTSIDE this group (a prior, unrelated migration), and + # "non_owned_preexisting" snaps -- ancestor snaps in this member's own + # chain that a DIFFERENT member of THIS SAME group owns and hasn't + # transferred/committed yet. Only the former may seed `committed` up + # front; seeding from the latter marks every ancestor snap "already + # committed" before its true owner ever gets a turn in the loop below, + # so add_clone/convert never runs for ANY snapshot in the tree. + owned_or_pending_uuids: set = set() + for m in member_migrations: + owned_or_pending_uuids.update(m.snap_migration_plan or []) + owned_or_pending_uuids.update(m.snaps_transferred_group or []) + # Global set of snaps that have been committed as immutable on the target, # either pre-existing or reconstructed in this call. committed: set = set() all_preexisting: set = set() for m in member_migrations: - committed.update(m.snaps_preexisting_on_target) - all_preexisting.update(m.snaps_preexisting_on_target) + truly_external_preexisting = [ + s for s in m.snaps_preexisting_on_target if s not in owned_or_pending_uuids] + committed.update(truly_external_preexisting) + all_preexisting.update(truly_external_preexisting) # Snaps already committed in a prior (crashed) run are in snaps_migrated. # Seeding committed from them prevents re-convert on re-entry (SPDK rejects # converting an already-immutable bdev, which would stall the group forever). @@ -334,22 +353,23 @@ def _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node """ After a successful bdev_lvol_batch_final_step, drive clients to the new target. - Mirrors the single-lvol Done-handler ANA sequence exactly: + Mirrors the single-lvol Done-handler ANA sequence, minus the SRC-inaccessible + step: _handle_intermediate_barrier's pre-final-step freeze already drives + every SRC path (primary + secondary/tertiary) inaccessible before + batch_final_step even runs, so re-flipping them here would just repeat + the same RPC calls with no effect. No-overlap: 1. TGT primary → optimized (required; logs error on failure but continues since bdev_lvol_batch_final_step cannot be undone) 2. TGT secondary/tertiary → non_optimized - 3. All SRC paths → inaccessible Overlap: 1. First non-overlap TGT → optimized (live path before touching overlap) - 2. Overlap SRC paths → inaccessible (at SRC port) - 3. Non-overlap SRC paths → inaccessible - 4. Namespace swap on overlap TGT paths: SRC bdev → migrated TGT bdev + 2. Namespace swap on overlap TGT paths: SRC bdev → migrated TGT bdev (uses _swap_namespace which dynamically re-queries nsid; respects crypto_bdev) - 5. All TGT paths → correct ANA state at TGT port - 6. Remove old SRC-port listener from overlap TGT nodes if port changed + 3. All TGT paths → correct ANA state at TGT port + 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) @@ -419,13 +439,10 @@ def _flip_all_required(rpc, ips, port, trtype, state, label, attempts=3): # Step 2: TGT secondary/tertiary → non_optimized for i, tp in enumerate(tgt_paths[1:], 1): _flip_all(tp['rpc'], tp['ips'], tp['port'], tp['trtype'], "non_optimized", f"TGT-rep{i}") - - # Step 3: all SRC paths → inaccessible - for src in src_paths: - _flip_all(src['rpc'], src['ips'], src['port'], src['trtype'], - "inaccessible", f"SRC-{src['node_id'][:8]}") else: - # Step 1: first non-overlap TGT → optimized before making any SRC inaccessible + # Step 1: first non-overlap TGT → optimized. SRC paths (overlap and + # non-overlap alike) are already inaccessible from the pre-final-step + # freeze -- no need to re-flip them here. non_overlap_tgt = next( (t for t in tgt_paths if t['node_id'] not in overlap_ids), None) if non_overlap_tgt: @@ -437,19 +454,7 @@ def _flip_all_required(rpc, ips, port, trtype, state, label, attempts=3): f"Group {group.uuid[:8]}: ANA flip non-overlap TGT→optimized failed; " f"proceeding anyway") - # Step 2: overlap SRC paths → inaccessible at SRC port - for src in src_paths: - if src['node_id'] in overlap_ids: - _flip_all(src['rpc'], src['ips'], src['port'], src['trtype'], - "inaccessible", f"SRC-{src['node_id'][:8]}(overlap)") - - # Step 3: non-overlap SRC paths → inaccessible - for src in src_paths: - if src['node_id'] not in overlap_ids: - _flip_all(src['rpc'], src['ips'], src['port'], src['trtype'], - "inaccessible", f"SRC-{src['node_id'][:8]}") - - # Step 4: namespace swap on overlap TGT paths. + # Step 2: namespace swap on overlap TGT paths. # Each member has its own namespace in the shared NQN. We look up each # member's nsid by matching ns['uuid'] == lvol.uuid so we never remove # the wrong namespace (positional removal would corrupt I/O for other members). @@ -520,18 +525,18 @@ def _flip_all_required(rpc, ips, port, trtype, state, label, attempts=3): f"Group {group.uuid[:8]}: add ns {tgt_ns_bdev} " f"on {tgt['node_id'][:8]} (non-fatal): {e}") - # Step 5: all TGT paths → correct ANA state at TGT port + # Step 3: all TGT paths → correct ANA state at TGT port primary_tgt = tgt_paths[0] if not _flip_all_required(primary_tgt['rpc'], primary_tgt['ips'], primary_tgt['port'], primary_tgt['trtype'], "optimized", f"TGT-{primary_tgt['node_id'][:8]}"): logger.error( - f"Group {group.uuid[:8]}: ANA flip TGT primary→optimized (step 5) failed") + f"Group {group.uuid[:8]}: ANA flip TGT primary→optimized (step 3) failed") for tgt in tgt_paths[1:]: _flip_all(tgt['rpc'], tgt['ips'], tgt['port'], tgt['trtype'], "non_optimized", f"TGT-{tgt['node_id'][:8]}") - # Step 6: remove old SRC-port listener from overlap TGT nodes if port changed + # Step 4: remove old SRC-port listener from overlap TGT nodes if port changed for tgt in tgt_paths: if tgt['node_id'] in overlap_ids: old_port = src_port_by_id.get(tgt['node_id']) @@ -560,6 +565,22 @@ def _handle_intermediate_barrier(group, member_migrations, src_node, tgt_node, s logger.debug(f"intermediates barrier: waiting for {len(waiting)} workers") return None, None # None = still waiting + # Every member finished this round. If any of them still has too much + # dirty delta to freeze quickly at cutover, start another synchronized + # round -- every member retakes a snapshot together, even ones whose own + # delta was already low -- up to the round cap. + if (group.intermediate_more_needed + and group.intermediate_round + 1 < constants.LVOL_MIG_MAX_INTERMEDIATE_SNAPS): + group.intermediate_round += 1 + group.intermediates_done = [] + group.intermediate_more_needed = [] + group.write_to_db(db.kv_store) + logger.info( + f"Group {group.uuid[:8]}: dirty delta still high after round " + f"{group.intermediate_round}/{constants.LVOL_MIG_MAX_INTERMEDIATE_SNAPS}; " + f"starting another synchronized intermediate round") + return None, None # None = still waiting -- workers will redo this round + logger.info( f"Group {group.uuid[:8]}: all workers reached intermediates_done; " f"calling bdev_lvol_batch_final_step") @@ -574,25 +595,16 @@ def _handle_intermediate_barrier(group, member_migrations, src_node, tgt_node, s lvol_names, lvol_ids, snapshot_names = _build_batch_final_args( group, member_migrations, src_node, tgt_node, tgt_rpc) except (ValueError, KeyError) as e: - try: - src_rpc.bdev_nvme_detach_controller(ctrl_name) - except Exception as detach_exc: - logger.warning( - f"Group {group.uuid[:8]}: hub controller detach after build-args " - f"failure (non-fatal): {detach_exc}") + # Hub controller left attached — hub_manager owns its lifecycle + # entirely via its own idle timeout. return None, str(e) - # Pre-freeze: take SRC secondary/tertiary out of the read path before the - # synchronous final-step transfer below. bdev_lvol_batch_transfer_final_step - # freezes the SRC primary internally for the duration of the transfer, but - # a client sitting on a SRC replica path is not covered by that freeze — - # without this, a write accepted by a SRC replica during the transfer (or - # in the gap before cutover flips SRC paths inaccessible) never reaches - # the copy that already ran, and is silently lost. Mirrors the single-lvol - # path's identical pre-freeze (tasks_runner_lvol_migration.py). + # Pre-freeze: take SRC/TGT paths out of the read/write path before the + # 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, _, _ = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) - src_replica_paths = src_paths[1:] # secondary/tertiary only; primary is frozen internally by the RPC below + 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 def _flip(rpc, ip, port, trtype, state, label): try: @@ -608,21 +620,37 @@ def _flip_all(rpc, ips, port, trtype, state, label): _flip(rpc, _ip, port, trtype, state, label) def _revert_src_replicas(reason): - # Final step didn't complete — put SRC secondary/tertiary back into - # the read path (their pre-freeze state) so clients keep multipath - # access to the still-live source instead of being stuck on primary only. - if not src_replica_paths: - return - logger.warning(f"Group {group.uuid[:8]}: {reason}; reverting SRC secondary/tertiary to non_optimized") + # Final step didn't complete — put every SRC path back into the + # read/write path (their pre-freeze state) so clients keep access to + # the still-live source instead of being stuck with nothing reachable. + # Primary -> optimized (it was driven inaccessible pre-final-step by + # the diagnostic widened freeze above); secondary/tertiary -> non_optimized. + logger.warning(f"Group {group.uuid[:8]}: {reason}; reverting SRC paths " + f"(primary optimized, replicas non_optimized)") + primary_src = src_paths[0] + _flip_all(primary_src['rpc'], primary_src['ips'], primary_src['port'], + primary_src['trtype'], "optimized", f"SRC-{primary_src['node_id'][:8]}(revert)") for p in src_replica_paths: _flip_all(p['rpc'], p['ips'], p['port'], p['trtype'], "non_optimized", f"SRC-{p['node_id'][:8]}(revert)") - if src_replica_paths: - logger.info(f"Group {group.uuid[:8]}: setting SRC secondary/tertiary inaccessible pre-final-step") - for p in src_replica_paths: - _flip_all(p['rpc'], p['ips'], p['port'], p['trtype'], - "inaccessible", f"SRC-{p['node_id'][:8]}(pre-freeze)") + # TEMPORARILY CHANGED for a diagnostic test: instead of only freezing SRC + # secondary/tertiary pre-final-step (primary relied on the RPC's own + # internal freeze), make EVERY path -- all SRC (primary included) and all + # TGT -- inaccessible up front, wait 2s so any in-flight client I/O has + # time to fully settle/drain before the data actually moves, THEN call + # final_step. Re-enable the narrower pre-freeze once this test is done. + logger.info(f"Group {group.uuid[:8]}: setting ALL SRC and TGT paths inaccessible " + f"pre-final-step (diagnostic)") + for p in src_paths: + _flip_all(p['rpc'], p['ips'], p['port'], p['trtype'], + "inaccessible", f"SRC-{p['node_id'][:8]}(pre-freeze)") + for p in tgt_paths: + _flip_all(p['rpc'], p['ips'], p['port'], p['trtype'], + "inaccessible", f"TGT-{p['node_id'][:8]}(pre-freeze)") + logger.info(f"Group {group.uuid[:8]}: sleeping 2s after all-paths-inaccessible " + f"before batch_final_step (diagnostic)") + time.sleep(2) logger.info( f"Group {group.uuid[:8]}: batch_final_step " @@ -630,8 +658,12 @@ def _revert_src_replicas(reason): batch_ok = False batch_err = None try: - ret = src_rpc.bdev_lvol_batch_transfer_final_step( - lvol_names, lvol_ids, snapshot_names, 2, hub_bdev, "migrate") + # This call moves real data and can legitimately run longer than the + # 5s blanket timeout _make_rpc()/src_rpc uses for every other RPC in + # this file -- use a dedicated, longer-timeout client just for it. + final_step_rpc = src_node.rpc_client(timeout=15, retry=2) + ret = final_step_rpc.bdev_lvol_batch_transfer_final_step( + lvol_names, lvol_ids, snapshot_names, 16, hub_bdev, "migrate") logger.info(f"Group {group.uuid[:8]}: bdev_lvol_batch_transfer_final_step returned {ret!r}") batch_ok = True except RPCRemoteError as e: @@ -646,8 +678,59 @@ def _revert_src_replicas(reason): if not batch_ok: _revert_src_replicas("batch_final_step failed") - # else: left as-is — the Done handler's ANA sequence (_flip_ana_to_optimized) - # already drives every SRC path (including primary) to inaccessible on success. + # The revert above reopened SRC to live client I/O. Retrying with the + # snapshots taken before this reopen would silently miss whatever the + # client writes in the meantime -- force every member through one + # more synchronized intermediate round first, same mechanism as the + # dirty-delta trigger above, so the retry's snapshots actually cover + # the reopen window. Falls through to the normal suspend/retry-budget + # path once the round cap is hit, so a persistently failing group + # still eventually resolves instead of looping forever. + if group.intermediate_round + 1 < constants.LVOL_MIG_MAX_INTERMEDIATE_SNAPS: + group.intermediate_round += 1 + group.intermediates_done = [] + group.intermediate_more_needed = [] + group.write_to_db(db.kv_store) + + # bdev_lvol_set_migration_flag drives the distrib-level special_io + # machinery for the target bdev (see snapshot_replication.py's + # comment on the same flag); it's only ever set once, at initial + # target-bdev creation (migration_controller.create_migration). + # A failed/aborted final_step attempt may clear it on the target, + # so re-assert it on every member's target bdev before retrying — + # otherwise the retry's cutover could run without the target + # being treated as migration-aware. + tgt_sec_node, _ = _get_target_secondary_node(tgt_node, src_node.get_id()) + tgt_ter_node, _ = _get_target_tertiary_node(tgt_node, src_node.get_id()) + tgt_sec_rpc_reflag = _make_rpc(tgt_sec_node) if tgt_sec_node else None + tgt_ter_rpc_reflag = _make_rpc(tgt_ter_node) if tgt_ter_node else None + for m in member_migrations: + try: + m_lvol = db.get_lvol_by_id(m.lvol_id) + m_tgt_composite = f"{tgt_node.lvstore}/{_lvol_tgt_bdev_name(m_lvol.lvol_bdev)}" + except KeyError: + continue + if not tgt_rpc.bdev_lvol_set_migration_flag(m_tgt_composite): + logger.warning( + f"Group {group.uuid[:8]}: re-assert migration flag on primary " + f"failed for {m_tgt_composite} (may already be flagged)") + for _extra_rpc in (tgt_sec_rpc_reflag, tgt_ter_rpc_reflag): + if _extra_rpc: + try: + _extra_rpc.bdev_lvol_set_migration_flag(m_tgt_composite) + except Exception as e: + logger.warning( + f"Group {group.uuid[:8]}: re-assert migration flag on " + f"replica failed for {m_tgt_composite} (non-fatal): {e}") + + logger.warning( + f"Group {group.uuid[:8]}: batch_final_step failed; forcing another " + f"synchronized intermediate round {group.intermediate_round}/" + f"{constants.LVOL_MIG_MAX_INTERMEDIATE_SNAPS} before retrying") + return None, None # None = still waiting -- workers will redo this round + # else: left as-is — all SRC/TGT paths were already driven inaccessible + # before final_step (diagnostic, see above); only TGT primary needs to + # come back optimized on success, handled below. if batch_ok: # bdev_lvol_batch_final_step handles add_clone on the primary internally. @@ -658,7 +741,8 @@ def _revert_src_replicas(reason): if sec_node or ter_node: sec_rpc_extra = _make_rpc(sec_node) if sec_node else None ter_rpc_extra = _make_rpc(ter_node) if ter_node else None - snap_by_migration_id = dict(zip(group.ordered_migration_ids(), snapshot_names)) + _reordered_ids = group.ordered_migration_ids() + snap_by_migration_id = dict(zip(_reordered_ids, snapshot_names)) for m in member_migrations: snap_composite = snap_by_migration_id.get(m.uuid, "") if not snap_composite: @@ -687,11 +771,10 @@ def _revert_src_replicas(reason): _flip_ana_to_optimized(group, member_migrations, src_node, src_rpc, tgt_node, tgt_rpc) - try: - src_rpc.bdev_nvme_detach_controller(ctrl_name) - except Exception as e: - logger.warning(f"Group {group.uuid[:8]}: hub detach (non-fatal): {e}") - + # Hub controller left attached on both success and failure — hub_manager + # owns its lifecycle entirely via its own idle timeout. Detaching it here + # unconditionally, on every group's final step, defeated the whole point + # of keeping it warm for the next group to reuse. return batch_ok, batch_err diff --git a/simplyblock_core/services/tasks_runner_lvol_migration.py b/simplyblock_core/services/tasks_runner_lvol_migration.py index e5c2a1d68..171f2330c 100644 --- a/simplyblock_core/services/tasks_runner_lvol_migration.py +++ b/simplyblock_core/services/tasks_runner_lvol_migration.py @@ -865,12 +865,12 @@ def _cleanup_final_migration(src_rpc, ctx, tgt_rpc=None, rollback_target=False, tgt_all_nodes=None, tgt_lvs_name=None): """Clean up after a final lvol migration attempt. - On the success path (rollback_target=False) the hub controller is kept - attached on source — detaching it would drop the migration path before - clients have switched to the new target path. + The hub controller is never touched here on either path — it is owned + and lifecycle-managed entirely by hub_manager's own activity-based idle + timeout, not by this function. - On the rollback path (rollback_target=True) the hub controller IS detached - and the target lvol/subsystem are torn down so a retry starts clean. + On the rollback path (rollback_target=True) the target lvol/subsystem + are torn down so a retry starts clean. ``nqn``/``lvol_uuid``/``subsystem_created_on_target`` must come from the caller (the lvol record and migration.target_subsystem_node_ids) — @@ -878,13 +878,12 @@ def _cleanup_final_migration(src_rpc, ctx, tgt_rpc=None, rollback_target=False, stage, so reading them from ``ctx`` here silently no-ops the subsystem cleanup entirely. """ - ctrl_name = ctx.get('ctrl_name') - if ctrl_name and rollback_target: - try: - src_rpc.bdev_nvme_detach_controller(ctrl_name) - except Exception as e: - logger.warning(f"detach hub ctrl {ctrl_name}: {e}") - + # The hub controller is intentionally left attached here, even on + # rollback: it's managed entirely by hub_manager's own activity-based + # idle timeout (IDLE_TIMEOUT with no acquire()s). A retry of this same + # migration will just reuse it via acquire() instead of paying the + # reattach + DETACH_COOLDOWN cost, and a sibling migration to the same + # target isn't disrupted. if rollback_target and tgt_rpc: tgt_composite = ctx.get('tgt_lvol_composite') _nqn = ctx.get('nqn') or nqn @@ -1953,7 +1952,8 @@ def _revert_src_replicas(reason): try: last_snap = db.get_snapshot_by_id(last_snap_uuid) except KeyError: - src_rpc.bdev_nvme_detach_controller(ctrl_name) + # Hub controller left attached — hub_manager owns its + # lifecycle entirely via its own idle timeout. try: _delete_bdev_blocking(tgt_lvol_composite, tgt_rpc, secondary_rpc=tgt_sec_rpc, tertiary_rpc=tgt_ter_rpc, @@ -1979,7 +1979,7 @@ def _revert_src_replicas(reason): tgt_snap_composite = snap_bdev break if not tgt_snap_composite: - src_rpc.bdev_nvme_detach_controller(ctrl_name) + # Hub controller left attached — see comment above. try: _delete_bdev_blocking(tgt_lvol_composite, tgt_rpc, secondary_rpc=tgt_sec_rpc, tertiary_rpc=tgt_ter_rpc, @@ -2041,7 +2041,8 @@ def _revert_src_replicas(reason): if not ret: if state not in ('Done', 'No process'): _revert_src_replicas("final migration failed") - src_rpc.bdev_nvme_detach_controller(ctrl_name) + # Hub controller left attached — see comment above; this is a + # retryable suspend, not an abandoned migration. # Do NOT delete the target bdev on transfer failure — the bdev is # still valid and retaining it keeps the map_id stable across retries. # Deleting it would force a recreate at a higher map_id (due to @@ -2541,7 +2542,7 @@ 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"{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, @@ -2554,19 +2555,29 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): logger.warning(f"Source snapshot {snap_uuid} not found in DB; skipping") # --- Source NVMe-oF subsystem teardown (best-effort) --- + # Batch group workers share ONE subsystem across every member -- deleting + # it here, per worker, would mean up to N-1 redundant attempts before the + # group orchestrator's own _delete_source_subsystem() runs once, after + # the barrier, on the master thread. Skip it here for group workers; + # single-lvol migrations (no group) still own their own subsystem and + # must still delete it themselves. lvol = None try: lvol = db.get_lvol_by_id(migration.lvol_id) - 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) - for _sp in _src_paths_cu: - if _sp['node_id'] in _overlap_ids_cu: - logger.info( - f"Step 8: skip subsystem delete on overlap node " - f"{_sp['node_id'][:8]} (now serving TGT)") - else: - migration_controller.cleanup_subsystem_or_ns(lvol.nqn, lvol.uuid, True, _sp['rpc']) + if migration.migration_group_id: + logger.info(f"Step 8: source subsystem delete deferred to group " + f"orchestrator (worker of group {migration.migration_group_id[:8]})") + 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) + for _sp in _src_paths_cu: + if _sp['node_id'] in _overlap_ids_cu: + logger.info( + f"Step 8: skip subsystem delete on overlap node " + f"{_sp['node_id'][:8]} (now serving TGT)") + else: + migration_controller.cleanup_subsystem_or_ns(lvol.nqn, lvol.uuid, True, _sp['rpc']) except Exception as e: logger.warning(f"Source subsystem cleanup failed: {e}") @@ -2621,7 +2632,7 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): return True, False, None -def _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=None): +def _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=None, src_node=None): """ Roll back a failed or cancelled migration: remove any partially-created target lvol/subsystem, then delete all snapshots copied to the target. @@ -2630,12 +2641,19 @@ def _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=None): on primary and secondary). Idempotent: "not found" (status 2) is treated as already done, so a crash-recovery re-run is safe. + Overlap safety: a target node that is also one of this lvol's SOURCE + replica paths (shared/overlap topology) still has its namespace pointing + at the SRC bdev pre-cutover -- it is the live path a client is currently + using, not a spare "target" namespace. Rollback must never touch the + subsystem/namespace on such a node; only non-overlap target-only nodes + are safe to tear down. + Returns (done: bool, suspend: bool, error: str|None). """ - # Immediately detach the hub controller on failure/cancel — don't leave it - # connected to a target whose snapshots we're about to roll back. - hub_manager.detach_now(migration.source_node_id, tgt_node.get_id(), src_rpc=src_rpc) + # Hub controller left attached here too — hub_manager owns its lifecycle + # entirely via its own idle timeout now; nothing in the migration runners + # calls detach_now() any more. ctx = migration.transfer_context or {} tgt_sec, _ = _get_target_secondary_node(tgt_node, migration.source_node_id) @@ -2643,6 +2661,15 @@ def _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=None): tgt_ter, _ = _get_target_tertiary_node(tgt_node, migration.source_node_id) tgt_ter_rpc = _make_rpc(tgt_ter) if tgt_ter else None + overlap_ids = set() + if src_node is not None: + try: + _, _, overlap_ids = _build_paths(src_node, tgt_node, src_rpc, tgt_rpc) + except Exception as e: + logger.warning( + f"cleanup_target: could not compute overlap nodes, treating " + f"none as overlap (safer would be all -- proceeding with caution): {e}") + # --- Step 0: delete dangling target lvol/subsystems from a failed LVOL_MIGRATE --- # Also handles the pre-create case where bdev/subsystems were set up by # create_migration() but migration was cancelled before LVOL_MIGRATE completed. @@ -2672,17 +2699,27 @@ def _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=None): # Clean up NVMe-oF subsystem — from ctx (LVOL_MIGRATE failure) or from pre-create. _nqn_to_clean = nqn or _pre_nqn if _nqn_to_clean: - try: - migration_controller.cleanup_subsystem_or_ns( - _nqn_to_clean, migration.lvol_id, - tgt_node.get_id() in owned_node_ids, tgt_rpc) - except Exception as e: - logger.warning(f"cleanup target subsystem {_nqn_to_clean}: {e}") + if tgt_node.get_id() in overlap_ids: + logger.info( + f"cleanup_target: skip subsystem/ns teardown on overlap " + f"node {tgt_node.get_id()[:8]} (still serving live SRC path)") + else: + try: + migration_controller.cleanup_subsystem_or_ns( + _nqn_to_clean, migration.lvol_id, + tgt_node.get_id() in owned_node_ids, tgt_rpc) + except Exception as e: + logger.warning(f"cleanup target subsystem {_nqn_to_clean}: {e}") for _label, _extra_node, _extra_rpc in [ ("secondary", tgt_sec, tgt_sec_rpc), ("tertiary", tgt_ter, tgt_ter_rpc), ]: if _extra_rpc and _extra_node: + if _extra_node.get_id() in overlap_ids: + logger.info( + f"cleanup_target: skip {_label} subsystem/ns teardown on " + f"overlap node {_extra_node.get_id()[:8]} (still serving live SRC path)") + continue try: migration_controller.cleanup_subsystem_or_ns( _nqn_to_clean, migration.lvol_id, @@ -2989,7 +3026,7 @@ def task_runner(task): next_phase = LVolMigration.PHASE_COMPLETED elif phase == LVolMigration.PHASE_CLEANUP_TARGET: - done, suspend, error = _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=src_rpc) + done, suspend, error = _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=src_rpc, src_node=src_node) next_phase = "" # terminal — done-handler always sets STATUS_FAILED/CANCELLED else: @@ -3133,7 +3170,10 @@ def _post_process_snap_group(snap, migration): if snap_uuid not in migration.snaps_transferred_group: migration.snaps_transferred_group.append(snap_uuid) migration_events.migration_snap_copied(migration, snap_uuid) - logger.info(f"Group worker: snap {snap_uuid} raw-transferred (pending tree reconstruction)") + logger.info( + f"Group worker {migration.uuid[:8]}: DIAG snap {snap_uuid[:8]} raw-transferred " + f"(pending tree reconstruction), lvol={migration.lvol_id[:8] if migration.lvol_id else None}, " + f"snaps_transferred_group now={list(migration.snaps_transferred_group)}") return True, None @@ -3283,22 +3323,34 @@ 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): +def _handle_group_intermediate(migration, src_node, tgt_node, src_rpc, tgt_rpc, target_round=0): """ INTERMEDIATE phase for a group worker. - Takes exactly one intermediate ("shrink") snapshot and transfers it to the - target, skipping add_clone/convert (same as snap_copy). After this the - worker signals intermediates_done to the group and waits for batch_result. + Takes one intermediate ("shrink") snapshot per round and transfers it to + the target, skipping add_clone/convert (same as snap_copy). After this + the worker signals intermediates_done to the group and waits for either + another round or batch_result. + + ``target_round`` is the group's current intermediate_round. If this + worker has already completed that round (migration.intermediate_snap_rounds + > target_round), it's done for now. Otherwise -- including when it was + previously done for an earlier round but the group has since asked for + another synchronized round -- it resets and takes a fresh snapshot. Returns (done: bool, suspend: bool, error: str|None). """ trtype, _ = _get_migration_nic(tgt_node) ctx = migration.transfer_context or {} - # If we already took and transferred the intermediate snap, we're done. + # 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 + # for another round since we last finished -- fall through and redo. if ctx.get('stage') == 'intermediate_done': - return True, False, None + if migration.intermediate_snap_rounds > target_round: + return True, False, None + ctx = {} + migration.transfer_context = {} # Take the intermediate snapshot if not already in flight. if ctx.get('stage') != 'intermediate_transfer': @@ -3463,9 +3515,10 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src Manages the group worker state machine: SNAP_COPY → transfer owned snaps (no add_clone/convert) → signal snap_copy_done to group → wait for INTERMEDIATE - LVOL_MIGRATE (repurposed as the single-intermediate phase for workers) - → take + transfer 1 intermediate snap - → signal intermediates_done → wait for batch_result + LVOL_MIGRATE (repurposed as the intermediate phase for workers) + → take + transfer 1 intermediate snap for the current round + → signal intermediates_done → wait for either another + synchronized round or batch_result CLEANUP_SOURCE → normal source cleanup + signal cleanup_source_done CLEANUP_TARGET → normal target rollback @@ -3483,8 +3536,14 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src if phase == LVolMigration.PHASE_SNAP_COPY: if migration_id not in group.snap_copy_done: # Still transferring owned snaps. - done, suspend, error = _handle_group_snap_copy( - migration, src_node, tgt_node, src_rpc, tgt_rpc) + try: + done, suspend, error = _handle_group_snap_copy( + migration, src_node, tgt_node, src_rpc, tgt_rpc) + except RPCException as exc: + # Charge this worker's own retry budget and report failure to + # the group -- never decide/roll back unilaterally (see + # _group_worker_budget_suspend's docstring). + return _group_worker_budget_suspend(task, migration, group_id, str(exc)) if error: return _group_worker_budget_suspend(task, migration, group_id, error) if suspend: @@ -3523,11 +3582,29 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src task.write_to_db(db.kv_store) return False - # --- LVOL_MIGRATE (group worker: take 1 intermediate + wait for batch_result) --- + # --- LVOL_MIGRATE (group worker: take intermediate round(s) + wait for batch_result) --- if phase == LVolMigration.PHASE_LVOL_MIGRATE: if migration_id not in group.intermediates_done: - done, suspend, error = _handle_group_intermediate( - migration, src_node, tgt_node, src_rpc, tgt_rpc) + # A sibling may have already failed and told the group to roll + # back while we were mid-retry ourselves -- notice it immediately + # instead of continuing to loop until our own budget runs out. + group = db.get_migration_group_by_id(group_id) + 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) + + try: + done, suspend, error = _handle_group_intermediate( + migration, src_node, tgt_node, src_rpc, tgt_rpc, + target_round=group.intermediate_round) + except RPCException as exc: + # Charge this worker's own retry budget and report failure to + # the group -- never decide/roll back unilaterally (see + # _group_worker_budget_suspend's docstring). + return _group_worker_budget_suspend(task, migration, group_id, str(exc)) if error: return _group_worker_budget_suspend(task, migration, group_id, error) if suspend: @@ -3535,11 +3612,33 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src if done: group = db.get_migration_group_by_id(group_id) if migration_id not in group.intermediates_done: + # Below the round cap, check whether this worker's dirty + # delta is still too large to freeze quickly at cutover -- + # if so, flag it so the orchestrator starts another + # synchronized round for every member (see + # LVolMigrationGroup's INTERMEDIATE docstring). + needs_more = False + if group.intermediate_round + 1 < constants.LVOL_MIG_MAX_INTERMEDIATE_SNAPS: + try: + lvol = db.get_lvol_by_id(migration.lvol_id) + src_composite = f"{src_node.lvstore}/{lvol.lvol_bdev}" + delta = _get_lvol_delta_bytes(src_rpc, src_composite) + needs_more = ( + delta is None + or delta > constants.LVOL_MIG_INTERMEDIATE_SNAP_THRESHOLD_BYTES) + except Exception as e: + logger.warning( + f"Group worker {migration_id[:8]}: delta check failed " + f"(assuming another round is needed): {e}") + needs_more = True + if needs_more and migration_id not in group.intermediate_more_needed: + group.intermediate_more_needed.append(migration_id) group.intermediates_done.append(migration_id) group.write_to_db(db.kv_store) logger.info( f"Group worker {migration_id[:8]}: signalled intermediates_done " - f"({len(group.intermediates_done)}/{group.member_count()})") + f"({len(group.intermediates_done)}/{group.member_count()})" + + (" [delta still high, requesting another round]" if needs_more else "")) migration.write_to_db(db.kv_store) task.write_to_db(db.kv_store) return False @@ -3606,7 +3705,7 @@ def _group_worker_phase_dispatch(task, migration, phase, src_node, tgt_node, src if phase == LVolMigration.PHASE_CLEANUP_TARGET: try: done, suspend, error = _handle_cleanup_target( - migration, tgt_node, tgt_rpc, src_rpc=src_rpc) + migration, tgt_node, tgt_rpc, src_rpc=src_rpc, src_node=src_node) except RPCException as exc: return _suspend_task(task, migration, str(exc))