diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 000000000..dfdb8b771 --- /dev/null +++ b/.gitattributes @@ -0,0 +1 @@ +*.sh text eol=lf diff --git a/simplyblock_core/constants.py b/simplyblock_core/constants.py index 911b63c3e..b606e9aa8 100644 --- a/simplyblock_core/constants.py +++ b/simplyblock_core/constants.py @@ -369,6 +369,10 @@ def get_config_var(name, default=None): # (snapshot -> wait replicated -> snapshot -> wait) before the final freeze. # Two rounds normally complete within 2 replication intervals + transfer time. REPL_CUTOVER_SHRINK_TIMEOUT_SEC = 900 +# Safety timeout for the operator preconnect signal. The task suspends indefinitely +# waiting for POST .../replication/cutover-proceed; this is the fallback deadline +# if the operator is unavailable. Cutover proceeds regardless after this many seconds. +REPL_CUTOVER_PROCEED_TIMEOUT_SEC = 120 SPDK_PROXY_MULTI_THREADING_ENABLED=True SPDK_PROXY_TIMEOUT=60*5 diff --git a/simplyblock_core/controllers/lvol_controller.py b/simplyblock_core/controllers/lvol_controller.py index 10c8a02ac..7891fa61f 100644 --- a/simplyblock_core/controllers/lvol_controller.py +++ b/simplyblock_core/controllers/lvol_controller.py @@ -1132,7 +1132,8 @@ def _lvol_secondary_index(lvol, node): return max(_lvol_path_index(lvol, node) - 1, 0) -def add_lvol_on_node(lvol, snode, is_primary=True, secondary_index=0): +def add_lvol_on_node(lvol, snode, is_primary=True, secondary_index=0, min_cntlid=None, ns_uuid=None, + primary_nsid=None): rpc_client = snode.rpc_client() # Refuse to attach a new namespace to a shared subsystem while any @@ -1165,7 +1166,8 @@ def add_lvol_on_node(lvol, snode, is_primary=True, secondary_index=0): return _fail_after_bdev(lvol, rpc_client, str(e)) if resolve_subsys: - min_cntlid = lvol_min_cntlid(0 if is_primary else secondary_index + 1) + if min_cntlid is None: + min_cntlid = lvol_min_cntlid(0 if is_primary else secondary_index + 1) allow_any = not bool(lvol.allowed_hosts) logger.info("creating subsystem %s (allow_any_host=%s)", lvol.nqn, allow_any) ret = rpc_client.subsystem_create(lvol.nqn, lvol.ha_type, lvol.uuid, min_cntlid, @@ -1275,7 +1277,13 @@ def add_lvol_on_node(lvol, snode, is_primary=True, secondary_index=0): # nsid (ns_id == 0, e.g. a drain-queued registration firing early) # must fail loudly instead of guessing. if is_primary: - requested_nsid = None + # primary_nsid is set by migration/failover callers to preserve the + # source cluster's nsid assignment. The client connects to both source + # and target under the same NQN during preconnect; if nsid positions + # differ, the kernel rejects the target namespaces as mismatched + # ("IDs don't match for shared namespace N"). Regular creates pass + # primary_nsid=None (auto-assign, the previous behaviour). + requested_nsid = primary_nsid else: if not lvol.ns_id: return _fail_after_bdev( @@ -1285,28 +1293,62 @@ def add_lvol_on_node(lvol, snode, is_primary=True, secondary_index=0): f"maps across the shared subsystem's paths)") requested_nsid = lvol.ns_id ret, err = rpc_client.nvmf_subsystem_add_ns2( - lvol.nqn, lvol.top_bdev, lvol.uuid, lvol.guid, nsid=requested_nsid) - if err: - if err and err["code"] == -32602 and lvol.namespace and lvol.node_id == snode.get_id(): - logger.info("Error adding namespace to subsystem, finding new subsystem for namespaced lvol") - # Re-claim transactionally, excluding the subsystem SPDK just - # rejected (the DB count said it had room — SPDK is the authority - # on its own namespace table). The lvol's record is rewritten in - # the same transaction, so its slot moves atomically from the - # rejected subsystem to the new one (or to a standalone one). - cluster = DBController().get_cluster_by_id(snode.cluster_id) - try: - DBController().claim_lvol_ns_slot( - lvol, snode, True, - standalone_nqn=cluster.nqn + ":lvol:" + lvol.uuid, - exclude_nqns={lvol.nqn}) - except SubsystemCapacityError as e: - logger.error(str(e)) - return _fail_after_bdev(lvol, rpc_client, str(e)) - return add_lvol_on_node(lvol, snode, is_primary=is_primary, secondary_index=secondary_index) - else: - return _fail_after_bdev( - lvol, rpc_client, "Failed to add bdev to subsystem") + lvol.nqn, lvol.top_bdev, ns_uuid or lvol.uuid, lvol.guid, nsid=requested_nsid) + if err: + if err["code"] == -32602 and lvol.namespace: + # -32602 from nvmf_subsystem_add_ns has two distinct causes: + # A) subsystem exists on this node but the nsid slot is occupied + # B) subsystem does not exist on this node at all (cross-LVS: + # the shared subsystem's LVS has no nodes in common with this + # lvol's LVS — e.g. a failover clone on LVS_6 whose shared + # subsystem lives on LVS_1) + # nvmf_subsystem_add_ns2 already ran the idempotency probe and + # returned the original error, so subsystem_get here distinguishes A/B. + subsys_here = rpc_client.subsystem_get(lvol.nqn) + if subsys_here is None: + # Case B — cross-LVS: subsystem not on this node. + # Primary: the namespace cannot be registered from this LVS; + # log a warning so the operator can investigate placement. + # Secondary: the primary already registered it on the subsystem's + # nodes; nothing more for this secondary to do. + # Either way, skip rather than deleting the bdev and retrying forever. + logger.warning( + "%s node %s: subsystem %s not present on this node " + "(cross-LVS namespaced lvol %s); skipping namespace add.", + "Primary" if is_primary else "Secondary", + snode.get_id(), lvol.nqn, lvol.get_id()) + return {'uuid': lvol.lvol_uuid, + 'driver_specific': {'lvol': {'blobid': lvol.blobid}}}, None + + # Case A — subsystem IS on this node; the nsid slot is occupied. + if lvol.node_id == snode.get_id(): + if primary_nsid is not None: + # Migration/failover: re-claiming to a different subsystem + # would lose the shared NQN. _evict_stale_namespace should + # have cleared the occupant; a persistent -32602 here means + # genuine unexpected state — fail hard. + return _fail_after_bdev( + lvol, rpc_client, + f"Failed to add bdev to subsystem at requested nsid={primary_nsid}: " + f"nsid occupied and eviction did not clear it") + logger.info("Error adding namespace to subsystem, finding new subsystem for namespaced lvol") + # Re-claim transactionally, excluding the subsystem SPDK just + # rejected (the DB count said it had room — SPDK is the authority + # on its own namespace table). The lvol's record is rewritten in + # the same transaction, so its slot moves atomically from the + # rejected subsystem to the new one (or to a standalone one). + cluster = DBController().get_cluster_by_id(snode.cluster_id) + try: + DBController().claim_lvol_ns_slot( + lvol, snode, True, + standalone_nqn=cluster.nqn + ":lvol:" + lvol.uuid, + exclude_nqns={lvol.nqn}) + except SubsystemCapacityError as e: + logger.error(str(e)) + return _fail_after_bdev(lvol, rpc_client, str(e)) + return add_lvol_on_node(lvol, snode, is_primary=is_primary, secondary_index=secondary_index) + return _fail_after_bdev( + lvol, rpc_client, "Failed to add bdev to subsystem") if is_primary: # Persist the target-assigned nsid; replicas re-add with exactly @@ -3341,6 +3383,10 @@ def _create_target_lvol_clone(db_controller, lvol, target_node, pool_uuid, snaps new_lvol.top_bdev = f"{new_lvol.lvs_name}/{new_lvol.lvol_bdev}" new_lvol.snapshot_name = snapshot.snap_bdev new_lvol.status = LVol.STATUS_IN_CREATION + # The source subsys_port is inherited from the deep copy but the target + # node may use a different per-lvstore port. Update it now so that + # suspend_lvol (and any future ANA flip) addresses the right listener. + new_lvol.subsys_port = target_node.get_lvol_subsys_port(target_node.lvstore) # Preserve the ORIGINAL subsystem NQN and namespace id: the client must # reconnect to the SAME NQN/NS on the target cluster — only the IP/port # differ. new_lvol is a deep copy of lvol, so nqn/ns_id are already @@ -3362,7 +3408,35 @@ def _create_target_lvol_clone(db_controller, lvol, target_node, pool_uuid, snaps new_lvol.write_to_db(db_controller.kv_store) - lvol_bdev, error = add_lvol_on_node(new_lvol, target_node) + _evict_stale_namespace(new_lvol, target_node) + + # The target clone shares the source NQN. During preconnect the host has + # live paths to the source (cntlids 1, 1000, 2000) AND tries to add paths + # to the inaccessible target simultaneously. If target uses the same + # min_cntlid the kernel rejects it as a duplicate cntlid. Use windows + # above 4000 so source (1/1000/2000) and target never collide. + _tgt_cntlids = [ + random.randint(4001, 4500), # primary + random.randint(5001, 5500), # secondary + random.randint(6001, 6500), # tertiary + ] + + # Preserve the source lvol's UUID as the NVMe namespace UUID so the kernel + # can recognise target paths as belonging to the same multipath namespace. + # new_lvol.uuid is a fresh DB key and must NOT be used as the namespace UUID. + _src_ns_uuid = lvol.uuid + + # For migration/failover, preserve the source nsid so the kernel can + # match target paths to source paths under the same NQN. new_lvol is a + # deepcopy of the source lvol, so new_lvol.ns_id is already the source + # nsid. Passing it explicitly prevents auto-assignment from choosing a + # different position on the target (concurrent migration tasks for + # sibling namespaces would arrive in arbitrary order, diverging the nsid + # map from the source and triggering "IDs don't match for shared + # namespace N" in the client kernel during preconnect). + lvol_bdev, error = add_lvol_on_node(new_lvol, target_node, + min_cntlid=_tgt_cntlids[0], ns_uuid=_src_ns_uuid, + primary_nsid=new_lvol.ns_id) if error: logger.error(error) db_controller.release_lvol_ns_slot(new_lvol) @@ -3373,16 +3447,22 @@ def _create_target_lvol_clone(db_controller, lvol, target_node, pool_uuid, snaps # Expose the volume on the secondary and tertiary target nodes too (HA), # so connect_lvol returns all client paths. + _tgt_cntlid_iter = iter(_tgt_cntlids[1:]) for peer_id in [target_node.secondary_node_id, target_node.tertiary_node_id]: if not peer_id: + next(_tgt_cntlid_iter, None) continue try: peer_node = db_controller.get_storage_node_by_id(peer_id) except KeyError: + next(_tgt_cntlid_iter, None) continue if peer_node.status != StorageNode.STATUS_ONLINE: + next(_tgt_cntlid_iter, None) continue - lvol_bdev, error = add_lvol_on_node(new_lvol, peer_node, is_primary=False) + _peer_cntlid = next(_tgt_cntlid_iter, None) + lvol_bdev, error = add_lvol_on_node(new_lvol, peer_node, is_primary=False, + min_cntlid=_peer_cntlid, ns_uuid=_src_ns_uuid) if error: logger.error(error) # remove lvol from primary @@ -3445,6 +3525,42 @@ def _last_replicated_target_snapshot(db_controller, lvol_id, cluster_id): return None +def _evict_stale_namespace(new_lvol, target_node): + """Make room for the preserved identity on a RECOVERED fail-back target. + + The cutover clone keeps the ORIGINAL volume's NQN and nsid so the client + reconnects to the same identity. Failing back to a recovered source means + that subsystem usually still exists there WITH the original volume's + namespace at exactly that nsid -- and its data is outdated by definition + (the other cluster served every write since the fail-over). add_ns then + fails with -32602 for every retry, and the whole cutover dies on max + retry: 2026-08-24, 5/5 fail-back cutovers, 40x "Failed to add bdev to + subsystem". Evict a namespace occupying the clone's nsid unless it is + already the clone's own bdev (idempotent re-run). + """ + try: + rpc = target_node.rpc_client() + subsystems = rpc.subsystem_get(new_lvol.nqn) + if not subsystems: + return + for ns in (subsystems.get("namespaces") or []): + if ns.get("nsid") != new_lvol.ns_id: + continue + if ns.get("bdev_name") == new_lvol.top_bdev: + return # already ours (re-run) + logger.info( + f"Fail-back cutover: evicting stale namespace nsid={ns.get('nsid')} " + f"(bdev {ns.get('bdev_name')}) from {new_lvol.nqn} on " + f"{target_node.get_id()} -- superseded by the failed-over data") + rpc.nvmf_subsystem_remove_ns(new_lvol.nqn, ns.get("nsid")) + return + except Exception as e: + # Best effort: if the subsystem is not there, add_lvol_on_node creates + # it; if the eviction genuinely failed, add_ns will say so loudly. + logger.warning(f"Stale-namespace check on {target_node.get_id()} for " + f"{new_lvol.nqn} raised: {e}") + + def _clone_from_last_replicated(db_controller, lvol_id, lvol, target_node, pool_uuid, cluster_id, attempts=3): """Pick the last fully replicated target snapshot and clone from it ATOMICALLY. diff --git a/simplyblock_core/controllers/replication_policy_controller.py b/simplyblock_core/controllers/replication_policy_controller.py index 12a868e6b..8dd47353a 100644 --- a/simplyblock_core/controllers/replication_policy_controller.py +++ b/simplyblock_core/controllers/replication_policy_controller.py @@ -359,6 +359,24 @@ def _failover_volumes(volumes, what): return results +def set_cutover_proceed(lvol_id): + """Signal that the operator has connected the target NVMe paths. + + Finds the cutover_pending LVolReplication for *lvol_id* (source side) and + sets cutover_proceed = True so the task runner advances past the wait. + + Returns the replication ID on success, raises KeyError when no matching + cutover_pending record is found. + """ + rep = _active_relationship(lvol_id) + if rep is None or rep.state != LVolReplication.STATE_CUTOVER_PENDING: + raise KeyError( + f"No cutover_pending replication found for volume {lvol_id}") + rep.cutover_proceed = True + rep.write_to_db(db.kv_store) + return rep.get_id() + + def get_relationship(lvol_id): """The replication relationship of *lvol_id*, source or target side. diff --git a/simplyblock_core/models/lvol_model.py b/simplyblock_core/models/lvol_model.py index 29ca15c6b..0b0eeb87d 100644 --- a/simplyblock_core/models/lvol_model.py +++ b/simplyblock_core/models/lvol_model.py @@ -144,6 +144,10 @@ class LVolReplication(BaseModel): # client keeps the same NQN/namespace across fail-over and migration. target_nqn: str = "" target_ns_id: int = 0 + # Set to True by POST .../replication/cutover-proceed once the operator has + # connected the target NVMe paths. The task runner waits for this before + # calling run_cutover(); REPL_CUTOVER_PROCEED_TIMEOUT_SEC is the safety fallback. + cutover_proceed: bool = False class LVolMini(BaseModel): lvol_uuid: str = "" diff --git a/simplyblock_core/services/tasks_runner_replication_final.py b/simplyblock_core/services/tasks_runner_replication_final.py index 6e2183945..f2fbe710b 100644 --- a/simplyblock_core/services/tasks_runner_replication_final.py +++ b/simplyblock_core/services/tasks_runner_replication_final.py @@ -166,6 +166,31 @@ def task_runner(task: JobSchedule): return _finalize(task, False, err) params = task.function_params + # Wait for the operator to signal that target NVMe paths are connected + # (operator calls POST .../replication/cutover-proceed after its preconnect + # Job succeeds). REPL_CUTOVER_PROCEED_TIMEOUT_SEC is the safety fallback + # so cutover proceeds even if the operator is unavailable. + replication_id = params.get("replication_id") + if replication_id: + try: + rep = db.get_lvol_replication_by_id(replication_id) + if not rep.cutover_proceed: + if "cutover_proceed_timeout" not in params: + params["cutover_proceed_timeout"] = ( + int(time.time()) + constants.REPL_CUTOVER_PROCEED_TIMEOUT_SEC) + task.write_to_db(db.kv_store) + if int(time.time()) < params["cutover_proceed_timeout"]: + task.function_result = "cutover_pending: waiting for preconnect signal" + task.status = JobSchedule.STATUS_SUSPENDED + task.write_to_db(db.kv_store) + return False + logger.warning( + "cutover proceed timeout for replication %s; proceeding without signal", + replication_id) + except KeyError: + logger.warning( + "replication record %s not found; proceeding with cutover", replication_id) + try: ok, err = replication_final_step.run_cutover( src_node, tgt_node, lvol, diff --git a/simplyblock_core/test/test_replication_chain_completeness.py b/simplyblock_core/test/test_replication_chain_completeness.py index 0d7e607a3..c31691090 100644 --- a/simplyblock_core/test/test_replication_chain_completeness.py +++ b/simplyblock_core/test/test_replication_chain_completeness.py @@ -263,3 +263,71 @@ def test_no_backward_task_for_a_policy_managed_clone(): assert gate in src, "the to-source enqueue must be gated on no forward policy" assert src.index(gate) < src.index("replicate_to_source=True"), \ "the gate must guard the to-source enqueue" + + +# --- fail-back cutover: the preserved identity must evict the stale ns ------ + + +class _EvictRPC: + def __init__(self, namespaces): + self.removed = [] + self._ns = namespaces + + def subsystem_get(self, nqn): + return [{"nqn": nqn, "namespaces": self._ns}] + + def nvmf_subsystem_remove_ns(self, nqn, nsid): + self.removed.append((nqn, nsid)) + return True + + +class _EvictNode: + def __init__(self, rpc): + self._rpc = rpc + + def get_id(self): + return "NODE_R" + + def rpc_client(self): + return self._rpc + + +class _CloneLvol: + nqn = "nqn.test:lvol:ORIG" + ns_id = 7 + top_bdev = "LVS_1/LVOL_CLONE" + + +def test_failback_evicts_the_recovered_sources_stale_namespace(): + """2026-08-24: on a recovered source the preserved-NQN subsystem still held + the ORIGINAL volume's namespace at the clone's nsid; add_ns failed -32602 + on every retry and 5/5 fail-back cutovers died on max retry.""" + from simplyblock_core.controllers import lvol_controller as lc + rpc = _EvictRPC([{"nsid": 7, "bdev_name": "LVS_1/LVOL_ORIG"}]) + lc._evict_stale_namespace(_CloneLvol(), _EvictNode(rpc)) + assert rpc.removed == [("nqn.test:lvol:ORIG", 7)] + + +def test_failback_eviction_is_idempotent_for_its_own_namespace(): + from simplyblock_core.controllers import lvol_controller as lc + rpc = _EvictRPC([{"nsid": 7, "bdev_name": "LVS_1/LVOL_CLONE"}]) + lc._evict_stale_namespace(_CloneLvol(), _EvictNode(rpc)) + assert rpc.removed == [] + + +def test_failback_eviction_leaves_other_namespaces_alone(): + from simplyblock_core.controllers import lvol_controller as lc + rpc = _EvictRPC([{"nsid": 3, "bdev_name": "LVS_1/OTHER"}]) + lc._evict_stale_namespace(_CloneLvol(), _EvictNode(rpc)) + assert rpc.removed == [] + + +def test_failback_eviction_tolerates_a_missing_subsystem(): + from simplyblock_core.controllers import lvol_controller as lc + + class _NoSubsysRPC(_EvictRPC): + def subsystem_get(self, nqn): + return None + rpc = _NoSubsysRPC([]) + lc._evict_stale_namespace(_CloneLvol(), _EvictNode(rpc)) # must not raise + assert rpc.removed == [] diff --git a/simplyblock_web/api/v2/cluster/storage_pool/volume/replication.py b/simplyblock_web/api/v2/cluster/storage_pool/volume/replication.py index 339e5ef34..e716ccb8b 100644 --- a/simplyblock_web/api/v2/cluster/storage_pool/volume/replication.py +++ b/simplyblock_web/api/v2/cluster/storage_pool/volume/replication.py @@ -159,6 +159,21 @@ def failback(cluster: Cluster, pool: StoragePool, volume: Volume, body: Failback return Response(status_code=204) +@api.post('/cutover-proceed', name='clusters:storage-pools:volumes:replication:cutover-proceed', + status_code=204, responses={204: {"content": None}}) +def cutover_proceed(cluster: Cluster, pool: StoragePool, volume: Volume) -> Response: + """Signal that target NVMe paths are connected and cutover may proceed. + + Called by the operator after its preconnect Job succeeds. The task runner + is suspended waiting for this signal; once set, it advances to the ANA flip. + """ + try: + replication_policy_controller.set_cutover_proceed(volume.get_id()) + except KeyError as exc: + raise HTTPException(404, str(exc)) + return Response(status_code=204) + + @api.get('/tasks', name='clusters:storage-pools:volumes:replication:tasks') def list_tasks(cluster: Cluster, pool: StoragePool, volume: Volume) -> List[TaskDTO]: return [TaskDTO.from_model(task) for task in lvol_controller.list_replication_tasks(volume.get_id())] diff --git a/tests/unit/test_repair_gating.py b/tests/unit/test_repair_gating.py index 94e6ce806..b66ce75ba 100644 --- a/tests/unit/test_repair_gating.py +++ b/tests/unit/test_repair_gating.py @@ -107,6 +107,52 @@ def test_no_repair_without_the_flag(self): def test_complete_controller_is_left_alone(self): self.assertFalse(self._call(self.BOTH_PATHS, repair_paths=True)) + def test_tertiary_missing_a_secondary_path_is_repaired(self): + """A tertiary connects to primary AND secondary. Three paths (primary + complete, secondary half-attached) must count as a missing path — on + the 2026-08-24 deploy both tertiaries sat at 3/4 forever because + expected_ips was built from the primary alone.""" + from simplyblock_core.controllers import health_controller as hc + from unittest.mock import MagicMock, patch + + def nic(ip): + iface = MagicMock(); iface.trtype = "TCP"; iface.ip4_address = ip + return iface + + primary = MagicMock() + primary.status = StorageNode.STATUS_ONLINE + primary.lvstore_status = "ready" + primary.active_rdma = False; primary.active_tcp = True + primary.data_nics = [nic("10.0.0.1"), nic("10.0.1.1")] + primary.hublvol.bdev_name = "LVS_1/hublvol" + primary.get_id.return_value = "primary-1" + primary.secondary_node_id = "sec-1" + + secondary = MagicMock() + secondary.status = StorageNode.STATUS_ONLINE + secondary.active_rdma = False; secondary.active_tcp = True + secondary.data_nics = [nic("10.0.0.2"), nic("10.0.1.2")] + + node = MagicMock() + node.status = StorageNode.STATUS_ONLINE + node.get_id.return_value = "tert-1" + # tertiary: primary's both paths + only ONE of the secondary's + node.rpc_client.return_value.bdev_nvme_controller_list.return_value = [ + {"ctrlrs": [ + {"state": "enabled", "trid": {"traddr": "10.0.0.1"}}, + {"state": "enabled", "trid": {"traddr": "10.0.1.1"}}, + {"state": "enabled", "trid": {"traddr": "10.0.0.2"}}]}] + node.lvstore_stack_tertiary = "primary-1" # makes is_sec2 True + + coordinator_cls = MagicMock() + with patch.object(hc, "DBController") as db, patch.object(hc, "_restart_owns_lvs", return_value=False), patch("simplyblock_core.utils.hublvol_reconnect." + "HublvolReconnectCoordinator", coordinator_cls): + db.return_value.get_storage_node_by_id.side_effect = lambda i: {"primary-1": primary, "sec-1": secondary}[i] + hc._check_sec_node_hublvol(node, primary_node_id="primary-1", + repair_paths=True) + self.assertTrue(coordinator_cls.return_value.reconcile.called, + "the missing secondary path was not reconciled") + def test_refused_when_the_primary_cannot_answer(self): self.assertFalse(self._call( self.ONE_PATH, primary_status=StorageNode.STATUS_UNREACHABLE,