diff --git a/simplyblock_core/constants.py b/simplyblock_core/constants.py index 8a6315b65..de09667aa 100644 --- a/simplyblock_core/constants.py +++ b/simplyblock_core/constants.py @@ -408,7 +408,7 @@ def get_config_var(name, default=None): # Live volume migration constants LVOL_MIG_MAX_RETRIES = 5 # max retry attempts before aborting -LVOL_MIG_DEADLINE_SEC = 360 # default 4-hour deadline (0 = no deadline) +LVOL_MIG_DEADLINE_SEC = 3600 # 1-hour deadline (0 = no deadline) LVOL_MIG_MAX_INTERMEDIATE_SNAPS = 3 # max recursive "shrink" snapshot rounds LVOL_MIG_INTERMEDIATE_SNAP_THRESHOLD_BYTES = 500 * 1024 * 1024 # 500 MiB — skip if delta is smaller LVOL_MIG_BDEV_SUFFIX = 'm' # appended to every migration bdev on the target to avoid collision with real bdevs diff --git a/simplyblock_core/controllers/migration_controller.py b/simplyblock_core/controllers/migration_controller.py index e28e258e4..e65e68448 100644 --- a/simplyblock_core/controllers/migration_controller.py +++ b/simplyblock_core/controllers/migration_controller.py @@ -44,7 +44,7 @@ from datetime import datetime from simplyblock_core import constants -from simplyblock_core.controllers import migration_events, tasks_controller, snapshot_controller +from simplyblock_core.controllers import migration_events, tasks_controller from simplyblock_core.exceptions import MigrationConflictError, PreconditionError from simplyblock_core.controllers.host_auth import _reapply_allowed_hosts from simplyblock_core.kms import create_kms_connection, lvol_dek_path, pool_kek_name @@ -54,7 +54,7 @@ from simplyblock_core.models.nvme_connect import NvmeConnectEntry from simplyblock_core.models.snapshot import SnapShot from simplyblock_core.models.storage_node import StorageNode -from simplyblock_core.utils import convert_size +from simplyblock_core.utils import convert_size, lvol_tgt_bdev_name # Note: JobSchedule is not imported directly here; task creation is delegated to # tasks_controller.add_lvol_mig_task() which handles event logging consistently. @@ -121,12 +121,6 @@ def start_migration(migration_id, raise ValueError(f"Target node is not online (status={target_node.status})") snap_plan = get_snapshot_chain(lvol_id, source_node_id) - if not snap_plan: - snap_name = f"_mig_{migration.uuid[:8]}_lvol_{migration.lvol_id[:8]}" - snap_uuid, err = snapshot_controller.add(lvol_id, snap_name, bypass_migration_check=True) - if err: - raise ValueError(f"Failed to create snapshot: {err}") - snap_plan = [snap_uuid] snaps_found_on_target = [s for s in snap_plan if _is_snap_on_node(s, target_node_id)] snap_migration_plan = [s for s in snap_plan if s not in snaps_found_on_target] @@ -222,7 +216,7 @@ def _cleanup_created(migration): return nqn = lvol.nqn - bdev_short = lvol.lvol_bdev + _MIGRATION_BDEV_SUFFIX + bdev_short = lvol_tgt_bdev_name(lvol.lvol_bdev) composite = f"{tgt_node.lvstore}/{bdev_short}" # Compute overlap: nodes shared between SRC and TGT paths. @@ -423,7 +417,8 @@ def _is_snap_on_node(snap_id, node_id): snap = db.get_snapshot_by_id(snap_id) except KeyError: return False - if snap.lvol.node_id == node_id: + # snap.lvol defaults to None (SnapShot.lvol) — guard before dereferencing. + if snap.lvol and snap.lvol.node_id == node_id: return True return any( inst.get('lvol', {}).get('node_id') == node_id @@ -433,10 +428,18 @@ def _is_snap_on_node(snap_id, node_id): def _get_snap_ancestry(snap_uuid): """ - Walk the ``snap_ref_id`` chain from *snap_uuid* upward to the root and + Walk the blobstore parent chain from *snap_uuid* upward to the root and return the UUIDs in root-first order (oldest ancestor first). - ``snap_ref_id`` points from a child snapshot to its parent snapshot. + Two parent pointers are tried in order: + - ``snap_ref_id``: set for snapshots taken from a clone volume (points to + the clone's parent snapshot, used for ref-count tracking). + - ``prev_snap_uuid``: set for all snapshots and records the previous + snapshot in the lvol's own sequence — this is the blobstore parent for + snapshots taken from a non-clone volume (e.g. snap_A3 → snap_A2 → + snap_A1 taken from the original lvol_A). Without this fallback, the + ancestry walk stops at the clone's immediate parent and misses the older + ancestor snapshots whose data is not copied standalone. """ chain = [] current = snap_uuid @@ -448,7 +451,7 @@ def _get_snap_ancestry(snap_uuid): except KeyError: break chain.append(current) - current = snap.snap_ref_id + current = snap.snap_ref_id or snap.prev_snap_uuid chain.reverse() # oldest → newest return chain @@ -495,7 +498,7 @@ def get_snaps_safe_to_delete_on_source(migration): if lvol.uuid == migration.lvol_id: continue if lvol.cloned_from_snap and lvol.cloned_from_snap in candidates: - _protect_snap_and_ancestors(lvol.cloned_from_snap, candidates) + candidates -= _collect_snap_ancestry(lvol.cloned_from_snap) return candidates @@ -525,7 +528,7 @@ def get_snaps_to_delete_on_target(migration): if lvol.uuid == migration.lvol_id: continue if lvol.cloned_from_snap: - _collect_snap_ancestry(lvol.cloned_from_snap, protected) + protected |= _collect_snap_ancestry(lvol.cloned_from_snap) return [ uid for uid in migration.snaps_migrated @@ -533,40 +536,24 @@ def get_snaps_to_delete_on_target(migration): ] -def _protect_snap_and_ancestors(snap_uuid, candidate_set): - """Remove *snap_uuid* and all its ancestors from *candidate_set*.""" +def _collect_snap_ancestry(snap_uuid) -> set: + """Return the set of *snap_uuid* and all its ancestors.""" + result: set = set() current = snap_uuid - visited = set() - while current and current not in visited: - visited.add(current) - candidate_set.discard(current) - try: - snap = db.get_snapshot_by_id(current) - current = snap.snap_ref_id - except KeyError: - break - - -def _collect_snap_ancestry(snap_uuid, out_set): - """Add *snap_uuid* and all its ancestors to *out_set*.""" - current = snap_uuid - visited = set() - while current and current not in visited: - visited.add(current) - out_set.add(current) + while current and current not in result: + result.add(current) try: snap = db.get_snapshot_by_id(current) - current = snap.snap_ref_id + current = snap.snap_ref_id or snap.prev_snap_uuid except KeyError: break + return result # --------------------------------------------------------------------------- # Post-migration DB updates # --------------------------------------------------------------------------- -_MIGRATION_BDEV_SUFFIX = constants.LVOL_MIG_BDEV_SUFFIX - def _build_connect_entries(node, port, lvol, nqn, ctrl_loss_tmo, cluster, host_entry, host_nqn): """Build NVMe connect-string entries for each data NIC on *node* at *port*.""" @@ -675,7 +662,7 @@ def create_migration(lvol_id, target_node_id, cluster = db.get_cluster_by_id(tgt_node.cluster_id) tgt_rpc = tgt_node.rpc_client() nqn = lvol.nqn - bdev_short = lvol.lvol_bdev + _MIGRATION_BDEV_SUFFIX + bdev_short = lvol_tgt_bdev_name(lvol.lvol_bdev) composite = f"{tgt_node.lvstore}/{bdev_short}" size_in_mib = convert_size(lvol.size, 'MiB') tgt_port = tgt_node.get_lvol_subsys_port(tgt_node.lvstore) @@ -738,7 +725,7 @@ def create_migration(lvol_id, target_node_id, f"create_migration: secondary registration error (continuing): {_e}") _pre_ter_node = None - if lvol.ha_type == "ha3" and tgt_node.tertiary_node_id: + if tgt_node.tertiary_node_id: try: _pre_ter_node = db.get_storage_node_by_id(tgt_node.tertiary_node_id) _ter_rpc_reg = _pre_ter_node.rpc_client() diff --git a/simplyblock_core/models/lvol_migration.py b/simplyblock_core/models/lvol_migration.py index 367a5654c..2b52673b4 100644 --- a/simplyblock_core/models/lvol_migration.py +++ b/simplyblock_core/models/lvol_migration.py @@ -30,10 +30,6 @@ class LVolMigration(BaseModel): STATUS_NEW = 'new' STATUS_RUNNING = 'running' STATUS_SUSPENDED = 'suspended' - # CUTOVER: the target subsystem is live and the client must reconnect before - # the source subsystem is removed. Set by PHASE_LVOL_MIGRATE (tgt_is_src_secondary - # path only); cleared to STATUS_DONE once PHASE_CLEANUP_SOURCE completes. - STATUS_CUTOVER = 'cutover' STATUS_DONE = 'done' STATUS_FAILED = 'failed' STATUS_CANCELLED = 'cancelled' @@ -49,7 +45,6 @@ class LVolMigration(BaseModel): STATUS_NEW: 0, STATUS_RUNNING: 1, STATUS_SUSPENDED: 2, - STATUS_CUTOVER: 6, STATUS_DONE: 3, STATUS_FAILED: 4, STATUS_CANCELLED: 5, @@ -124,7 +119,7 @@ def write_to_db(self, kv_store=None): def is_active(self): return self.status in ( self.STATUS_NEW, self.STATUS_RUNNING, - self.STATUS_SUSPENDED, self.STATUS_CUTOVER, + self.STATUS_SUSPENDED, ) def has_deadline_passed(self): diff --git a/simplyblock_core/rpc_client.py b/simplyblock_core/rpc_client.py index 3448a0ce0..b7afa1ae4 100755 --- a/simplyblock_core/rpc_client.py +++ b/simplyblock_core/rpc_client.py @@ -1286,6 +1286,9 @@ def bdev_lvol_get_lvstores(self, name): params = {"lvs_name": name} return self._request("bdev_lvol_get_lvstores", params) + def bdev_lvol_rename(self, old_name, new_name): + return self._request3("bdev_lvol_rename", old_name=old_name, new_name=new_name) + def bdev_lvol_resize(self, name, size_in_mib): params = { "name": name, diff --git a/simplyblock_core/services/hub_controller_manager.py b/simplyblock_core/services/hub_controller_manager.py new file mode 100644 index 000000000..83d8862c2 --- /dev/null +++ b/simplyblock_core/services/hub_controller_manager.py @@ -0,0 +1,326 @@ +# coding=utf-8 +""" +hub_controller_manager.py — lifecycle manager for migration hub NVMe-oF controllers. + +Goal +---- +A migration hub controller (mighub*n1 bdev on the source SPDK) connects source to +target's dedicated transfer-hub subsystem. Two problems arise when controllers are +detached naively: + + 1. Concurrent migrations: migration A finishes and detaches the controller while + migration B (to the same target) is still running and still needs it. + + 2. Deleting-state race: bdev_nvme_detach_controller triggers an async NVMe + disconnect handshake. If a new attach arrives before the handshake completes + the controller is in 'deleting' state and the attach fails. + +This module solves both: + + • Activity-based lifetime — the controller stays alive as long as any migration + calls acquire() within IDLE_TIMEOUT seconds. No reference counting needed; + each acquire() simply refreshes the last-used timestamp. + + • Cooldown enforcement — a minimum DETACH_COOLDOWN gap is maintained between + the moment a detach RPC is issued and any subsequent re-attach attempt. If + acquire() is called during the cooldown window it returns a transient error so + the task runner retries on the next loop iteration (~3 s) until the gap passes. + +Public API (called by tasks_runner_lvol_migration) +-------------------------------------------------- + hub_manager.acquire(src_node_id, src_rpc, tgt_node, trtype) + → attach (or reuse) the controller, refresh the idle timer + → returns (ctrl_name, hub_bdev, error) + + hub_manager.detach_now(src_node_id, tgt_node_id, src_rpc=None) + → immediate detach (failure / cancel path), records detach timestamp +""" + +import threading +import time +from typing import Optional + +from simplyblock_core import utils + +logger = utils.get_logger(__name__) + + +class _HubEntry: + __slots__ = ('ctrl_name', 'hub_bdev', 'src_rpc', 'tgt_node', 'last_used') + + def __init__(self, ctrl_name: str, hub_bdev: str, src_rpc, tgt_node): + self.ctrl_name = ctrl_name + self.hub_bdev = hub_bdev # e.g. "mighubXXXXXXXXn1" + self.src_rpc = src_rpc # refreshed on every acquire — used by GC + self.tgt_node = tgt_node + self.last_used = time.monotonic() + + +class HubControllerManager: + """ + Thread-safe lifecycle manager for migration hub NVMe-oF controllers. + + One instance (``hub_manager``) is created at module level and shared by + the entire migration task runner process. + """ + + # Minimum seconds between issuing a detach RPC and allowing any re-attach. + # Must be long enough for the NVMe TCP disconnect handshake to complete. + DETACH_COOLDOWN = 10 + + # 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 + + # GC sweep period. + GC_INTERVAL = 30 + + def __init__(self): + self._lock = threading.Lock() + self._entries: dict = {} # (src_node_id, tgt_node_id) → _HubEntry + self._detach_ts: dict = {} # (src_node_id, tgt_node_id) → monotonic ts of last detach + self._gc_thread: Optional[threading.Thread] = None + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def acquire(self, src_node_id: str, src_rpc, tgt_node, trtype: str): + """ + Ensure the hub controller from *src_node_id* to *tgt_node* is attached. + + Refreshes the idle timer so the controller stays alive as long as + migrations are actively using it. + + Returns (ctrl_name, hub_bdev, error). On transient cooldown the error + string starts with "HUB_COOLDOWN:" so callers can log it distinctly. + """ + key = (src_node_id, tgt_node.get_id()) + + with self._lock: + entry = self._entries.get(key) + if entry is not None: + try: + if src_rpc.get_bdevs(entry.hub_bdev): + entry.last_used = time.monotonic() + entry.src_rpc = src_rpc + logger.info( + f"[HubMgr] reusing {entry.ctrl_name} " + f"src={src_node_id[:8]} tgt={tgt_node.get_id()[:8]}" + ) + return entry.ctrl_name, entry.hub_bdev, None + except Exception: + logger.exception( + f"[HubMgr] failed to validate cached hub entry {entry.ctrl_name}; " + "treating as stale and re-attaching" + ) + logger.info(f"[HubMgr] stale entry for {entry.ctrl_name}; re-attaching") + del self._entries[key] + + # Enforce cooldown before any fresh attach + cooldown_remaining = self._cooldown_remaining(key) + if cooldown_remaining > 0: + msg = ( + f"HUB_COOLDOWN: {cooldown_remaining:.1f}s remaining after last detach " + f"(src={src_node_id[:8]} tgt={tgt_node.get_id()[:8]})" + ) + logger.info(f"[HubMgr] {msg}") + return None, None, msg + + # Attach outside the lock so we don't block the GC thread + ctrl_name, hub_bdev, err = self._attach(src_rpc, tgt_node, trtype) + if err: + return None, None, err + + with self._lock: + # Another thread may have attached concurrently — prefer the winner + existing = self._entries.get(key) + if existing is not None: + try: + if src_rpc.get_bdevs(existing.hub_bdev): + existing.last_used = time.monotonic() + existing.src_rpc = src_rpc + try: + src_rpc.bdev_nvme_detach_controller(ctrl_name) + except Exception as e: + logger.warning( + f"[HubMgr] best-effort detach of duplicate controller failed " + f"(ctrl={ctrl_name}, src={src_node_id[:8]} tgt={tgt_node.get_id()[:8]}): {e}" + ) + return existing.ctrl_name, existing.hub_bdev, None + except Exception as exc: + logger.debug( + f"[HubMgr] failed to validate concurrently attached controller " + f"{existing.ctrl_name}: {exc}" + ) + + entry = _HubEntry(ctrl_name, hub_bdev, src_rpc, tgt_node) + self._entries[key] = entry + self._ensure_gc_running() + + logger.info( + f"[HubMgr] attached {ctrl_name} " + f"src={src_node_id[:8]} tgt={tgt_node.get_id()[:8]}" + ) + return ctrl_name, hub_bdev, None + + def detach_now(self, src_node_id: str, tgt_node_id: str, src_rpc=None): + """ + Immediately detach the hub controller (failure / cancel path). + + Records the detach timestamp so subsequent acquire() calls observe + the DETACH_COOLDOWN gap. + """ + key = (src_node_id, tgt_node_id) + with self._lock: + entry = self._entries.pop(key, None) + if entry is None: + return + rpc = src_rpc or entry.src_rpc + self._do_detach(key, entry, rpc, reason="detach_now") + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _cooldown_remaining(self, key) -> float: + """Return seconds remaining in the post-detach cooldown (0 if none).""" + ts = self._detach_ts.get(key, 0) + if ts == 0: + return 0.0 + elapsed = time.monotonic() - ts + return max(0.0, self.DETACH_COOLDOWN - elapsed) + + def _do_detach(self, key, entry: _HubEntry, rpc, reason: str): + """Issue the detach RPC and record the detach timestamp.""" + try: + if rpc.get_bdevs(entry.hub_bdev): + rpc.bdev_nvme_detach_controller(entry.ctrl_name) + logger.info(f"[HubMgr] detached {entry.ctrl_name} ({reason})") + else: + logger.info( + f"[HubMgr] {entry.ctrl_name} bdev already gone ({reason}); skipping detach" + ) + except Exception as e: + logger.warning(f"[HubMgr] detach {entry.ctrl_name} ({reason}) non-fatal: {e}") + finally: + # Record timestamp regardless of whether the RPC succeeded — the + # controller state is unknown so enforce the cooldown either way. + with self._lock: + self._detach_ts[key] = time.monotonic() + + @staticmethod + def _attach(src_rpc, tgt_node, trtype: str): + """ + Attach a hub controller on *src_rpc* to *tgt_node*'s migration hub subsystem. + + Returns (ctrl_name, hub_bdev, error). + """ + if tgt_node.transfer_hublvol is None or not tgt_node.transfer_hublvol.bdev_name: + tgt_node.create_transfer_hublvol() + + # Already attached (crash recovery or concurrent acquire) + if src_rpc.get_bdevs(tgt_node.transfer_hublvol.get_remote_bdev_name()): + return ( + tgt_node.transfer_hublvol.bdev_name, + tgt_node.transfer_hublvol.get_remote_bdev_name(), + None, + ) + + attached = False + for iface in tgt_node.data_nics: + ip = iface.ip4_address + if tgt_node.active_rdma: + if iface.trtype != "RDMA": + continue + nic_trtype = "RDMA" + else: + if iface.trtype != "TCP": + continue + nic_trtype = "TCP" + + ret = src_rpc.bdev_nvme_attach_controller( + tgt_node.transfer_hublvol.bdev_name, + tgt_node.transfer_hublvol.nqn, + ip, + tgt_node.transfer_hublvol.nvmf_port, + nic_trtype, + ) + if not ret: + if src_rpc.bdev_nvme_controller_list(tgt_node.transfer_hublvol.bdev_name): + logger.info("[HubMgr] zombie controller found; detaching and reattaching") + src_rpc.bdev_nvme_detach_controller(tgt_node.transfer_hublvol.bdev_name) + try: + tgt_node.create_transfer_hublvol() + except Exception as e: + logger.warning(f"[HubMgr] hub subsystem re-create (non-fatal): {e}") + ret = src_rpc.bdev_nvme_attach_controller( + tgt_node.transfer_hublvol.bdev_name, + tgt_node.transfer_hublvol.nqn, + ip, + tgt_node.transfer_hublvol.nvmf_port, + nic_trtype, + ) + if not ret: + return None, None, ( + f"Failed to attach migration hub controller to {tgt_node.get_id()}" + ) + attached = True + break + + if not attached: + return None, None, f"No suitable NIC found on target node {tgt_node.get_id()}" + + return ( + tgt_node.transfer_hublvol.bdev_name, + tgt_node.transfer_hublvol.get_remote_bdev_name(), + None, + ) + + def _ensure_gc_running(self): + """Start the GC daemon thread if not already running. Must be called under self._lock.""" + if self._gc_thread is None or not self._gc_thread.is_alive(): + self._gc_thread = threading.Thread( + target=self._gc_loop, name="HubMgrGC", daemon=True + ) + self._gc_thread.start() + + def _gc_loop(self): + logger.info("[HubMgr] GC thread started") + while True: + time.sleep(self.GC_INTERVAL) + now = time.monotonic() + + with self._lock: + to_evict = [ + key + for key, e in self._entries.items() + if (now - e.last_used) >= self.IDLE_TIMEOUT + ] + evicted = [(key, self._entries.pop(key)) for key in to_evict] + + # Prune stale detach timestamps older than DETACH_COOLDOWN. + stale = [k for k, ts in self._detach_ts.items() + if (now - ts) > self.DETACH_COOLDOWN * 2] + for k in stale: + del self._detach_ts[k] + + should_exit = not self._entries + + for key, entry in evicted: + logger.info( + f"[HubMgr] GC evicting {entry.ctrl_name} " + f"(idle {now - entry.last_used:.0f}s >= {self.IDLE_TIMEOUT}s)" + ) + self._do_detach(key, entry, entry.src_rpc, "gc-idle") + + if should_exit: + with self._lock: + self._gc_thread = None + logger.info("[HubMgr] GC thread exiting (no active entries)") + break # restarted on next acquire() + + +# Module-level singleton — imported and used by tasks_runner_lvol_migration +hub_manager = HubControllerManager() diff --git a/simplyblock_core/services/storage_node_monitor.py b/simplyblock_core/services/storage_node_monitor.py index 30e84ad7e..1646d30fb 100644 --- a/simplyblock_core/services/storage_node_monitor.py +++ b/simplyblock_core/services/storage_node_monitor.py @@ -321,7 +321,9 @@ def get_next_cluster_status(cluster_id): else: node_offline_devices += 1 - if node_offline_devices > 0 or (node_online_devices == 0 and node.status != StorageNode.STATUS_REMOVED): + if (node_offline_devices > 0 + or (node_online_devices == 0 and node.status != StorageNode.STATUS_REMOVED) + or node.status == StorageNode.STATUS_OFFLINE): affected_nodes += 1 if node.mgmt_ip not in affected_physical_nodes: affected_physical_nodes.append(node.mgmt_ip) diff --git a/simplyblock_core/services/tasks_runner_lvol_migration.py b/simplyblock_core/services/tasks_runner_lvol_migration.py index 1368787af..18cfb3cbd 100644 --- a/simplyblock_core/services/tasks_runner_lvol_migration.py +++ b/simplyblock_core/services/tasks_runner_lvol_migration.py @@ -97,6 +97,7 @@ from simplyblock_core.models.storage_node import StorageNode from simplyblock_core.models.snapshot import SnapShot from simplyblock_core.rpc_client import RPCException, RPCClient +from simplyblock_core.services.hub_controller_manager import hub_manager logger = utils.get_logger(__name__) db = db_mod.DBController() @@ -134,76 +135,12 @@ def _get_migration_nic(node): return trtype, node.mgmt_ip -def _ensure_hub_attached(src_rpc, tgt_rpc, tgt_node, trtype, target_ip): - """ - Ensure a migration-specific hub NVMe-oF controller is attached on source. - - Creates a dedicated hublvol (named "mighub") on the target lvstore - and exposes it via its own NVMe-oF subsystem. Using a separate bdev and NQN - avoids two Case B (src-is-tgt-secondary) problems: - • TRID conflict: the migration NQN differs from the node's regular hub NQN, - so SPDK does not reject a second controller to the same (addr, NQN). - • Exclusive claim: the new hublvol is unclaimed at creation, so - bdev_lvol_transfer can open it without hitting the NVMe-oF Target's - exclusive_write claim on the permanent hublvol. - - The hub controller is shared across all snapshot transfers and the final - migration — created once during SNAP_COPY, reused by PHASE_LVOL_MIGRATE, - detached in CLEANUP_SOURCE. All names/NQNs are deterministic so crash - recovery works without additional FDB state. - - Returns (ctrl_name, hub_bdev, error_string|None). - """ - - if tgt_node.transfer_hublvol is None or not tgt_node.transfer_hublvol.bdev_name: - tgt_node.create_transfer_hublvol() - - # Already attached (prior iteration or crash recovery). - if src_rpc.get_bdevs(tgt_node.transfer_hublvol.get_remote_bdev_name()): - return tgt_node.transfer_hublvol.bdev_name, tgt_node.transfer_hublvol.get_remote_bdev_name(), None - - for iface in tgt_node.data_nics: - ip = iface.ip4_address - if tgt_node.active_rdma: - if iface.trtype != "RDMA": - logger.debug("Skipping non-RDMA iface %s (active_rdma=True)", ip) - continue - trtype = "RDMA" - else: - if iface.trtype != "TCP": - logger.debug("Skipping non-TCP iface %s (active_tcp=True)", ip) - continue - trtype = "TCP" - - # Attach NVMe controller on source to the migration subsystem. - ret = src_rpc.bdev_nvme_attach_controller( - tgt_node.transfer_hublvol.bdev_name, tgt_node.transfer_hublvol.nqn, - ip, tgt_node.transfer_hublvol.nvmf_port, trtype) - if not ret: - # Attach can fail with EEXIST if a prior crashed attempt attached the controller - # but the namespace wasn't in the subsystem yet, or if the target SPDK restarted - # between migrations dropping the subsystem while the source controller persisted. - # Detach the zombie, re-ensure the hub subsystem is healthy on the target (idempotent), - # then retry once so the controller reconnects to a populated subsystem. - if src_rpc.bdev_nvme_controller_list(tgt_node.transfer_hublvol.bdev_name): - logger.info( - "_ensure_hub_attached: zombie mighub controller found (no bdev); " - "detaching and reattaching" - ) - src_rpc.bdev_nvme_detach_controller(tgt_node.transfer_hublvol.bdev_name) - try: - tgt_node.create_transfer_hublvol() - except Exception as e: - logger.warning(f"_ensure_hub_attached: hub subsystem re-create (non-fatal): {e}") - ret = src_rpc.bdev_nvme_attach_controller( - tgt_node.transfer_hublvol.bdev_name, tgt_node.transfer_hublvol.nqn, - ip, tgt_node.transfer_hublvol.nvmf_port, trtype) - if not ret: - return None, None, f"Failed to attach migration hub controller to {tgt_node.get_id()}" - return tgt_node.transfer_hublvol.bdev_name, tgt_node.transfer_hublvol.get_remote_bdev_name(), None - _MIGRATION_BDEV_SUFFIX = constants.LVOL_MIG_BDEV_SUFFIX +# Suffix applied when the canonical (no-suffix) name is already taken on the +# target after migration. "am" = "after migration" — stays distinct from the +# in-flight 'm' suffix so pre-existing migrated snapshots remain identifiable. +_MIGRATION_BDEV_SUFFIX_DONE = 'am' def _apply_migration_to_db(migration, tgt_lvol_uuid=None, tgt_lvol_bdev=None): @@ -309,16 +246,17 @@ def _apply_migration_to_db(migration, tgt_lvol_uuid=None, tgt_lvol_bdev=None): logger.warning(f"_apply_migration_to_db: snapshot not found: {snap_uuid}") continue - # snap_bdev: update lvstore prefix and add migration suffix. - # Guard is idempotent: if called twice after crash+retry, src_short - # already carries the suffix — do not append it again. + # snap_bdev: update lvstore prefix and normalise the migration suffix. + # Use the lvstore to detect a retry (snap already updated) rather than + # endswith(suffix), which gives a false-positive on back-to-back migrations + # where the source bdev legitimately ends with the suffix from the prior run. tgt_short = None if snap.snap_bdev and '/' in snap.snap_bdev: - src_short = snap.snap_bdev.split('/', 1)[1] - if src_short.endswith(_MIGRATION_BDEV_SUFFIX): - tgt_short = src_short + src_lvstore, src_short = snap.snap_bdev.split('/', 1) + if src_lvstore == tgt_node.lvstore: + tgt_short = src_short # already updated by a previous call — idempotent else: - tgt_short = src_short + _MIGRATION_BDEV_SUFFIX + tgt_short = _snap_tgt_short_name(snap) snap.snap_bdev = f"{tgt_node.lvstore}/{tgt_short}" if tgt_short and tgt_short in spdk_info: @@ -398,18 +336,25 @@ def _snap_short_name(snap): def _snap_tgt_short_name(snap): """Return the migration-target bdev short name for a snapshot. - Normally the target bdev is named + _MIGRATION_BDEV_SUFFIX. - However, when apply_migration_to_db() is called early (at cutover time), - snap.snap_bdev is already updated to the target name which already carries the - suffix. In that case return it as-is to avoid producing a double suffix (e.g. - 'SNAP_16745mm' instead of 'SNAP_16745m'). + Always strips any existing migration suffix before adding one so that + back-to-back migrations (where the source bdev already carries the suffix + from the previous migration) do not produce a double suffix like 'SNAP_16745mm'. """ short = _snap_short_name(snap) if short.endswith(_MIGRATION_BDEV_SUFFIX): - return short # apply_migration_to_db already updated this snap + short = short[:-len(_MIGRATION_BDEV_SUFFIX)] return short + _MIGRATION_BDEV_SUFFIX +def _lvol_tgt_bdev_name(lvol_bdev: str) -> str: + """Return the migration-target bdev short name for a writable lvol. + + Thin wrapper around the shared utils.lvol_tgt_bdev_name so existing + call-sites in this module don't need to change. + """ + return utils.lvol_tgt_bdev_name(lvol_bdev) + + def _snap_composite(lvstore, snap): """SPDK composite bdev name for a snapshot on a given node: ``/``.""" return f"{lvstore}/{_snap_short_name(snap)}" @@ -461,46 +406,50 @@ def _log_spdk_bdev_size(rpc, composite_name, label): return None -def _delete_bdev_blocking(bdev_name, primary_rpc, secondary_rpc=None, max_polls=120): +def _delete_bdev_blocking(bdev_name, primary_rpc, secondary_rpc=None, tertiary_rpc=None, + timeout_s=120, coalescing=False): """ - Full 3-step async-delete sequence for use in synchronous error-recovery paths. - Mirrors the control-plane pattern in storage_node_ops.safe_delete_bdev(): - - 1. delete_lvol(sync=False) on primary – start async background deletion - 2. poll bdev_lvol_get_lvol_delete_status until complete (0) or not-found (2) - 3. delete_lvol(sync=True) on primary – sync finalize / confirm removal - delete_lvol(sync=True) on secondary – sync finalize (best-effort) - - Blocks for up to max_polls × 0.25 s. Use only in error-recovery paths where - a bdev was just created and must be cleaned up before returning. + Two-phase blocking bdev delete. + + Phase 1 — primary only, sync=False (del_async=False): initiates the async + delete. By default (coalescing=False) special_delete=True tells SPDK to + free the bdev's own clusters without merging them into any child — + correct for source cleanup, rollback, and any path where no child needs + to inherit data. Pass coalescing=True when the bdev's child must + inherit its clusters instead (e.g. deleting a migration intermediate + snapshot, where the surviving lvol/snap needs the merged data). + Wait — poll bdev_lvol_get_lvol_delete_status on primary until done. + Phase 2 — all nodes (primary + secondary + tertiary), sync=True + (del_async=True, special_delete=False): finalises the deletion on every replica. """ - # Step 1: start async deletion - ret, _ = primary_rpc.delete_lvol(bdev_name) + ret, _ = primary_rpc.delete_lvol(bdev_name, del_async=False, special_delete=not coalescing) if not ret: - logger.warning(f"delete bdev {bdev_name}: async start failed (continuing)") + raise RuntimeError(f"delete bdev {bdev_name}: initiation failed") - # Step 2: poll - for _ in range(max_polls): - status = primary_rpc.bdev_lvol_get_lvol_delete_status(bdev_name) - if status in (0, 2): - break - if status == 1: - time.sleep(0.25) - else: - logger.warning(f"delete bdev {bdev_name}: unexpected status {status}") - break + deadline = time.monotonic() + timeout_s + while primary_rpc.bdev_lvol_get_lvol_delete_status(bdev_name) == 1: + if time.monotonic() > deadline: + if not primary_rpc.get_bdevs(bdev_name): + logger.warning( + f"delete bdev {bdev_name}: poll timed out after {timeout_s}s " + f"but bdev is gone — treating as success") + break + raise RuntimeError( + f"delete bdev {bdev_name}: timed out after {timeout_s}s, bdev still present") + time.sleep(0.2) - # Step 3: sync finalize - primary_rpc.delete_lvol(bdev_name, del_async=True) - if secondary_rpc: - secondary_rpc.delete_lvol(bdev_name, del_async=True) + for rpc in filter(None, [primary_rpc, secondary_rpc, tertiary_rpc]): + try: + rpc.delete_lvol(bdev_name, del_async=True, special_delete=False) + except Exception as e: + logger.warning(f"delete bdev {bdev_name} finalize on replica (non-fatal): {e}") # --------------------------------------------------------------------------- # Secondary-node helpers # --------------------------------------------------------------------------- -def _get_target_secondary_node(tgt_node, src_node_id=None): +def _get_target_secondary_node(tgt_node, src_node_id): """ Return ``(sec_node, error_string)`` describing how to handle the target's secondary node when creating a new object on the target primary. @@ -513,6 +462,9 @@ def _get_target_secondary_node(tgt_node, src_node_id=None): overlap drain: source is being drained but is still the target's secondary; migration must continue through it - Any other status → (None, err) block creation on primary + + ``src_node_id`` must always be supplied so the overlap-drain case (SUSPENDED + secondary that is also the migration source) is handled correctly. """ if not tgt_node.secondary_node_id: return None, None @@ -533,8 +485,23 @@ def _get_target_secondary_node(tgt_node, src_node_id=None): ) -def _get_target_tertiary_node(tgt_node, src_node_id=None): - """Like _get_target_secondary_node but for the tertiary node.""" +def _get_target_tertiary_node(tgt_node, src_node_id): + """ + Return ``(ter_node, error_string)`` describing how to handle the target's + tertiary node when creating a new object on the target primary. + + Rules mirror _get_target_secondary_node: + - No tertiary configured → (None, None) skip silently + - Tertiary STATUS_ONLINE → (ter_node, None) register on tertiary + - Tertiary STATUS_OFFLINE → (None, None) administratively down, skip + - Tertiary STATUS_SUSPENDED and node == src_node → (ter_node, None) + overlap drain: source is being drained but is still the target's + tertiary; migration must continue through it + - Any other status → (None, err) block creation on primary + + ``src_node_id`` must always be supplied so the overlap-drain case is + handled correctly. + """ if not tgt_node.tertiary_node_id: return None, None try: @@ -553,40 +520,42 @@ def _get_target_tertiary_node(tgt_node, src_node_id=None): ) +def _get_source_secondary_node(src_node): + """ + Return the secondary node of src_node for source cleanup operations, or None. - -def _get_target_secondary_nodes(tgt_node): + During source cleanup the secondary still holds replica data that must be + deleted. SUSPENDED is accepted because a draining node is still reachable + via RPC; OFFLINE is skipped as unreachable. """ - Return ``(nodes, error_string)`` for all secondary/tertiary nodes of tgt_node. + if not src_node.secondary_node_id: + return None + try: + sec = db.get_storage_node_by_id(src_node.secondary_node_id) + except KeyError: + return None + if sec.status in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): + return sec + return None - Rules (applied independently to each secondary and tertiary slot): - - Not configured → skip - - STATUS_ONLINE → include in result list - - STATUS_OFFLINE → skip (administratively down, non-blocking) - - Any other status → block: return ([], error_str) immediately - If any node blocks, the entire result is empty and the error describes - which node and what state it's in. +def _get_source_tertiary_node(src_node): """ - nodes = [] - for attr in ("secondary_node_id", "tertiary_node_id"): - node_id = getattr(tgt_node, attr, None) - if not node_id: - continue - try: - n = db.get_storage_node_by_id(node_id) - except KeyError: - continue - if n.status == StorageNode.STATUS_ONLINE: - nodes.append(n) - elif n.status == StorageNode.STATUS_OFFLINE: - continue - else: - return [], ( - f"Target node {node_id} is in state '{n.status}'; " - f"cannot proceed with migration" - ) - return nodes, None + Return the tertiary node of src_node for source cleanup operations, or None. + + Mirrors _get_source_secondary_node: ONLINE or SUSPENDED are both accepted; + OFFLINE is skipped. + """ + if not src_node.tertiary_node_id: + return None + try: + ter = db.get_storage_node_by_id(src_node.tertiary_node_id) + except KeyError: + return None + if ter.status in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): + return ter + return None + def _build_paths(src_node, tgt_node, src_rpc, tgt_rpc): @@ -658,7 +627,8 @@ def _swap_namespace(rpc, nqn, new_bdev, uuid, guid, label): # --------------------------------------------------------------------------- -def _cleanup_final_migration(src_rpc, ctx, tgt_rpc=None, rollback_target=False): +def _cleanup_final_migration(src_rpc, ctx, tgt_rpc=None, rollback_target=False, + tgt_sec_rpc=None, tgt_ter_rpc=None): """Clean up after a final lvol migration attempt. On the success path (rollback_target=False) the hub controller is kept @@ -687,7 +657,8 @@ def _cleanup_final_migration(src_rpc, ctx, tgt_rpc=None, rollback_target=False): logger.warning(f"cleanup target subsystem {nqn}: {e}") if tgt_composite and tgt_rpc.get_bdevs(tgt_composite): try: - _delete_bdev_blocking(tgt_composite, tgt_rpc) + _delete_bdev_blocking(tgt_composite, tgt_rpc, + secondary_rpc=tgt_sec_rpc, tertiary_rpc=tgt_ter_rpc) except Exception as e: logger.warning(f"cleanup target lvol {tgt_composite}: {e}") @@ -697,13 +668,14 @@ def _cleanup_final_migration(src_rpc, ctx, tgt_rpc=None, rollback_target=False): # --------------------------------------------------------------------------- -def _setup_snap_transfer(snap, snap_index, migration, src_node, tgt_node, - src_rpc, tgt_rpc, trtype, target_ip, - tgt_sec=None, sec_rpc=None, lvol_size_mib=None): +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): """ Prepare a single snapshot for async transfer: 1. Create writable lvol on target primary - 2. Register on target secondary immediately (keeps secondary consistent) + 2. Register on target secondary/tertiary immediately (keeps HA nodes consistent) 3. Set migration flag on primary 4. Get map_id of target bdev for hub-based transfer 5. Ensure hub NVMe-oF controller is attached on source @@ -728,7 +700,16 @@ def _setup_snap_transfer(snap, snap_index, migration, src_node, tgt_node, f"size_in_mib={size_in_mib} (lvol_size_mib={lvol_size_mib})" ) _log_spdk_bdev_size(src_rpc, src_composite, f"SRC snap[{snap_uuid[:8]}] pre-create") - ret = tgt_rpc.create_lvol(snap_short, size_in_mib, tgt_node.lvstore) + # snap.lvol is the embedded copy of the owning lvol captured at snapshot + # creation time; it should always be populated but we degrade gracefully + # for legacy/corrupt records by falling back to pool defaults (0 = use + # the pool's default replication scheme / priority class) rather than + # failing the migration outright. + snap_lvol = snap.lvol + _ndcs = snap_lvol.ndcs if snap_lvol else 0 + _npcs = snap_lvol.npcs if snap_lvol else 0 + _priority_class = snap_lvol.lvol_priority_class if snap_lvol else 0 + ret = tgt_rpc.create_lvol(snap_short, size_in_mib, tgt_node.lvstore, ndcs=_ndcs, npcs=_npcs) if not ret: return None, f"Failed to create target lvol for snap {snap_uuid}" _log_spdk_bdev_size(tgt_rpc, tgt_composite, f"TGT snap[{snap_uuid[:8]}] post-create") @@ -737,25 +718,47 @@ def _setup_snap_transfer(snap, snap_index, migration, src_node, tgt_node, # If registration fails we clean up the primary bdev and abort — continuing # with an unregistered secondary would leave the cluster in split state. sec_registered = False + ter_registered = False if tgt_sec and sec_rpc: bdev_info = tgt_rpc.get_bdevs(tgt_composite) if not bdev_info: - _delete_bdev_blocking(tgt_composite, tgt_rpc) + try: + _delete_bdev_blocking(tgt_composite, tgt_rpc) + except Exception as e: + logger.warning(f"cleanup target lvol {tgt_composite} (non-fatal): {e}") return None, f"Could not get bdev info for {tgt_composite} after creation" snap_blobid = bdev_info[0]['driver_specific']['lvol']['blobid'] snap_uuid_on_tgt = bdev_info[0]['uuid'] ret_sec = sec_rpc.bdev_lvol_register( snap_short, tgt_node.lvstore, snap_uuid_on_tgt, snap_blobid, - snap.lvol.lvol_priority_class if hasattr(snap, 'lvol') else 0) + _priority_class) if not ret_sec: - _delete_bdev_blocking(tgt_composite, tgt_rpc, sec_rpc) + try: + _delete_bdev_blocking(tgt_composite, tgt_rpc, sec_rpc) + except Exception as e: + logger.warning(f"cleanup target lvol {tgt_composite} (non-fatal): {e}") return None, f"bdev_lvol_register on secondary failed for snap {snap_uuid}" sec_registered = True + if tgt_ter and ter_rpc: + ret_ter = ter_rpc.bdev_lvol_register( + snap_short, tgt_node.lvstore, snap_uuid_on_tgt, snap_blobid, + _priority_class) + if not ret_ter: + try: + _delete_bdev_blocking(tgt_composite, tgt_rpc, sec_rpc, ter_rpc) + except Exception as e: + logger.warning(f"cleanup target lvol {tgt_composite} (non-fatal): {e}") + return None, f"bdev_lvol_register on tertiary failed for snap {snap_uuid}" + ter_registered = True - # Helper: clean primary and secondary (if registered) on error + # Helper: clean primary, secondary, and tertiary (if registered) on error def _cleanup(): - _delete_bdev_blocking(tgt_composite, tgt_rpc, - secondary_rpc=sec_rpc if sec_registered else None) + try: + _delete_bdev_blocking(tgt_composite, tgt_rpc, + secondary_rpc=sec_rpc if sec_registered else None, + tertiary_rpc=ter_rpc if ter_registered else None) + except Exception as e: + logger.warning(f"cleanup target lvol {tgt_composite} (non-fatal): {e}") # Step 3: migration flag on primary ret = tgt_rpc.bdev_lvol_set_migration_flag(tgt_composite) @@ -777,10 +780,10 @@ def _cleanup(): return None, f"Could not get map_id for snap {snap_uuid} on target" # Step 5: ensure hub controller is attached on source (shared across all - # snapshot transfers; created once, reused by PHASE_LVOL_MIGRATE, detached - # in CLEANUP_SOURCE). - _, hub_bdev, hub_err = _ensure_hub_attached( - src_rpc, tgt_rpc, tgt_node, trtype, target_ip) + # snapshot transfers; created once, reused by PHASE_LVOL_MIGRATE, released + # in CLEANUP_SOURCE and detached lazily by HubControllerManager). + _, hub_bdev, hub_err = hub_manager.acquire( + src_node.get_id(), src_rpc, tgt_node, trtype) if hub_err: _cleanup() return None, hub_err @@ -801,7 +804,8 @@ def _cleanup(): def _post_process_snap(snap: SnapShot, tgt_node: StorageNode, tgt_rpc: RPCClient, migration: LVolMigration, - transfer: dict, tgt_sec:Optional[StorageNode]=None, sec_rpc: Optional[RPCClient]=None): + transfer: dict, tgt_sec:Optional[StorageNode]=None, sec_rpc: Optional[RPCClient]=None, + tgt_ter:Optional[StorageNode]=None, ter_rpc: Optional[RPCClient]=None): """ Post-transfer steps for a single snapshot whose data has been fully copied: add_clone → convert (on primary, then mirrored on secondary) → cleanup. @@ -839,6 +843,10 @@ def _post_process_snap(snap: SnapShot, tgt_node: StorageNode, tgt_rpc: RPCClient ret_sec = sec_rpc.bdev_lvol_add_clone(tgt_composite, pred_composite) if not ret_sec: return False, f"bdev_lvol_add_clone on secondary failed for {snap_uuid}" + if tgt_ter and ter_rpc: + ret_ter = ter_rpc.bdev_lvol_add_clone(tgt_composite, pred_composite) + if not ret_ter: + return False, f"bdev_lvol_add_clone on tertiary failed for {snap_uuid}" except KeyError: logger.warning(f"Predecessor snap {pred_uuid} not found; skipping add_clone") @@ -853,6 +861,10 @@ def _post_process_snap(snap: SnapShot, tgt_node: StorageNode, tgt_rpc: RPCClient ret_sec = sec_rpc.bdev_lvol_convert(tgt_composite) if not ret_sec: return False, f"bdev_lvol_convert on secondary failed for {snap_uuid}" + if tgt_ter and ter_rpc: + ret_ter = ter_rpc.bdev_lvol_convert(tgt_composite) + if not ret_ter: + return False, f"bdev_lvol_convert on tertiary failed for {snap_uuid}" # Early partial DB update: route health-check and delete to the target node # immediately after convert. snap_bdev keeps its source path here; the full @@ -912,7 +924,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): Returns (done: bool, suspend: bool, error: str|None). """ plan = migration.snap_migration_plan - trtype, target_ip = _get_migration_nic(tgt_node) + trtype, _ = _get_migration_nic(tgt_node) ctx = migration.transfer_context or {} # Snap bdevs on TGT must cover the full logical address range of the lvol, @@ -923,18 +935,70 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): except KeyError: _snap_lvol_size_mib = None + tgt_sec = None + sec_rpc = None + tgt_ter = None + ter_rpc = None + + # ── PRE-SCAN: mark snapshots already on target as pre-existing ──────────── + # Query the target lvstore once. Any planned snap whose target bdev already + # exists AND is immutable (is_snapshot=true in SPDK) was fully converted by + # a prior migration — mark pre-existing so we skip the transfer entirely. + # We use the SPDK is_snapshot flag rather than the DB snap_bdev field because + # snap_bdev is not updated until apply_migration_to_db() runs at the very end + # of CLEANUP_SOURCE; a migration that succeeded at SNAP_COPY but failed later + # would leave immutable snapshots on the target with stale DB records. + # Writable bdevs (is_snapshot=false) are leftovers from a crashed transfer — + # they fall through to the per-snap pre-cleanup to be deleted and retried. + if ctx.get('stage') != 'parallel_transfer' and plan: + try: + _tgt_lvols = tgt_rpc.bdev_lvol_get_lvols(tgt_node.lvstore) or [] + _tgt_immutable = { + e.get('name', '').split('/')[-1] + for e in _tgt_lvols + if e.get('is_snapshot', False) + } + except Exception as _pre_e: + logger.warning(f"Pre-scan: bdev_lvol_get_lvols failed ({_pre_e}); skipping") + _tgt_immutable = set() + + _pre_scan_updated = False + for _snap_uuid in plan: + if (_snap_uuid in migration.snaps_migrated + or _snap_uuid in migration.snaps_preexisting_on_target): + continue + try: + _s = db.get_snapshot_by_id(_snap_uuid) + except KeyError: + continue + _short_tgt = _snap_tgt_short_name(_s) # SNAP_Xm (in-flight) + _short_canonical = _snap_short_name(_s) # SNAP_X (post-rename) + _short_am = _short_canonical + _MIGRATION_BDEV_SUFFIX_DONE # SNAP_Xam (fallback) + _found_as = next( + (n for n in (_short_tgt, _short_canonical, _short_am) + if n in _tgt_immutable), + None) + if _found_as: + logger.info( + f"Pre-scan: {_snap_uuid} ({_found_as}) is already an immutable " + f"snapshot on target; marking pre-existing") + migration.snaps_preexisting_on_target.append(_snap_uuid) + _pre_scan_updated = True + if _pre_scan_updated: + migration.write_to_db(db.kv_store) + # ── A. Launch / resume planned snapshots one at a time ─────────────────── # SPDK only supports one bdev_lvol_transfer per poller group at a time; # launching multiple causes "poller already exists" and stuck transfers. _PARALLEL_BATCH = 1 if ctx.get('stage') != 'parallel_transfer': - all_unprocessed = [u for u in plan if u not in migration.snaps_migrated] + all_unprocessed = [u for u in plan + if u not in migration.snaps_migrated + and u not in migration.snaps_preexisting_on_target] unprocessed = all_unprocessed[:_PARALLEL_BATCH] if unprocessed: - # HA secondary gate – check once; all snaps belong to the same volume - tgt_sec = None - sec_rpc = None + # HA secondary/tertiary gate – check once; all snaps belong to the same volume for snap_uuid in unprocessed: try: snap = db.get_snapshot_by_id(snap_uuid) @@ -948,7 +1012,22 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): return False, True, _WAIT if tgt_sec: sec_rpc = _make_rpc(tgt_sec) - break # one check is enough + elif snap.lvol.ha_type == "ha3": + tgt_sec, sec_err = _get_target_secondary_node(tgt_node, src_node.get_id()) + if sec_err: + migration.error_message = sec_err + migration.write_to_db(db.kv_store) + return False, True, _WAIT + if tgt_sec: + sec_rpc = _make_rpc(tgt_sec) + tgt_ter, ter_err = _get_target_tertiary_node(tgt_node, src_node.get_id()) + if ter_err: + migration.error_message = ter_err + migration.write_to_db(db.kv_store) + return False, True, _WAIT + if tgt_ter: + ter_rpc = _make_rpc(tgt_ter) + break # one check is enough transfers: list[dict] = [] for snap_uuid in unprocessed: @@ -977,30 +1056,26 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): }) continue - # Check for a bdev already present on the target. - # Two distinct cases must be handled: - # 1. Genuinely pre-existing (placed by a sibling migration that - # already ran) — snap is already in snaps_migrated → skip. - # 2. Leftover from a previous failed attempt of THIS migration — - # snap is NOT yet in snaps_migrated → delete and retry, so we - # do not resume from a partially-written bdev. + # Pre-existing (immutable) bdevs were caught by the pre-scan above and + # excluded from unprocessed. Anything still found here is a writable + # leftover from a previous failed attempt — delete and retry. if tgt_rpc.get_bdevs(tgt_composite): - if snap_uuid in migration.snaps_migrated: - logger.info( - f"Snapshot {snap_uuid} already on target; skipping transfer") - migration.snaps_preexisting_on_target.append(snap_uuid) - continue logger.info( - f"Removing leftover target bdev {tgt_composite} from failed attempt") + f"Removing writable leftover target bdev {tgt_composite}") try: - _delete_bdev_blocking(tgt_composite, tgt_rpc, sec_rpc) + _delete_bdev_blocking(tgt_composite, tgt_rpc, sec_rpc, ter_rpc) + for _ in range(10): + if not tgt_rpc.get_bdevs(tgt_composite): + break + time.sleep(0.2) except Exception as e: logger.warning(f"Pre-cleanup of {tgt_composite} failed (continuing): {e}") t, err = _setup_snap_transfer( - snap, snap_index, migration, src_node, tgt_node, - src_rpc, tgt_rpc, trtype, target_ip, + snap, snap_index, src_node, tgt_node, + src_rpc, tgt_rpc, trtype, tgt_sec=tgt_sec, sec_rpc=sec_rpc, + tgt_ter=tgt_ter, ter_rpc=ter_rpc, lvol_size_mib=_snap_lvol_size_mib) if t is None: return False, True, err @@ -1030,7 +1105,7 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): if ctx.get('stage') == 'parallel_transfer': transfers = ctx['transfers'] # Resolve secondary once for the whole poll pass - tgt_sec, _sec_err = _get_target_secondary_node(tgt_node) + tgt_sec, _sec_err = _get_target_secondary_node(tgt_node, src_node.get_id()) sec_rpc = _make_rpc(tgt_sec) if tgt_sec and not _sec_err else None # Process in snap_index order: add_clone requires predecessor to be # converted first. prev_post_done tracks whether the predecessor has @@ -1082,7 +1157,8 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): ok, err = _post_process_snap( snap, tgt_node, tgt_rpc, migration, t, - tgt_sec=tgt_sec, sec_rpc=sec_rpc) + tgt_sec=tgt_sec, sec_rpc=sec_rpc, + tgt_ter=tgt_ter, ter_rpc=ter_rpc) if not ok: migration.transfer_context = {} if err is _WAIT: @@ -1111,7 +1187,9 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): # If there are more unprocessed snaps, return now so the next tick # launches the next batch. - remaining = [u for u in plan if u not in migration.snaps_migrated] + remaining = [u for u in plan + if u not in migration.snaps_migrated + and u not in migration.snaps_preexisting_on_target] if remaining: return False, False, None @@ -1123,33 +1201,27 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): # Before each round check the current dirty delta. If it is already below # the threshold the remaining freeze window will be short enough that no # additional shrink pass is worth the overhead. - # If no snapshots have been migrated at all (volume had no pre-existing snapshots - # and none were planned), we must take at least one intermediate snapshot so that - # _handle_lvol_migrate has a base to clone from. Skip the delta threshold check - # on the first pass in this case. - _must_snap = (not migration.snaps_migrated - and not migration.snaps_preexisting_on_target) - 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}" _delta = _get_lvol_delta_bytes(src_rpc, _src_composite) _threshold = constants.LVOL_MIG_INTERMEDIATE_SNAP_THRESHOLD_BYTES - if not _must_snap and _delta is not None and _delta <= _threshold: + if migration.intermediate_snap_rounds > 0 and _delta is not None and _delta <= _threshold: logger.info( f"Intermediate snapshot skipped: delta {convert_size(_delta, 'MiB')} MiB " f"<= {convert_size(_threshold, 'MiB')} MiB threshold " f"(round {migration.intermediate_snap_rounds}/{migration.max_intermediate_snap_rounds})") break - logger.info( - f"Intermediate snapshot triggered: delta " - f"{'unknown' if _delta is None else str(convert_size(_delta, 'MiB')) + ' MiB'} " - f"exceeds {convert_size(_threshold, 'MiB')} MiB threshold") + _reason = ( + "forced (round 0)" if migration.intermediate_snap_rounds == 0 + else f"delta {'unknown' if _delta is None else str(convert_size(_delta, 'MiB')) + ' MiB'} " + f"exceeds {convert_size(_threshold, 'MiB')} MiB threshold" + ) + logger.info(f"Intermediate snapshot triggered: {_reason}") _take_intermediate_snapshot(migration) - _must_snap = False plan = migration.snap_migration_plan if not plan: - return False, True, "Intermediate snapshot failed and no snapshots available for migration" + return False, True, "Intermediate snapshot failed" snap_uuid = plan[-1] snap_index = len(plan) - 1 @@ -1160,7 +1232,9 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): tgt_sec = None sec_rpc = None - if snap.lvol.ha_type == "ha": + tgt_ter = None + ter_rpc = None + if snap.lvol.ha_type in ("ha", "ha3"): tgt_sec, sec_err = _get_target_secondary_node(tgt_node, src_node.get_id()) if sec_err: migration.error_message = sec_err @@ -1168,24 +1242,39 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): return False, True, _WAIT if tgt_sec: sec_rpc = _make_rpc(tgt_sec) + if snap.lvol.ha_type == "ha3": + tgt_ter, ter_err = _get_target_tertiary_node(tgt_node, src_node.get_id()) + if ter_err: + migration.error_message = ter_err + migration.write_to_db(db.kv_store) + return False, True, _WAIT + if tgt_ter: + ter_rpc = _make_rpc(tgt_ter) snap_short_tgt = _snap_tgt_short_name(snap) src_composite = _snap_composite(src_node.lvstore, snap) tgt_composite = f"{tgt_node.lvstore}/{snap_short_tgt}" - # Pre-cleanup: only delete if the bdev actually exists on the target - # (stale from a previous crashed run). Deleting blindly masks real errors. + # Pre-cleanup: if a bdev exists on the target it is a writable leftover + # from a previous crashed run — intermediate snaps are always freshly + # created by this migration so they can never be pre-existing. if tgt_rpc.get_bdevs(tgt_composite): logger.info(f"Pre-cleanup: removing stale intermediate bdev {tgt_composite}") try: - _delete_bdev_blocking(tgt_composite, tgt_rpc) + _delete_bdev_blocking(tgt_composite, tgt_rpc, + secondary_rpc=sec_rpc, tertiary_rpc=ter_rpc) + for _ in range(10): + if not tgt_rpc.get_bdevs(tgt_composite): + break + time.sleep(0.2) except Exception as e: logger.warning(f"Pre-cleanup of {tgt_composite} failed (continuing): {e}") t, err = _setup_snap_transfer( - snap, snap_index, migration, src_node, tgt_node, - src_rpc, tgt_rpc, trtype, target_ip, + snap, snap_index, src_node, tgt_node, + src_rpc, tgt_rpc, trtype, tgt_sec=tgt_sec, sec_rpc=sec_rpc, + tgt_ter=tgt_ter, ter_rpc=ter_rpc, lvol_size_mib=_snap_lvol_size_mib) if t is None: return False, True, err @@ -1198,25 +1287,38 @@ def _handle_snap_copy(migration, src_node, tgt_node, src_rpc, tgt_rpc): for _ in range(_INTERMEDIATE_POLL_MAX): result = src_rpc.bdev_lvol_transfer_stat(src_composite) if result is None: - _delete_bdev_blocking(tgt_composite, tgt_rpc, secondary_rpc=sec_rpc) + try: + _delete_bdev_blocking(tgt_composite, tgt_rpc, + secondary_rpc=sec_rpc, tertiary_rpc=ter_rpc) + except Exception as e: + logger.warning(f"cleanup target snap {tgt_composite} (non-fatal): {e}") return False, True, ( f"Transfer stat failed for intermediate snap {snap_uuid}") state = result.get('transfer_state', 'No process') if state == 'Done': break if state in ('Failed', 'No process'): - _delete_bdev_blocking(tgt_composite, tgt_rpc, secondary_rpc=sec_rpc) + try: + _delete_bdev_blocking(tgt_composite, tgt_rpc, + secondary_rpc=sec_rpc, tertiary_rpc=ter_rpc) + except Exception as e: + logger.warning(f"cleanup target snap {tgt_composite} (non-fatal): {e}") return False, True, ( f"Intermediate snap transfer {state} for {snap_uuid}") time.sleep(_INTERMEDIATE_POLL_INTERVAL_S) else: - _delete_bdev_blocking(tgt_composite, tgt_rpc, secondary_rpc=sec_rpc) + try: + _delete_bdev_blocking(tgt_composite, tgt_rpc, + secondary_rpc=sec_rpc, tertiary_rpc=ter_rpc) + except Exception as e: + logger.warning(f"cleanup target snap {tgt_composite} (non-fatal): {e}") return False, True, ( f"Intermediate snap transfer timed out for {snap_uuid}") ok, err = _post_process_snap( snap, tgt_node, tgt_rpc, migration, t, - tgt_sec=tgt_sec, sec_rpc=sec_rpc) + tgt_sec=tgt_sec, sec_rpc=sec_rpc, + tgt_ter=tgt_ter, ter_rpc=ter_rpc) if not ok: if err is _WAIT: migration.error_message = ( @@ -1298,7 +1400,8 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): Creates the target lvol with the same NQN as the source subsystem, connects the source to the target's hub lvol, and issues bdev_lvol_final_migration (synchronous — blocks until SPDK completes the delta copy). On success, - immediately rebuilds the TGT NVMe-oF subsystem and sets STATUS_CUTOVER. + immediately rebuilds the TGT NVMe-oF subsystem and applies the DB records + so clients polling migration status see the TGT endpoints at cutover time. Note: apply_migration_to_db() is NOT called here; it is deferred to the end of PHASE_CLEANUP_SOURCE after source snap deletion is complete. @@ -1310,11 +1413,15 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): except KeyError as e: return False, True, str(e) - trtype, target_ip = _get_migration_nic(tgt_node) + trtype, _ = _get_migration_nic(tgt_node) src_lvol_composite = f"{src_node.lvstore}/{lvol.lvol_bdev}" - tgt_lvol_bdev = lvol.lvol_bdev + _MIGRATION_BDEV_SUFFIX + 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 {} + tgt_sec, _ = _get_target_secondary_node(tgt_node, src_node.get_id()) + tgt_sec_rpc = _make_rpc(tgt_sec) if tgt_sec else None + tgt_ter, _ = _get_target_tertiary_node(tgt_node, src_node.get_id()) + tgt_ter_rpc = _make_rpc(tgt_ter) if tgt_ter else None # --- Crash recovery: Done handler was interrupted mid-run --- # bdev_lvol_final_migration is synchronous — it blocks until SPDK completes. @@ -1323,13 +1430,15 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): if ctx.get('stage') == 'transfer': result = src_rpc.bdev_lvol_transfer_stat(src_lvol_composite) if result is None: - _cleanup_final_migration(src_rpc, ctx, tgt_rpc, rollback_target=True) + _cleanup_final_migration(src_rpc, ctx, tgt_rpc, rollback_target=True, + tgt_sec_rpc=tgt_sec_rpc, tgt_ter_rpc=tgt_ter_rpc) migration.transfer_context = {} migration.write_to_db(db.kv_store) return False, True, "bdev_lvol_transfer_stat returned None (crash recovery)" state = result.get('transfer_state', 'No process') if state == 'Failed': - _cleanup_final_migration(src_rpc, ctx, tgt_rpc, rollback_target=True) + _cleanup_final_migration(src_rpc, ctx, tgt_rpc, rollback_target=True, + tgt_sec_rpc=tgt_sec_rpc, tgt_ter_rpc=tgt_ter_rpc) migration.transfer_context = {} migration.write_to_db(db.kv_store) return False, True, "Final migration Failed (crash recovery)" @@ -1387,27 +1496,24 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): return False, True, f"Could not find map_id for {lvol.lvol_bdev} on target" logger.info(f"[MAP_ID] {tgt_lvol_bdev} map_id={tgt_map_id} uuid={tgt_uuid} on {tgt_node.get_id()[:8]}") - sec_setup_rpc = None # NVMe-oF subsystem setup is deferred to the Done handler — the subsystem # is deleted and recreated fresh after transfer completes so all paths get # a clean primary-port subsystem (min_cntlid=2000). # Step 3: connect source to target migration hub lvol - ctrl_name, hub_bdev, hub_err = _ensure_hub_attached( - src_rpc, tgt_rpc, tgt_node, trtype, target_ip) + ctrl_name, hub_bdev, hub_err = hub_manager.acquire( + src_node.get_id(), src_rpc, tgt_node, trtype) if hub_err: # Do NOT delete the target bdev on hub error — it is unrelated to # the hub connection and deleting it forces a recreate on retry, # which changes its map_id and breaks concurrent migration tracking. return False, True, hub_err - # Step 4: locate the last migrated snapshot's composite name on the target - if not migration.snaps_migrated and not migration.snaps_preexisting_on_target: - src_rpc.bdev_nvme_detach_controller(ctrl_name) - _delete_bdev_blocking(tgt_lvol_composite, tgt_rpc, secondary_rpc=sec_setup_rpc) - return False, True, "No snapshots migrated; cannot perform final migration" - + # Step 4: locate the last migrated snapshot's composite name on the target. + # At least one intermediate snapshot is always taken (round 0 is unconditional), + # so snaps_migrated is non-empty in the normal path. snaps_preexisting_on_target + # covers the case where a prior migration already placed immutable snapshots. tgt_snap_composite = "" if migration.snaps_migrated: last_snap_uuid = migration.snaps_migrated[-1] @@ -1415,10 +1521,14 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): last_snap = db.get_snapshot_by_id(last_snap_uuid) except KeyError: src_rpc.bdev_nvme_detach_controller(ctrl_name) - _delete_bdev_blocking(tgt_lvol_composite, tgt_rpc, secondary_rpc=sec_setup_rpc) + try: + _delete_bdev_blocking(tgt_lvol_composite, tgt_rpc, + secondary_rpc=tgt_sec_rpc, tertiary_rpc=tgt_ter_rpc) + except Exception as e: + logger.warning(f"cleanup target lvol {tgt_lvol_composite} (non-fatal): {e}") return False, True, f"Last snapshot {last_snap_uuid} not found" tgt_snap_composite = f"{tgt_node.lvstore}/{_snap_tgt_short_name(last_snap)}" - else: + elif migration.snaps_preexisting_on_target: last_snap_uuid = migration.snaps_preexisting_on_target[-1] last_snap = db.get_snapshot_by_id(last_snap_uuid) if last_snap.lvol.node_id == tgt_node.get_id(): @@ -1433,11 +1543,16 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): if isinstance(snap_bdev, str): tgt_snap_composite = snap_bdev break - - if not tgt_snap_composite: - src_rpc.bdev_nvme_detach_controller(ctrl_name) - _delete_bdev_blocking(tgt_lvol_composite, tgt_rpc, secondary_rpc=sec_setup_rpc) - return False, True, f"snapshot {last_snap_uuid} not found on target" + if not tgt_snap_composite: + src_rpc.bdev_nvme_detach_controller(ctrl_name) + try: + _delete_bdev_blocking(tgt_lvol_composite, tgt_rpc, + secondary_rpc=tgt_sec_rpc, tertiary_rpc=tgt_ter_rpc) + except Exception as e: + logger.warning(f"cleanup target lvol {tgt_lvol_composite} (non-fatal): {e}") + return False, True, f"snapshot {last_snap_uuid} not found on target" + # else: no snapshots at all — tgt_snap_composite stays "" and SPDK + # performs a full-lvol transfer without a snapshot base. # Step 5: start final migration — synchronous: blocks until SPDK completes # the IO drain and delta copy. Returns success/failure directly; no polling needed. @@ -1447,16 +1562,25 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): ret = src_rpc.bdev_lvol_final_migration( src_lvol_composite, tgt_map_id, tgt_snap_composite, 2, hub_bdev) if ret is None: - src_rpc.bdev_nvme_detach_controller(ctrl_name) - # 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 - # concurrent migrations creating bdevs in the interim). - return False, True, "bdev_lvol_final_migration failed" - - logger.info( - f"[IO-RESUME] {_now_ms()} final migration Done: " - f"lvol={migration.lvol_id} io now live on target") + # Connection timeout or SPDK error (e.g. "File exists" = already in progress). + # SPDK may have completed the migration while the RPC connection dropped. + # Check transfer_stat before treating this as a hard failure. + stat = src_rpc.bdev_lvol_transfer_stat(src_lvol_composite) + state = (stat or {}).get('transfer_state') if stat is not None else None + if state not in ('Done', 'No process'): + src_rpc.bdev_nvme_detach_controller(ctrl_name) + # 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 + # concurrent migrations creating bdevs in the interim). + return False, True, "bdev_lvol_final_migration failed" + logger.info( + f"[IO-RESUME] {_now_ms()} final migration complete (recovered from RPC error, " + f"transfer_state={state}): lvol={migration.lvol_id} io now live on target") + else: + logger.info( + f"[IO-RESUME] {_now_ms()} final migration Done: " + f"lvol={migration.lvol_id} io now live on target") # add_clone on secondary and tertiary to link the final migrated lvol to # its predecessor snapshot in their ancestry chain. @@ -1464,8 +1588,8 @@ def _handle_lvol_migrate(migration, src_node, tgt_node, src_rpc, tgt_rpc): # non-primary nodes need an explicit call. _clone_tgt_composite = f"{tgt_node.lvstore}/{tgt_lvol_bdev}" for _extra_node in filter(None, [ - _get_target_secondary_node(tgt_node)[0], - _get_target_tertiary_node(tgt_node)[0], + _get_target_secondary_node(tgt_node, src_node.get_id())[0], + _get_target_tertiary_node(tgt_node, src_node.get_id())[0], ]): _ret = _make_rpc(_extra_node).bdev_lvol_add_clone( _clone_tgt_composite, tgt_snap_composite) @@ -1531,27 +1655,51 @@ def _flip(rpc, ip, port, trtype, state, label): rpc.nvmf_subsystem_listener_set_ana_state( nqn, ip, port, trtype=trtype, ana=state) logger.info(f"ANA {nqn} {label} {ip}:{port} → {state}") + return True except Exception as e: - logger.error(f"ANA {label} failed (non-fatal): {e}") + logger.error(f"ANA {label} failed: {e}") + return False + + def _flip_required(rpc, ip, port, trtype, state, label, attempts=3): + for i in range(attempts): + if _flip(rpc, ip, port, trtype, state, label): + return True + if i < attempts - 1: + time.sleep(1.0) + return False if not overlap_ids: - # Step 1 (no-overlap): set all TGT to final states before killing SRC - for i, tgt in enumerate(tgt_paths): - state = "optimized" if i == 0 else "non_optimized" + # Step 1 (no-overlap): TGT primary must be confirmed optimized before + # making SRC inaccessible — otherwise clients lose all I/O paths. + primary_tgt = tgt_paths[0] + if not _flip_required(primary_tgt['rpc'], primary_tgt['ip'], primary_tgt['port'], + primary_tgt['trtype'], "optimized", + f"TGT-{primary_tgt['node_id'][:8]}"): + return False, False, ( + "ANA flip: TGT primary→optimized failed after retries " + "— aborting to keep SRC paths accessible") + + # Secondary TGT paths best-effort — clients survive without them + for tgt in tgt_paths[1:]: _flip(tgt['rpc'], tgt['ip'], tgt['port'], tgt['trtype'], - state, f"TGT-{tgt['node_id'][:8]}") + "non_optimized", f"TGT-{tgt['node_id'][:8]}") # Step 3 (no-overlap): all SRC paths → inaccessible for src in src_paths: _flip(src['rpc'], src['ip'], src['port'], src['trtype'], "inaccessible", f"SRC-{src['node_id'][:8]}") else: - # Step 1: first non-overlap TGT → optimized (live path before touching overlap) - for tgt in tgt_paths: - if tgt['node_id'] not in overlap_ids: - _flip(tgt['rpc'], tgt['ip'], tgt['port'], tgt['trtype'], - "optimized", f"TGT-{tgt['node_id'][:8]}(pre)") - break + # Step 1: first non-overlap TGT → optimized. Must succeed before + # making any SRC path inaccessible. + non_overlap_tgt = next( + (t for t in tgt_paths if t['node_id'] not in overlap_ids), None) + if non_overlap_tgt: + if not _flip_required(non_overlap_tgt['rpc'], non_overlap_tgt['ip'], + non_overlap_tgt['port'], non_overlap_tgt['trtype'], + "optimized", f"TGT-{non_overlap_tgt['node_id'][:8]}(pre)"): + return False, False, ( + "ANA flip: non-overlap TGT primary→optimized failed after retries " + "— aborting to keep SRC paths accessible") # Step 2: overlap SRC paths → inaccessible at SRC port for src in src_paths: @@ -1570,17 +1718,20 @@ def _flip(rpc, ip, port, trtype, state, label): # create_migration; for plain it is the raw migration lvol. for tgt in tgt_paths: if tgt['node_id'] in overlap_ids: - try: - _swap_namespace(tgt['rpc'], nqn, tgt_ns_bdev, - lvol.uuid, lvol.guid, tgt['node_id'][:8]) - except Exception as e: - logger.error(f"Namespace swap on {tgt['node_id'][:8]} failed: {e}") - - # Step 5: all TGT paths → correct ANA state at TGT port - for i, tgt in enumerate(tgt_paths): - state = "optimized" if i == 0 else "non_optimized" + _swap_namespace(tgt['rpc'], nqn, tgt_ns_bdev, + lvol.uuid, lvol.guid, tgt['node_id'][:8]) + + # Step 5: all TGT paths → correct ANA state at TGT port. + # Primary required; secondaries best-effort. + primary_tgt = tgt_paths[0] + if not _flip_required(primary_tgt['rpc'], primary_tgt['ip'], primary_tgt['port'], + primary_tgt['trtype'], "optimized", + f"TGT-{primary_tgt['node_id'][:8]}"): + return False, False, ( + "ANA flip: TGT primary→optimized (step 5) failed after retries") + for tgt in tgt_paths[1:]: _flip(tgt['rpc'], tgt['ip'], tgt['port'], tgt['trtype'], - state, f"TGT-{tgt['node_id'][:8]}") + "non_optimized", f"TGT-{tgt['node_id'][:8]}") # Step 6: overlap TGT paths → remove old SRC listener if port changed for tgt in tgt_paths: @@ -1655,7 +1806,7 @@ def _cleanup_subsystem_or_ns(nqn, ns_id, subsystem_was_created_by_migration, rpc rpc.subsystem_delete(nqn) -def _delete_intermediate_snaps_on_target(migration, tgt_rpc, tgt_sec_rpc=None): +def _delete_intermediate_snaps_on_target(migration, tgt_rpc, tgt_sec_rpc=None, tgt_ter_rpc=None): """ Delete migration-created intermediate ('shrink') snapshots from the target after a successful migration. @@ -1663,10 +1814,10 @@ def _delete_intermediate_snaps_on_target(migration, tgt_rpc, tgt_sec_rpc=None): Must be called AFTER apply_migration_to_db() — at that point snap.snap_bdev already holds the target composite path (e.g. LVS_TGT/SNAP_xxxm). - No migration flag is set before deletion so SPDK actually coalesces the - clusters into the child bdev and frees them. Deletion proceeds oldest-first - so each snapshot has at most one child when it is removed, satisfying SPDK's - snapshot-deletion constraint. + Delegates to _delete_bdev_blocking(coalescing=True): the intermediate + snapshot's clusters must be merged into its child bdev before being freed + (special_delete=False), or the child's parent chain breaks and + pre-migration data is lost. """ for snap_uuid in migration.intermediate_snaps: try: @@ -1677,16 +1828,18 @@ def _delete_intermediate_snaps_on_target(migration, tgt_rpc, tgt_sec_rpc=None): tgt_composite = snap.snap_bdev # updated to target path by apply_migration_to_db - if tgt_rpc.get_bdevs(tgt_composite): + if not tgt_rpc.get_bdevs(tgt_composite): + logger.info( + f"Intermediate snap bdev {tgt_composite} absent from target; skipping SPDK delete") + else: try: - _delete_bdev_blocking(tgt_composite, tgt_rpc, secondary_rpc=tgt_sec_rpc) + _delete_bdev_blocking(tgt_composite, tgt_rpc, + secondary_rpc=tgt_sec_rpc, tertiary_rpc=tgt_ter_rpc, + coalescing=True) logger.info(f"Deleted intermediate snap bdev {tgt_composite} from target") except Exception as e: logger.warning( f"Could not delete intermediate snap {tgt_composite} from target: {e}") - else: - logger.info( - f"Intermediate snap bdev {tgt_composite} absent from target; skipping SPDK delete") try: snap.remove(db.kv_store) @@ -1695,22 +1848,151 @@ def _delete_intermediate_snaps_on_target(migration, tgt_rpc, tgt_sec_rpc=None): logger.warning(f"Could not remove intermediate snap {snap_uuid} from DB: {e}") -def _get_secondary_rpc(node): - """Return RPC client for node's secondary if reachable (online or suspended).""" - if not node.secondary_node_id: - return None - try: - sec = db.get_storage_node_by_id(node.secondary_node_id) - if sec.status in (StorageNode.STATUS_ONLINE, StorageNode.STATUS_SUSPENDED): - return _make_rpc(sec) - except KeyError: - pass - return None +def _rename_migrated_bdevs(migration, tgt_node, tgt_rpc, tgt_sec_rpc=None, tgt_ter_rpc=None): + """ + After migration completes, rename 'm'-suffixed bdevs on the target back to + their canonical names (without the suffix). This prevents suffix accumulation + on repeated migrations of the same volume. + + bdev_lvol_rename returns JSON-RPC error -32602 "File exists" (None to caller) + when the target name is already taken; we use that to try the fallback + (_MIGRATION_BDEV_SUFFIX_DONE) instead of doing a pre-query of all bdevs. + Must be called AFTER _apply_migration_to_db() — snap.snap_bdev and + lvol.lvol_bdev already point to the 'm'-suffixed target paths at that point. + """ + lvstore = tgt_node.lvstore + preexisting = set(migration.snaps_preexisting_on_target or []) + + _EXISTS = "EXISTS" + + def _do_rename(old_composite, new_short, label): + """Rename on prim + sec + ter. Returns 'EXISTS' if the target name is + already taken (SPDK returns JSON-RPC error -32602 'File exists' -> None), + True on success. new_short must be the short name only (no lvstore prefix).""" + ret = tgt_rpc.bdev_lvol_rename(old_composite, new_short) + logger.debug(f"_do_rename prim: {old_composite!r} -> {new_short!r}: ret={ret!r}") + # SPDK returns None on name collision (-32602 "File exists"); treat any falsy as collision. + exists = (not ret) or (ret == _EXISTS) + for role, rpc in [("sec", tgt_sec_rpc), ("ter", tgt_ter_rpc)]: + if rpc: + try: + r = rpc.bdev_lvol_rename(old_composite, new_short) + logger.debug(f"_do_rename {role}: {old_composite!r} -> {new_short!r}: ret={r!r}") + if (not r) or r == _EXISTS: + exists = True + except Exception as exc: + logger.warning( + f"_rename_migrated_bdevs: {role} rename {label} (non-fatal): {exc}") + if exists: + return _EXISTS + return True + def _rename_with_fallback(current_short, label): + """Try canonical name; on EXISTS try the _done fallback. + Returns the target short name on success, None if skipped.""" + canonical = current_short[:-len(_MIGRATION_BDEV_SUFFIX)] + old = f"{lvstore}/{current_short}" + ret = _do_rename(old, canonical, label) + if ret == _EXISTS: + fallback = canonical + _MIGRATION_BDEV_SUFFIX_DONE + logger.info( + f"_rename_migrated_bdevs: {canonical} exists - trying fallback {fallback}") + ret2 = _do_rename(old, fallback, label) + if ret2 == _EXISTS: + logger.warning( + f"_rename_migrated_bdevs: both {canonical} and {fallback} " + f"exist - leaving {current_short} as-is") + return None + target = fallback + else: + target = canonical -_SKIP_CLEANUP_SOURCE = False # DEBUG: set True to skip source cleanup + logger.info(f"_rename_migrated_bdevs: {current_short} -> {target}") + return target + + # --- Snapshots (owned) --- + for snap_uuid in migration.snaps_migrated: + if snap_uuid in preexisting: + continue + try: + snap = db.get_snapshot_by_id(snap_uuid) + except KeyError: + continue + + snap_bdev = snap.snap_bdev + if '/' not in snap_bdev: + continue + current_short = snap_bdev.split('/', 1)[1] + if not current_short.endswith(_MIGRATION_BDEV_SUFFIX): + continue + + if snap.lvol.uuid != migration.lvol_id: + continue # ancestor snap — handled below + + target = _rename_with_fallback(current_short, current_short) + if target: + snap.snap_bdev = f"{lvstore}/{target}" + snap.write_to_db(db.kv_store) + + # --- Ancestor chain blobs (non-owned snaps) --- + # _apply_migration_to_db added an instances entry with the _m bdev name. + # Rename it and update the entry in place. + for snap_uuid in migration.snaps_migrated: + if snap_uuid in preexisting: + continue + try: + snap = db.get_snapshot_by_id(snap_uuid) + except KeyError: + continue + if snap.lvol.uuid == migration.lvol_id: + continue # owned snap — handled above + + updated = False + for inst in snap.instances: + inst_bdev = inst.get('snap_bdev', '') + if '/' not in inst_bdev: + continue + inst_lvstore, inst_short = inst_bdev.split('/', 1) + if inst_lvstore != lvstore: + continue + if not inst_short.endswith(_MIGRATION_BDEV_SUFFIX): + continue + + try: + target = _rename_with_fallback(inst_short, inst_short) + except Exception as exc: + logger.warning( + f"_rename_migrated_bdevs: ancestor {inst_short} failed: {exc}") + continue + if target: + inst['snap_bdev'] = f"{lvstore}/{target}" + updated = True + + if updated: + snap.write_to_db(db.kv_store) + + # --- Lvol --- + try: + lvol = db.get_lvol_by_id(migration.lvol_id) + except KeyError: + logger.warning(f"_rename_migrated_bdevs: lvol {migration.lvol_id} not found") + return + + current_lvol_short = lvol.lvol_bdev + if not current_lvol_short.endswith(_MIGRATION_BDEV_SUFFIX): + return + + target = _rename_with_fallback(current_lvol_short, current_lvol_short) + if target: + lvol.lvol_bdev = target + lvol.top_bdev = f"{lvstore}/{target}" + for entry in lvol.bdev_stack: + if (entry.get('type') == 'bdev_lvol' + and entry.get('params', {}).get('name') == current_lvol_short): + entry['params']['name'] = target + lvol.write_to_db(db.kv_store) def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): @@ -1726,26 +2008,8 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): Returns (done: bool, suspend: bool, error: str|None). """ - if _SKIP_CLEANUP_SOURCE: - logger.info("SKIP_CLEANUP_SOURCE flag is set — skipping source cleanup, applying DB only") - _ctx = migration.transfer_context or {} - _apply_migration_to_db( - migration, - tgt_lvol_uuid=_ctx.get('tgt_lvol_uuid'), - tgt_lvol_bdev=_ctx.get('tgt_lvol_bdev')) - if migration.intermediate_snaps: - tgt_sec_rpc = _get_secondary_rpc(tgt_node) - _delete_intermediate_snaps_on_target(migration, tgt_rpc, tgt_sec_rpc) - return True, False, None - ctx = migration.transfer_context or {} - # --- Cutover grace period --- - # Wait here until the client has had time to reconnect to the new TGT - # subsystem before we tear down the source subsystem. The grace period - # begins when PHASE_LVOL_MIGRATE sets cutover_notified_at and changes - # migration.status to STATUS_CUTOVER — the test script polls for that - # status and triggers client reconnect immediately upon seeing it. # --- First entry: initialize cleanup state --- if ctx.get('stage') != 'cleanup_src': # Preserve the target lvol UUID and bdev name written by PHASE_LVOL_MIGRATE @@ -1756,21 +2020,34 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): # bdevs even after apply_migration_to_db renamed them to target names in DB. source_snap_bdevs_saved = ctx.get('source_snap_bdevs', {}) source_lvol_bdev_saved = ctx.get('source_lvol_bdev', '') + if not source_lvol_bdev_saved: + return False, False, ( + "source_lvol_bdev missing from transfer_context — cannot safely " + "identify source bdev; apply_migration_to_db may have already " + "renamed lvol.lvol_bdev to the target name") to_delete = migration_controller.get_snaps_safe_to_delete_on_source(migration) # Verify each snapshot to be deleted physically exists on the target - # before we remove anything from the source. Target bdevs carry the - # migration suffix (e.g. SNAP_xxxm) — derive from source short name. + # before we remove anything from the source. Bdevs may carry the + # migration suffix (SNAP_Xm), canonical name (SNAP_X), or the + # post-rename fallback (SNAP_Xam) — check all three so that a + # crash-recovery re-run after a partial rename still passes. tgt_lvols = tgt_rpc.bdev_lvol_get_lvols(tgt_node.lvstore) or [] tgt_names = {e.get('name', '').split('/')[-1] for e in tgt_lvols} for snap_uuid in to_delete: try: snap = db.get_snapshot_by_id(snap_uuid) - snap_short_tgt = _snap_tgt_short_name(snap) - if snap_short_tgt not in tgt_names: + # snap.snap_bdev was updated to the target path by apply_migration_to_db; + # use it as the primary lookup, then fall back to derived names. + _snap_bdev = snap.snap_bdev or '' + _primary = _snap_bdev.split('/', 1)[1] if '/' in _snap_bdev else _snap_bdev + _m_name = _snap_tgt_short_name(snap) + _canonical = _snap_short_name(snap) + _am_name = _canonical + _MIGRATION_BDEV_SUFFIX_DONE + if not any(n in tgt_names for n in (_primary, _m_name, _canonical, _am_name)): return False, False, ( - f"Target missing snapshot {snap_short_tgt} ({snap_uuid}) " + f"Target missing snapshot {_m_name} ({snap_uuid}) " "— aborting source cleanup to prevent data loss" ) except KeyError: @@ -1789,7 +2066,10 @@ 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_rpc = _get_secondary_rpc(src_node) + src_sec = _get_source_secondary_node(src_node) + src_sec_rpc = _make_rpc(src_sec) if src_sec else None + src_ter = _get_source_tertiary_node(src_node) + src_ter_rpc = _make_rpc(src_ter) if src_ter else None # --- Delete source snapshots (blocking, idempotent) --- # Recompute to_delete here so re-entries after crash work without stored pending list. @@ -1804,13 +2084,12 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): bdev_name = source_snap_bdevs[snap_uuid] else: bdev_name = f"{src_node.lvstore}/{_snap_short_name(snap)}" - # Mark as migration-source so SPDK drops only the blob reference - # without freeing the physical clusters (data lives on the target now). - src_rpc.bdev_lvol_set_migration_flag(bdev_name) - if src_sec_rpc: - src_sec_rpc.bdev_lvol_set_migration_flag(bdev_name) - _delete_bdev_blocking(bdev_name, src_rpc, secondary_rpc=src_sec_rpc) - logger.info(f"Deleted source bdev {bdev_name}") + try: + _delete_bdev_blocking(bdev_name, src_rpc, + secondary_rpc=src_sec_rpc, tertiary_rpc=src_ter_rpc) + logger.info(f"Deleted source bdev {bdev_name}") + except Exception as e: + logger.warning(f"delete source bdev {bdev_name} (non-fatal): {e}") except KeyError: logger.warning(f"Source snapshot {snap_uuid} not found in DB; skipping") @@ -1822,15 +2101,6 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): # bdev_nvme_detach_controller can still reach the hub bdev. # Step 8: delete source primary NVMe-oF subsystem. # Then: delete source lvol bdev. - # TODO: reenable hub detach after migration — currently disabled to keep hub controller attached for debugging. - # TEMP: keep hub controller attached after migration - # hub_ctrl_name = ctx.get('hub_ctrl_name') - # if hub_ctrl_name: - # try: - # src_rpc.bdev_nvme_detach_controller(hub_ctrl_name) - # logger.info(f"Step 7: deferred hub controller detach: {hub_ctrl_name}") - # except Exception as e: - # logger.warning(f"Deferred hub detach {hub_ctrl_name} (non-fatal): {e}") lvol = None try: lvol = db.get_lvol_by_id(migration.lvol_id) @@ -1856,20 +2126,27 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): try: # Use the saved pre-apply bdev name; apply_migration_to_db already # renamed lvol.lvol_bdev to the target 'm'-suffix name in DB. - _raw_src_bdev = ctx.get('source_lvol_bdev') - src_bdev_short = _raw_src_bdev if isinstance(_raw_src_bdev, str) and _raw_src_bdev else lvol.lvol_bdev + src_bdev_short = ctx.get('source_lvol_bdev') + if not src_bdev_short: + raise RuntimeError( + "source_lvol_bdev missing from ctx at delete site — " + "refusing to fall back to lvol.lvol_bdev which may point to target") src_lvol_composite = f"{src_node.lvstore}/{src_bdev_short}" # Set migration flag so SPDK drops only the blob reference without # freeing the physical clusters — data now lives on the target. try: src_rpc.bdev_lvol_set_migration_flag(src_lvol_composite) + if src_sec_rpc: + src_sec_rpc.bdev_lvol_set_migration_flag(src_lvol_composite) + if src_ter_rpc: + src_ter_rpc.bdev_lvol_set_migration_flag(src_lvol_composite) except Exception as _mf_err: logger.warning( f"bdev_lvol_set_migration_flag for source lvol failed " f"(non-fatal): {_mf_err}") _delete_bdev_blocking( src_lvol_composite, src_rpc, - secondary_rpc=src_sec_rpc) + secondary_rpc=src_sec_rpc, tertiary_rpc=src_ter_rpc) logger.info(f"Deleted source lvol bdev {src_lvol_composite}") except Exception as e: logger.warning(f"Source lvol delete failed (non-fatal): {e}") @@ -1881,12 +2158,20 @@ def _handle_cleanup_source(migration, src_node, src_rpc, tgt_node, tgt_rpc): migration, tgt_lvol_uuid=tgt_lvol_uuid, tgt_lvol_bdev=tgt_lvol_bdev): return False, False, "Failed to update DB records after source cleanup" + tgt_sec, _ = _get_target_secondary_node(tgt_node, src_node.get_id()) + tgt_sec_rpc = _make_rpc(tgt_sec) if tgt_sec else None + tgt_ter, _ = _get_target_tertiary_node(tgt_node, src_node.get_id()) + tgt_ter_rpc = _make_rpc(tgt_ter) if tgt_ter else None + # Delete intermediate (shrink) snapshots from the target — they are migration # artifacts and do not need to be preserved. No migration flag so SPDK # coalesces and frees their clusters into the child bdev. - if migration.intermediate_snaps: - tgt_sec_rpc = _get_secondary_rpc(tgt_node) - _delete_intermediate_snaps_on_target(migration, tgt_rpc, tgt_sec_rpc) + try: + if migration.intermediate_snaps: + _delete_intermediate_snaps_on_target(migration, tgt_rpc, tgt_sec_rpc, tgt_ter_rpc) + _rename_migrated_bdevs(migration, tgt_node, tgt_rpc, tgt_sec_rpc, tgt_ter_rpc) + except Exception as e: + return False, False, str(e) return True, False, None @@ -1902,20 +2187,15 @@ def _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=None): Returns (done: bool, suspend: bool, error: str|None). """ - # Detach the shared hub controller on the source (best-effort). - # The hub is attached once during SNAP_COPY and must be torn down on any - # failure/cancel path; if the source is unreachable this is non-fatal. - if src_rpc is not None: - hub_ctrl_name = tgt_node.transfer_hublvol.bdev_name - try: - if src_rpc.get_bdevs(f"{hub_ctrl_name}n1"): - src_rpc.bdev_nvme_detach_controller(hub_ctrl_name) - logger.info(f"cleanup_target: detached hub controller {hub_ctrl_name}") - except Exception as e: - logger.warning(f"cleanup_target: hub detach {hub_ctrl_name} (non-fatal): {e}") + # 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) ctx = migration.transfer_context or {} - tgt_sec_rpc = _get_secondary_rpc(tgt_node) + tgt_sec, _ = _get_target_secondary_node(tgt_node, migration.source_node_id) + tgt_sec_rpc = _make_rpc(tgt_sec) if tgt_sec else 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 # --- 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 @@ -1931,7 +2211,7 @@ def _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=None): _pre_nqn: Optional[str] = None try: _lvol = db.get_lvol_by_id(migration.lvol_id) - _pre_bdev = f"{tgt_node.lvstore}/{_lvol.lvol_bdev}{_MIGRATION_BDEV_SUFFIX}" + _pre_bdev = f"{tgt_node.lvstore}/{_lvol_tgt_bdev_name(_lvol.lvol_bdev)}" _pre_nqn = _lvol.nqn except Exception: _pre_bdev = None @@ -1944,18 +2224,20 @@ def _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=None): _cleanup_subsystem_or_ns(_nqn_to_clean, tgt_ns_id, subsystem_created_on_target, tgt_rpc) except Exception as e: logger.warning(f"cleanup target subsystem {_nqn_to_clean}: {e}") - if tgt_sec_rpc: - try: - _cleanup_subsystem_or_ns(_nqn_to_clean, tgt_ns_id, subsystem_created_on_target, - tgt_sec_rpc) - except Exception as e: - logger.warning(f"cleanup target secondary subsystem {_nqn_to_clean}: {e}") + for _label, _extra_rpc in [("secondary", tgt_sec_rpc), ("tertiary", tgt_ter_rpc)]: + if _extra_rpc: + try: + _cleanup_subsystem_or_ns(_nqn_to_clean, tgt_ns_id, + subsystem_created_on_target, _extra_rpc) + except Exception as e: + logger.warning(f"cleanup target {_label} subsystem {_nqn_to_clean}: {e}") # Delete target migration bdev — prefer ctx composite, fall back to derived name. _bdev_to_delete = tgt_lvol_composite or _pre_bdev if _bdev_to_delete and tgt_rpc.get_bdevs(_bdev_to_delete): try: - _delete_bdev_blocking(_bdev_to_delete, tgt_rpc, secondary_rpc=tgt_sec_rpc) + _delete_bdev_blocking(_bdev_to_delete, tgt_rpc, + secondary_rpc=tgt_sec_rpc, tertiary_rpc=tgt_ter_rpc) logger.info(f"Deleted target lvol {_bdev_to_delete}") except Exception as e: logger.warning(f"delete target lvol {_bdev_to_delete} (non-fatal): {e}") @@ -1968,16 +2250,33 @@ def _handle_cleanup_target(migration, tgt_node, tgt_rpc, src_rpc=None): # Reverse order: children/leaves before parents/roots (SPDK open-ref constraint). # _delete_bdev_blocking handles "not found" (status 2) gracefully, so a # crash-recovery re-run that re-deletes already-removed bdevs is safe. - for snap_uuid in reversed(migration.snaps_migrated): + # Uses get_snaps_to_delete_on_target to skip pre-existing snaps and snaps + # still referenced by other lvols already on the target node. + for snap_uuid in reversed(migration_controller.get_snaps_to_delete_on_target(migration)): try: snap = db.get_snapshot_by_id(snap_uuid) - snap_short = _snap_tgt_short_name(snap) - bdev_name = f"{tgt_node.lvstore}/{snap_short}" - if not tgt_rpc.get_bdevs(bdev_name): - logger.info(f"Target bdev {bdev_name} not found; skipping (already cleaned up)") + # Try all possible bdev name variants: in-flight (m), canonical, am-fallback. + # After a partial rename (crash mid-cleanup) the bdev may have been + # renamed before the rollback was triggered. + _lvstore = tgt_node.lvstore + _m_name = _snap_tgt_short_name(snap) + _canonical = _snap_short_name(snap) + _am_name = _canonical + _MIGRATION_BDEV_SUFFIX_DONE + bdev_name = next( + (f"{_lvstore}/{n}" for n in (_m_name, _canonical, _am_name) + if tgt_rpc.get_bdevs(f"{_lvstore}/{n}")), + None) + if not bdev_name: + logger.info( + f"Target bdev {_lvstore}/{_m_name} not found in any variant; " + f"skipping (already cleaned up)") continue - _delete_bdev_blocking(bdev_name, tgt_rpc, secondary_rpc=tgt_sec_rpc) - logger.info(f"Deleted target snapshot bdev {bdev_name}") + try: + _delete_bdev_blocking(bdev_name, tgt_rpc, + secondary_rpc=tgt_sec_rpc, tertiary_rpc=tgt_ter_rpc) + logger.info(f"Deleted target snapshot bdev {bdev_name}") + except Exception as e: + logger.warning(f"delete target snapshot bdev {bdev_name} (non-fatal): {e}") except KeyError: logger.warning(f"Target snapshot {snap_uuid} not found in DB; skipping") diff --git a/simplyblock_core/utils/__init__.py b/simplyblock_core/utils/__init__.py index 4199eba8d..4b340d65b 100644 --- a/simplyblock_core/utils/__init__.py +++ b/simplyblock_core/utils/__init__.py @@ -375,6 +375,19 @@ def sum_records(records): return total +def lvol_tgt_bdev_name(lvol_bdev: str) -> str: + """Return the migration-target bdev short name for a writable lvol. + + Idempotent: strips any existing migration suffix before adding one, so + retries or back-to-back migrations (where lvol_bdev already ends in the + suffix from a previous run) never accumulate it (e.g. 'LVOL_Xmm'). + """ + suffix = constants.LVOL_MIG_BDEV_SUFFIX + if lvol_bdev.endswith(suffix): + lvol_bdev = lvol_bdev[:-len(suffix)] + return lvol_bdev + suffix + + def _used_bdev_name_numbers(db_controller, all_lvols=None, all_snapshots=None): used = set() if not all_lvols: