Replication api v2

docs/replication-policies-design.md

@@ -143,13 +143,14 @@ the input of the deprecated `cluster add-replication`.
 
 Per source cluster:
 
-- `POST/GET/DELETE /clusters/{id}/replication-targets[/{target_id}]`
-- `POST/GET/DELETE /clusters/{id}/replication-policies[/{policy_id}]`
-- `PUT/DELETE /clusters/{c}/storage-pools/{p}/volumes/{v}/replication-policy`
+- `POST/GET/DELETE /clusters/{id}/replication/targets[/{target_id}]`
+- `POST/GET/DELETE /clusters/{id}/replication/policies[/{policy_id}]`
+- `PUT /clusters/{c}/storage-pools/{p}/volumes/{v}` with `replication_policy_id`
+  (`null` detaches)
 
 Two REST defects to fix while touching this surface:
 
-1. `POST .../replication_start` passes the **path** cluster as
+1. `POST .../replication/start` passes the **path** cluster as
    `replication_cluster_id`, i.e. the volume's own cluster, so it self-targets
    and never falls back to the configured destination.
 2. The same route takes no body, so `mode` and `interval_min` are unreachable
@@ -209,15 +210,15 @@ aggregates into one multipath device on the client, so returning both during
 
 ### Group fail-over
 
-Per-volume fail-over (`POST .../volumes/{v}/replicate_lvol`) is the only trigger
+Per-volume fail-over (`POST .../volumes/{v}/replication/failover`) is the only trigger
 today: there is no CLI verb, and no automatic trigger anywhere in the control
 plane — the Kubernetes operator supplies the automation by polling health and
 calling that route per volume. A site loss needs to fail over **every** volume
 replicating to a given destination, so add a group call:
 
-- `POST /clusters/{c}/replication-targets/{target_id}/failover` — fails over
+- `POST /clusters/{c}/replication/targets/{target_id}/failover` — fails over
   every volume whose policy points at that target.
-- `POST /clusters/{c}/replication-policies/{policy_id}/failover` — same, scoped
+- `POST /clusters/{c}/replication/policies/{policy_id}/failover` — same, scoped
   to one policy.
 
 Both run as background tasks, are idempotent per volume (a volume already

simplyblock_core/controllers/lvol_controller.py

@@ -3583,7 +3583,7 @@ def replicate_lvol_on_target_cluster(lvol_id):
     # A failed-over volume no longer lives on the source, so there is nothing
     # left to replicate from it; any further source delta is by definition past
     # the RPO the fail-over accepted.
-    replication_stop(lvol_id)
+    replication_stop(lvol_id, from_policy=True)
 
     lvol = db_controller.get_lvol_by_id(lvol_id)
     lvol.from_source = False

simplyblock_web/api/v2/_dependencies.py

@@ -14,6 +14,10 @@
 from simplyblock_core.models.mgmt_node import MgmtNode
 from simplyblock_core.models.nvme_device import NVMeDevice
 from simplyblock_core.models.pool import Pool as PoolModel
+from simplyblock_core.models.replication import (
+    ReplicationPolicy as ReplicationPolicyModel,
+    ReplicationTarget as ReplicationTargetModel,
+)
 from simplyblock_core.models.snapshot import SnapShot as SnapshotModel
 from simplyblock_core.models.storage_node import StorageNode as StorageNodeModel
 
@@ -143,6 +147,32 @@ def _lookup_backup_policy(policy_id: UUID, cluster: Cluster) -> BackupPolicy:
 Policy = Annotated[BackupPolicy, Depends(_lookup_backup_policy)]
 
 
+def _lookup_replication_target(target_id: UUID, cluster: Cluster) -> ReplicationTargetModel:
+    try:
+        target = _db.get_replication_target_by_id(str(target_id))
+    except KeyError as e:
+        raise HTTPException(404, str(e))
+    if target.cluster_id != cluster.get_id():
+        raise HTTPException(404, f'ReplicationTarget {target_id} not found')
+    return target
+
+
+ReplicationTarget = Annotated[ReplicationTargetModel, Depends(_lookup_replication_target)]
+
+
+def _lookup_replication_policy(policy_id: UUID, cluster: Cluster) -> ReplicationPolicyModel:
+    try:
+        policy = _db.get_replication_policy_by_id(str(policy_id))
+    except KeyError as e:
+        raise HTTPException(404, str(e))
+    if policy.cluster_id != cluster.get_id():
+        raise HTTPException(404, f'ReplicationPolicy {policy_id} not found')
+    return policy
+
+
+ReplicationPolicy = Annotated[ReplicationPolicyModel, Depends(_lookup_replication_policy)]
+
+
 def _lookup_subsystem(nqn: str, cluster: Cluster) -> str:
     """Validate that `nqn` roughly looks like a real NQN and return it as-is.
 

simplyblock_web/api/v2/_dtos.py

@@ -16,6 +16,7 @@
 from simplyblock_core.utils.nvme import NvmeConnectEntry
 from simplyblock_core.models.nvme_device import NVMeDevice
 from simplyblock_core.models.pool import Pool
+from simplyblock_core.models.replication import ReplicationPolicy, ReplicationTarget
 from simplyblock_core.models.snapshot import SnapShot
 from simplyblock_core.models.storage_node import StorageNode
 from simplyblock_core.models.backup import Backup, BackupPolicy
@@ -54,6 +55,18 @@
 
 TaskStatus = Literal["new", "running", "suspended", "done"]
 
+ReplicationTargetStatus = Literal["active", "inactive"]
+
+ReplicationPolicyStatus = Literal["active", "inactive"]
+
+ReplicationMode = Literal["failover", "migration"]
+
+FailoverStatus = Literal["failed_over", "skipped", "failed"]
+
+ReplicationState = Literal["replicating", "cutover_pending", "cutover_done", "failed_over"]
+
+ReplicationDirection = Literal["to_target", "to_source"]
+
 TaskFunctionName = Literal[
     "device_restart",
     "node_restart",
@@ -558,6 +571,79 @@ def from_model(model: BackupPolicy):
         )
 
 
+def _record_uuid(record_id: str) -> UUID:
+    """Replication records carry a composite ``cluster_id/uuid`` id."""
+    return UUID(record_id.split('/')[-1])
+
+
+class ReplicationTargetDTO(BaseModel):
+    id: UUID
+    cluster_id: UUID
+    target_name: str
+    target_cluster_id: UUID
+    target_pool_uuid: util.OptionalUUID
+    timeout_sec: util.Unsigned
+    status: ReplicationTargetStatus
+
+    @staticmethod
+    def from_model(model: ReplicationTarget):
+        return ReplicationTargetDTO(
+            id=UUID(model.uuid),
+            cluster_id=UUID(model.cluster_id),
+            target_name=model.target_name,
+            target_cluster_id=UUID(model.target_cluster_id),
+            target_pool_uuid=UUID(model.target_pool_uuid) if model.target_pool_uuid else None,
+            timeout_sec=model.timeout_sec,
+            status=cast(ReplicationTargetStatus, model.status),
+        )
+
+
+class ReplicationPolicyDTO(BaseModel):
+    id: UUID
+    cluster_id: UUID
+    policy_name: str
+    target_id: UUID
+    interval_min: util.Unsigned
+    mode: ReplicationMode
+    keep_replicated: int
+    status: ReplicationPolicyStatus
+
+    @staticmethod
+    def from_model(model: ReplicationPolicy):
+        return ReplicationPolicyDTO(
+            id=UUID(model.uuid),
+            cluster_id=UUID(model.cluster_id),
+            policy_name=model.policy_name,
+            target_id=_record_uuid(model.target_id),
+            interval_min=model.interval_min,
+            mode=cast(ReplicationMode, model.mode),
+            keep_replicated=model.keep_replicated,
+            status=cast(ReplicationPolicyStatus, model.status),
+        )
+
+
+class ReplicationRelationshipDTO(BaseModel):
+    replication_id: UUID
+    source_lvol_id: util.OptionalUUID
+    target_lvol_id: util.OptionalUUID
+    source_cluster_id: util.OptionalUUID
+    target_cluster_id: util.OptionalUUID
+    mode: ReplicationMode
+    state: ReplicationState
+    direction: ReplicationDirection
+    target_nqn: str
+    target_ns_id: int
+    is_source: bool
+
+
+class FailoverResultDTO(BaseModel):
+    lvol_id: UUID
+    status: FailoverStatus
+    detail: Optional[str] = None
+    target_lvol_id: util.OptionalUUID = None
+    connection_strings: List[NvmeConnectEntry] = []
+
+
 class MigrationDTO(BaseModel):
     id: UUID
     lvol_id: str

simplyblock_web/api/v2/cluster/__init__.py

@@ -263,5 +263,5 @@ def rebalance_cluster( cluster: Cluster) -> Response:
 instance_api.include_router(pool_api, prefix='/storage-pools')
 instance_api.include_router(backup_api, prefix='/backups')
 instance_api.include_router(subsystem_api, prefix='/subsystems')
-instance_api.include_router(replication_api)
+instance_api.include_router(replication_api, prefix='/replication')
 api.include_router(instance_api)