diff --git a/docs/replication-policies-design.md b/docs/replication-policies-design.md index 02be19eea..19c8c5b68 100644 --- a/docs/replication-policies-design.md +++ b/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 diff --git a/simplyblock_core/controllers/lvol_controller.py b/simplyblock_core/controllers/lvol_controller.py index 4992555ae..405841f82 100644 --- a/simplyblock_core/controllers/lvol_controller.py +++ b/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 diff --git a/simplyblock_web/api/v2/_dependencies.py b/simplyblock_web/api/v2/_dependencies.py index 0e71900a3..5d139f479 100644 --- a/simplyblock_web/api/v2/_dependencies.py +++ b/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. diff --git a/simplyblock_web/api/v2/_dtos.py b/simplyblock_web/api/v2/_dtos.py index 0e40fbf1d..7e45cde0b 100644 --- a/simplyblock_web/api/v2/_dtos.py +++ b/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 diff --git a/simplyblock_web/api/v2/cluster/__init__.py b/simplyblock_web/api/v2/cluster/__init__.py index df9b49264..25b2f316e 100644 --- a/simplyblock_web/api/v2/cluster/__init__.py +++ b/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) diff --git a/simplyblock_web/api/v2/cluster/replication.py b/simplyblock_web/api/v2/cluster/replication.py index f2c35d645..0a3c17d93 100644 --- a/simplyblock_web/api/v2/cluster/replication.py +++ b/simplyblock_web/api/v2/cluster/replication.py @@ -1,169 +1,174 @@ -from typing import List, Optional +from typing import Annotated, List, Optional +from uuid import UUID -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, HTTPException, Request, Response from pydantic import BaseModel, Field -from simplyblock_core.controllers import replication_policy_controller as rpc +from simplyblock_core.db_controller import DBController +from simplyblock_core.controllers import replication_policy_controller from simplyblock_core.controllers.replication_policy_controller import ReplicationConfigError -from .._dependencies import Cluster +from .. import util +from .._dependencies import Cluster, ReplicationPolicy, ReplicationTarget +from .._dtos import ( + FailoverResultDTO, + ReplicationMode, + ReplicationPolicyDTO, + ReplicationTargetDTO, +) api = APIRouter(tags=['replication']) - - -class ReplicationTargetDTO(BaseModel): - id: str - cluster_id: str - target_name: str - target_cluster_id: str - target_pool_uuid: str - timeout_sec: int - status: str - - @classmethod - def from_model(cls, model): - return cls( - id=model.get_id(), - cluster_id=model.cluster_id, - target_name=model.target_name, - target_cluster_id=model.target_cluster_id, - target_pool_uuid=model.target_pool_uuid, - timeout_sec=model.timeout_sec, - status=model.status, - ) - - -class ReplicationPolicyDTO(BaseModel): - id: str - cluster_id: str - policy_name: str - target_id: str - interval_min: int - mode: str - keep_replicated: int - status: str - - @classmethod - def from_model(cls, model): - return cls( - id=model.get_id(), - cluster_id=model.cluster_id, - policy_name=model.policy_name, - target_id=model.target_id, - interval_min=model.interval_min, - mode=model.mode, - keep_replicated=model.keep_replicated, - status=model.status, - ) +db = DBController() class TargetParams(BaseModel): target_name: str - target_cluster_id: str - target_pool: Optional[str] = None - timeout_sec: Optional[int] = None + target_cluster_id: UUID + target_pool_id: Optional[UUID] = None + timeout_sec: Optional[util.Unsigned] = None class PolicyParams(BaseModel): policy_name: str - target: str # target id or name - interval_min: int = 1 - mode: Optional[str] = None # failover | migration - keep_replicated: Optional[int] = Field(None, ge=2) - - -class FailoverResultDTO(BaseModel): - lvol_id: str - status: str # failed_over | skipped | failed - detail: Optional[str] = None - target_lvol_id: Optional[str] = None - connection_strings: Optional[List[str]] = None + target_id: UUID + interval_min: util.Unsigned = 1 + mode: Optional[ReplicationMode] = None + keep_replicated: Optional[Annotated[int, Field(ge=2)]] = None def _config_error(e: ReplicationConfigError): return HTTPException(status_code=400, detail=str(e)) -@api.get('/replication-targets', name='clusters:replication-targets:list') +targets_api = APIRouter() + + +@targets_api.get('/', name='clusters:replication:targets:list') def list_targets(cluster: Cluster) -> List[ReplicationTargetDTO]: - return [ReplicationTargetDTO.from_model(t) for t in rpc.list_targets(cluster.get_id())] + return [ + ReplicationTargetDTO.from_model(target) + for target in replication_policy_controller.list_targets(cluster.get_id()) + ] -@api.post('/replication-targets', name='clusters:replication-targets:create', status_code=201) -def create_target(cluster: Cluster, parameters: TargetParams) -> ReplicationTargetDTO: +@targets_api.post('/', name='clusters:replication:targets:create', status_code=201, + responses={201: {"content": None}}) +def create_target(request: Request, cluster: Cluster, parameters: TargetParams, + response_format: util.CreationResponseFormatParameter = "full") -> Response: try: - target_id = rpc.add_target( - cluster.get_id(), parameters.target_name, parameters.target_cluster_id, - target_pool=parameters.target_pool, timeout_sec=parameters.timeout_sec) + target_id = replication_policy_controller.add_target( + cluster.get_id(), parameters.target_name, str(parameters.target_cluster_id), + target_pool=str(parameters.target_pool_id) if parameters.target_pool_id else None, + timeout_sec=parameters.timeout_sec) except ReplicationConfigError as e: raise _config_error(e) except KeyError as e: raise HTTPException(status_code=404, detail=str(e)) - return ReplicationTargetDTO.from_model(rpc.db.get_replication_target_by_id(target_id)) + target = db.get_replication_target_by_id(target_id) + return util.creation_response( + request, response_format, + entity_id=UUID(target.uuid), + route_name='clusters:replication:targets:detail', + route_kwargs={'cluster_id': UUID(cluster.get_id()), 'target_id': UUID(target.uuid)}, + get_full=lambda _: ReplicationTargetDTO.from_model(target), + ) -@api.delete('/replication-targets/{target_id}', name='clusters:replication-targets:delete', - status_code=204, responses={204: {"content": None}}) -def delete_target(cluster: Cluster, target_id: str) -> None: + +target_instance_api = APIRouter(prefix='/{target_id}') + + +@target_instance_api.get('/', name='clusters:replication:targets:detail') +def get_target(cluster: Cluster, target: ReplicationTarget) -> ReplicationTargetDTO: + return ReplicationTargetDTO.from_model(target) + + +@target_instance_api.delete('/', name='clusters:replication:targets:delete', + status_code=204, responses={204: {"content": None}}) +def delete_target(cluster: Cluster, target: ReplicationTarget) -> Response: try: - rpc.remove_target(target_id) + replication_policy_controller.remove_target(target.get_id()) except ReplicationConfigError as e: raise _config_error(e) - except KeyError as e: - raise HTTPException(status_code=404, detail=str(e)) + return Response(status_code=204) -@api.post('/replication-targets/{target_id}/failover', - name='clusters:replication-targets:failover') -def failover_target(cluster: Cluster, target_id: str) -> List[FailoverResultDTO]: +@target_instance_api.post('/failover', name='clusters:replication:targets:failover') +def failover_target(cluster: Cluster, target: ReplicationTarget) -> List[FailoverResultDTO]: """Fail over EVERY volume replicating to this target. A site loss has to move all volumes at once; doing it volume by volume was the only option before. Idempotent per volume and reports one result per volume, so a partial failure is visible instead of silent. """ - try: - results = rpc.failover_target(target_id) - except KeyError as e: - raise HTTPException(status_code=404, detail=str(e)) - return [FailoverResultDTO(**r) for r in results] + return [ + FailoverResultDTO(**result) + for result in replication_policy_controller.failover_target(target.get_id()) + ] + + +policies_api = APIRouter() -@api.get('/replication-policies', name='clusters:replication-policies:list') +@policies_api.get('/', name='clusters:replication:policies:list') def list_policies(cluster: Cluster) -> List[ReplicationPolicyDTO]: - return [ReplicationPolicyDTO.from_model(p) for p in rpc.list_policies(cluster.get_id())] + return [ + ReplicationPolicyDTO.from_model(policy) + for policy in replication_policy_controller.list_policies(cluster.get_id()) + ] -@api.post('/replication-policies', name='clusters:replication-policies:create', status_code=201) -def create_policy(cluster: Cluster, parameters: PolicyParams) -> ReplicationPolicyDTO: +@policies_api.post('/', name='clusters:replication:policies:create', status_code=201, + responses={201: {"content": None}}) +def create_policy(request: Request, cluster: Cluster, parameters: PolicyParams, + response_format: util.CreationResponseFormatParameter = "full") -> Response: try: - policy_id = rpc.add_policy( - cluster.get_id(), parameters.policy_name, parameters.target, + policy_id = replication_policy_controller.add_policy( + cluster.get_id(), parameters.policy_name, str(parameters.target_id), interval_min=parameters.interval_min, mode=parameters.mode, keep_replicated=parameters.keep_replicated) except ReplicationConfigError as e: raise _config_error(e) except KeyError as e: raise HTTPException(status_code=404, detail=str(e)) - return ReplicationPolicyDTO.from_model(rpc.db.get_replication_policy_by_id(policy_id)) + policy = db.get_replication_policy_by_id(policy_id) + return util.creation_response( + request, response_format, + entity_id=UUID(policy.uuid), + route_name='clusters:replication:policies:detail', + route_kwargs={'cluster_id': UUID(cluster.get_id()), 'policy_id': UUID(policy.uuid)}, + get_full=lambda _: ReplicationPolicyDTO.from_model(policy), + ) + + +policy_instance_api = APIRouter(prefix='/{policy_id}') + + +@policy_instance_api.get('/', name='clusters:replication:policies:detail') +def get_policy(cluster: Cluster, policy: ReplicationPolicy) -> ReplicationPolicyDTO: + return ReplicationPolicyDTO.from_model(policy) -@api.delete('/replication-policies/{policy_id}', name='clusters:replication-policies:delete', - status_code=204, responses={204: {"content": None}}) -def delete_policy(cluster: Cluster, policy_id: str) -> None: + +@policy_instance_api.delete('/', name='clusters:replication:policies:delete', + status_code=204, responses={204: {"content": None}}) +def delete_policy(cluster: Cluster, policy: ReplicationPolicy) -> Response: try: - rpc.remove_policy(policy_id) + replication_policy_controller.remove_policy(policy.get_id()) except ReplicationConfigError as e: raise _config_error(e) - except KeyError as e: - raise HTTPException(status_code=404, detail=str(e)) + return Response(status_code=204) -@api.post('/replication-policies/{policy_id}/failover', - name='clusters:replication-policies:failover') -def failover_policy(cluster: Cluster, policy_id: str) -> List[FailoverResultDTO]: - try: - results = rpc.failover_policy(policy_id) - except KeyError as e: - raise HTTPException(status_code=404, detail=str(e)) - return [FailoverResultDTO(**r) for r in results] +@policy_instance_api.post('/failover', name='clusters:replication:policies:failover') +def failover_policy(cluster: Cluster, policy: ReplicationPolicy) -> List[FailoverResultDTO]: + return [ + FailoverResultDTO(**result) + for result in replication_policy_controller.failover_policy(policy.get_id()) + ] + + +targets_api.include_router(target_instance_api) +policies_api.include_router(policy_instance_api) +api.include_router(targets_api, prefix='/targets') +api.include_router(policies_api, prefix='/policies') diff --git a/simplyblock_web/api/v2/cluster/storage_pool/volume/__init__.py b/simplyblock_web/api/v2/cluster/storage_pool/volume/__init__.py index dc444be2b..10b6e9120 100644 --- a/simplyblock_web/api/v2/cluster/storage_pool/volume/__init__.py +++ b/simplyblock_web/api/v2/cluster/storage_pool/volume/__init__.py @@ -6,12 +6,17 @@ from simplyblock_core.db_controller import DBController from simplyblock_core import utils as core_utils -from simplyblock_core.controllers import backup_controller, lvol_controller, snapshot_controller, replication_policy_controller +from simplyblock_core.controllers import backup_controller, lvol_controller, snapshot_controller from simplyblock_core.models.lvol_model import LVol from ...._dependencies import Cluster, StoragePool, Volume -from ...._dtos import BackupDTO, VolumeDTO, SnapshotDTO, TaskDTO +from ...._dtos import BackupDTO, VolumeDTO, SnapshotDTO from .... import util +from .replication import ( + api as replication_api, + apply_policy as apply_replication_policy, + collection_api as replication_collection_api, +) api = APIRouter() @@ -131,15 +136,6 @@ def add( ) -class ReplicateLVolParams(BaseModel): - lvol_id: Optional[str] = None - - -@api.post('/replicate_lvol_on_source_cluster', name='clusters:storage-pools:replicate_lvol_on_source_cluster') -def replicate_lvol_on_source_cluster(cluster: Cluster, pool: StoragePool, body: ReplicateLVolParams): - return lvol_controller.replicate_lvol_on_source_cluster(body.lvol_id, cluster.get_id(), pool.get_id()) - - instance_api = APIRouter(prefix='/{volume_id}') @@ -157,6 +153,8 @@ class UpdatableLVolParams(BaseModel): max_r_mbytes: util.Unsigned = 0 max_w_mbytes: util.Unsigned = 0 size: Optional[util.Size] = None + # Omitted leaves the volume's policy alone; null detaches it. + replication_policy_id: Optional[UUID] = None @instance_api.put('/', name='clusters:storage-pools:volumes:update', status_code=204, responses={204: {"content": None}}) @@ -174,6 +172,9 @@ def update(cluster: Cluster, pool: StoragePool, volume: Volume, body: UpdatableL if 'size' in body.model_fields_set: lvol_controller.resize_lvol(volume.get_id(), body.size) + if 'replication_policy_id' in body.model_fields_set: + apply_replication_policy(volume, body.replication_policy_id) + return Response(status_code=204) @@ -226,47 +227,6 @@ def inflate(cluster: Cluster, pool: StoragePool, volume: Volume) -> Response: return Response(status_code=204) -@instance_api.post('/replication_trigger', name='clusters:storage-pools:volumes:replication_start', status_code=204, responses={204: {"content": None}}) -def replication_trigger(cluster: Cluster, pool: StoragePool, volume: Volume) -> Response: - if not lvol_controller.replication_trigger(volume.get_id()): - raise ValueError('Failed to start volume snapshot replication') - - return Response(status_code=204) - -class ReplicationStartParams(BaseModel): - replication_cluster_id: Optional[str] = None # destination; None = cluster default - mode: Optional[str] = None # failover | migration - interval_min: Annotated[Optional[int], Field(None, ge=0)] = None - - -@instance_api.post('/replication_start', name='clusters:storage-pools:volumes:replication_start', status_code=204, responses={204: {"content": None}}) -def replication_start(cluster: Cluster, pool: StoragePool, volume: Volume, - body: Optional[ReplicationStartParams] = None) -> Response: - """Start replicating a volume. - - The destination is the request's replication_cluster_id, else the cluster's - configured target. It used to pass the PATH cluster — the volume's OWN - cluster — as the destination, which self-targets and never falls back to the - configured target, so replication could not be started correctly over REST - at all. mode/interval_min were likewise unreachable. - """ - params = body or ReplicationStartParams() - if not lvol_controller.replication_start( - volume.get_id(), - replication_cluster_id=params.replication_cluster_id, - mode=params.mode, - interval_min=params.interval_min): - raise ValueError('Failed to start volume snapshot replication') - - return Response(status_code=204) - -@instance_api.post('/replication_stop', name='clusters:storage-pools:volumes:replication_stop', status_code=204, responses={204: {"content": None}}) -def replication_stop(cluster: Cluster, pool: StoragePool, volume: Volume) -> Response: - if not lvol_controller.replication_stop(volume.get_id()): - raise ValueError('Failed to stop volume snapshot replication') - - return Response(status_code=204) - @instance_api.get('/connect', name='clusters:storage-pools:volumes:connect') def connect(cluster: Cluster, pool: StoragePool, volume: Volume, host_nqn: Optional[str] = None): details, err = lvol_controller.connect_lvol(volume.get_id(), host_nqn=host_nqn) @@ -334,30 +294,6 @@ def create_snapshot( return Response(status_code=201, headers={'Location': entity_url}) -@instance_api.post('/replicate_lvol', name='clusters:storage-pools:volumes:replicate_lvol') -def replicate_lvol_on_target_cluster(cluster: Cluster, pool: StoragePool, volume: Volume): - return lvol_controller.replicate_lvol_on_target_cluster(volume.get_id()) - - -@instance_api.post('/replication_commit', name='clusters:storage-pools:volumes:replication_commit') -def replication_commit(cluster: Cluster, pool: StoragePool, volume: Volume): - return lvol_controller.replication_commit(volume.get_id()) - - -class FailbackParams(BaseModel): - source_cluster_id: Optional[str] = None - - -@instance_api.post('/replication_failback', name='clusters:storage-pools:volumes:replication_failback') -def replication_failback(cluster: Cluster, pool: StoragePool, volume: Volume, body: FailbackParams): - return lvol_controller.replication_failback(volume.get_id(), source_cluster_id=body.source_cluster_id) - - -@instance_api.get('/list_replication_tasks', name='clusters:storage-pools:volumes:list_replication_tasks') -def list_replication_tasks(cluster: Cluster, pool: StoragePool, volume: Volume) -> List[TaskDTO]: - tasks = lvol_controller.list_replication_tasks(volume.get_id()) - return [TaskDTO.from_model(task) for task in tasks] - @instance_api.route( '/suspend', name='clusters:storage-pools:volumes:suspend', @@ -418,68 +354,6 @@ def delete_backups(cluster: Cluster, pool: StoragePool, volume: Volume) -> Respo raise HTTPException(400, error) return Response(status_code=204) - - -class PolicyAssignParams(BaseModel): - policy: str # policy id or name - - -class ReplicationRelationshipDTO(BaseModel): - replication_id: str - source_lvol_id: str - target_lvol_id: str - source_cluster_id: str - target_cluster_id: str - mode: str - state: str - direction: str - target_nqn: str - target_ns_id: int - is_source: bool - - -@instance_api.get('/replication', name='clusters:storage-pools:volumes:replication') -def get_replication_relationship(cluster: Cluster, pool: StoragePool, volume: Volume) -> ReplicationRelationshipDTO: - """Resolve a volume to its counterpart on the other cluster. - - Answers "what is the TARGET volume uuid for this SOURCE volume uuid" (and the - reverse). Before this the ids were only returned by the fail-over or commit - call itself, so a caller that had not kept them could not find the target - volume through the API at all -- LVolReplication was exposed nowhere. - """ - rel = replication_policy_controller.get_relationship(volume.get_id()) - if rel is None: - raise HTTPException(status_code=404, detail='Volume has no replication relationship') - return ReplicationRelationshipDTO(**rel) - - -@instance_api.put('/replication-policy', name='clusters:storage-pools:volumes:replication-policy:set', - status_code=204, responses={204: {"content": None}}) -def set_replication_policy(cluster: Cluster, pool: StoragePool, volume: Volume, - body: PolicyAssignParams) -> Response: - """Attach a policy, or change it (detach then attach, so the new target - receives a FULL copy).""" - try: - replication_policy_controller.attach_policy(volume.get_id(), body.policy) - except replication_policy_controller.ReplicationConfigError as e: - raise HTTPException(status_code=400, detail=str(e)) - except KeyError as e: - raise HTTPException(status_code=404, detail=str(e)) - return Response(status_code=204) - - -@instance_api.delete('/replication-policy', name='clusters:storage-pools:volumes:replication-policy:clear', - status_code=204, responses={204: {"content": None}}) -def clear_replication_policy(cluster: Cluster, pool: StoragePool, volume: Volume) -> Response: - """Detach: stop replicating and delete the internal replication snapshots on - both sides. Refused while a cutover is in flight.""" - try: - replication_policy_controller.detach_policy(volume.get_id()) - except replication_policy_controller.ReplicationConfigError as e: - raise HTTPException(status_code=409, detail=str(e)) - except KeyError as e: - raise HTTPException(status_code=404, detail=str(e)) - return Response(status_code=204) - - +api.include_router(replication_collection_api) +instance_api.include_router(replication_api, prefix='/replication') api.include_router(instance_api) diff --git a/simplyblock_web/api/v2/cluster/storage_pool/volume/replication.py b/simplyblock_web/api/v2/cluster/storage_pool/volume/replication.py new file mode 100644 index 000000000..339e5ef34 --- /dev/null +++ b/simplyblock_web/api/v2/cluster/storage_pool/volume/replication.py @@ -0,0 +1,189 @@ +from typing import List, Optional +from uuid import UUID + +from fastapi import APIRouter, HTTPException, Request, Response +from pydantic import BaseModel + +from simplyblock_core.controllers import lvol_controller, replication_policy_controller +from simplyblock_core.controllers.replication_policy_controller import ReplicationConfigError +from simplyblock_core.models.lvol_model import LVol + +from .... import util +from ...._dependencies import Cluster, StoragePool, Volume +from ...._dtos import ReplicationMode, ReplicationRelationshipDTO, TaskDTO + + +api = APIRouter(tags=['replication']) +collection_api = APIRouter(tags=['replication']) + + +def apply_policy(volume: LVol, policy_id: Optional[UUID]) -> None: + """Put *volume* under replication policy *policy_id*, or take it out (None). + + Changing policy is detach-then-attach, so the new target receives a FULL + copy. Detaching stops replication and deletes the internal replication + snapshots on both sides. + """ + if policy_id is None: + try: + replication_policy_controller.detach_policy(volume.get_id()) + except ReplicationConfigError as e: + raise HTTPException(409, str(e)) # a cutover is in flight + except KeyError as e: + raise HTTPException(404, str(e)) + else: + try: + replication_policy_controller.attach_policy(volume.get_id(), str(policy_id)) + except ReplicationConfigError as e: + raise HTTPException(400, str(e)) + except KeyError as e: + raise HTTPException(404, str(e)) + + +@api.get('/', name='clusters:storage-pools:volumes:replication:detail') +def get_relationship(cluster: Cluster, pool: StoragePool, volume: Volume) -> ReplicationRelationshipDTO: + """Resolve a volume to its counterpart on the other cluster. + + Answers "what is the TARGET volume uuid for this SOURCE volume uuid" (and the + reverse). Before this the ids were only returned by the fail-over or commit + call itself, so a caller that had not kept them could not find the target + volume through the API at all -- LVolReplication was exposed nowhere. + """ + relationship = replication_policy_controller.get_relationship(volume.get_id()) + if relationship is None: + raise HTTPException(404, 'Volume has no replication relationship') + return ReplicationRelationshipDTO(**relationship) + + +class ReplicationStartParams(BaseModel): + replication_cluster_id: Optional[UUID] = None # destination; None = cluster default + mode: Optional[ReplicationMode] = None + interval_min: Optional[util.Unsigned] = None + + +@api.post('/start', name='clusters:storage-pools:volumes:replication:start', + status_code=204, responses={204: {"content": None}}) +def start(cluster: Cluster, pool: StoragePool, volume: Volume, + body: Optional[ReplicationStartParams] = None) -> Response: + """Start replicating a volume. + + The destination is the request's replication_cluster_id, else the cluster's + configured target. It used to pass the PATH cluster — the volume's OWN + cluster — as the destination, which self-targets and never falls back to the + configured target, so replication could not be started correctly over REST + at all. mode/interval_min were likewise unreachable. + """ + parameters = body or ReplicationStartParams() + if not lvol_controller.replication_start( + volume.get_id(), + replication_cluster_id=( + str(parameters.replication_cluster_id) + if parameters.replication_cluster_id else None + ), + mode=parameters.mode, + interval_min=parameters.interval_min): + raise HTTPException(500, 'Failed to start volume snapshot replication') + + return Response(status_code=204) + + +@api.post('/stop', name='clusters:storage-pools:volumes:replication:stop', + status_code=204, responses={204: {"content": None}}) +def stop(cluster: Cluster, pool: StoragePool, volume: Volume) -> Response: + if not lvol_controller.replication_stop(volume.get_id()): + raise HTTPException(500, 'Failed to stop volume snapshot replication') + + return Response(status_code=204) + + +@api.post('/trigger', name='clusters:storage-pools:volumes:replication:trigger', + status_code=204, responses={204: {"content": None}}) +def trigger(cluster: Cluster, pool: StoragePool, volume: Volume) -> Response: + if not lvol_controller.replication_trigger(volume.get_id()): + raise HTTPException(500, 'Failed to start volume snapshot replication') + + return Response(status_code=204) + + +@api.post('/failover', name='clusters:storage-pools:volumes:replication:failover', + status_code=204, responses={204: {"content": None}}) +def failover(cluster: Cluster, pool: StoragePool, volume: Volume) -> Response: + """Bring the volume up on the target cluster. + + The counterpart's id is read back from this volume's replication + relationship, its connection paths from the target volume's `connect`. + """ + result = lvol_controller.replicate_lvol_on_target_cluster(volume.get_id()) + if isinstance(result, tuple): # (False, error) + raise HTTPException(500, str(result[1])) + if not result: + raise HTTPException(500, 'Failed to fail the volume over to the target cluster') + + return Response(status_code=204) + + +@api.post('/commit', name='clusters:storage-pools:volumes:replication:commit', + status_code=202, responses={202: {"content": None}}) +def commit(request: Request, cluster: Cluster, pool: StoragePool, volume: Volume) -> Response: + """Queue the planned cutover. Progress is the returned task.""" + result = lvol_controller.replication_commit(volume.get_id()) + if isinstance(result, tuple): # (False, error) + raise HTTPException(500, str(result[1])) + if not result: + raise HTTPException(500, 'Failed to queue the replication cutover') + + return Response(status_code=202, headers={'Location': str(request.app.url_path_for( + 'clusters:tasks:detail', + cluster_id=cluster.get_id(), task_id=result['task_id'], + ))}) + + +class FailbackParams(BaseModel): + source_cluster_id: Optional[UUID] = None + + +@api.post('/failback', name='clusters:storage-pools:volumes:replication:failback', + status_code=204, responses={204: {"content": None}}) +def failback(cluster: Cluster, pool: StoragePool, volume: Volume, body: FailbackParams) -> Response: + """Point replication back at a source cluster. The cutover itself is + `commit`.""" + result = lvol_controller.replication_failback( + volume.get_id(), + source_cluster_id=str(body.source_cluster_id) if body.source_cluster_id else None, + ) + if isinstance(result, tuple): # (False, error) + raise HTTPException(500, str(result[1])) + if not result: + raise HTTPException(500, 'Failed to configure fail-back of the volume') + + 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())] + + +class ReplicateLVolParams(BaseModel): + lvol_id: UUID + + +@collection_api.post('/replicate_lvol_on_source_cluster', + name='clusters:storage-pools:replicate_lvol_on_source_cluster', + status_code=204, responses={204: {"content": None}}) +def replicate_lvol_on_source_cluster(cluster: Cluster, pool: StoragePool, + body: ReplicateLVolParams) -> Response: + """Rebuild a volume on the source cluster. + + Collection-scoped rather than an operation on `/{volume_id}`: the volume is + typically gone from the source cluster by the time this is called, so the + controller falls back to resolving the id through the replication records. + """ + result = lvol_controller.replicate_lvol_on_source_cluster( + str(body.lvol_id), cluster.get_id(), pool.get_id()) + if isinstance(result, tuple): # (False, error) + raise HTTPException(500, str(result[1])) + if not result: + raise HTTPException(500, 'Failed to rebuild the volume on the source cluster') + + return Response(status_code=204) diff --git a/simplyblock_web/api/v2/util.py b/simplyblock_web/api/v2/util.py index 539b9c684..c451443b7 100644 --- a/simplyblock_web/api/v2/util.py +++ b/simplyblock_web/api/v2/util.py @@ -14,6 +14,8 @@ Size = Annotated[Unsigned, BeforeValidator(core_utils.parse_size)] Percent = Annotated[int, Field(ge=0, le=100)] Port = Annotated[int, Field(ge=0, lt=65536)] +# Records spell an unset reference as an empty string rather than omitting it. +OptionalUUID = Annotated[Optional[UUID], BeforeValidator(lambda value: value or None)] def _validate_url_path(value: Any) -> str: diff --git a/tests/unit/web/api/v2/_factories.py b/tests/unit/web/api/v2/_factories.py index 4320cd54a..0196501d9 100644 --- a/tests/unit/web/api/v2/_factories.py +++ b/tests/unit/web/api/v2/_factories.py @@ -16,6 +16,7 @@ from simplyblock_core.models.mgmt_node import MgmtNode 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 @@ -31,6 +32,10 @@ BACKUP_ID = '99999999-9999-9999-9999-999999999999' POLICY_ID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa' MIGRATION_ID = 'bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb' +REPLICATION_TARGET_ID = 'cccccccc-cccc-cccc-cccc-cccccccccccc' +REPLICATION_POLICY_ID = 'dddddddd-dddd-dddd-dddd-dddddddddddd' +TARGET_CLUSTER_ID = 'eeeeeeee-eeee-eeee-eeee-eeeeeeeeeeee' +TARGET_POOL_ID = 'ffffffff-ffff-ffff-ffff-ffffffffffff' VOLUME_NQN = 'nqn.2023-02.io.simplyblock:volume-1' @@ -181,3 +186,28 @@ def make_migration(**attrs) -> LVolMigration: migration.phase = LVolMigration.PHASE_SNAP_COPY migration.status = LVolMigration.STATUS_RUNNING return _apply(migration, attrs) + + +def make_replication_target(**attrs) -> ReplicationTarget: + target = ReplicationTarget() + target.uuid = REPLICATION_TARGET_ID + target.cluster_id = CLUSTER_ID + target.target_name = 'site-b' + target.target_cluster_id = TARGET_CLUSTER_ID + target.target_pool_uuid = TARGET_POOL_ID + target.timeout_sec = 600 + target.status = ReplicationTarget.STATUS_ACTIVE + return _apply(target, attrs) + + +def make_replication_policy(**attrs) -> ReplicationPolicy: + policy = ReplicationPolicy() + policy.uuid = REPLICATION_POLICY_ID + policy.cluster_id = CLUSTER_ID + policy.policy_name = 'nightly' + policy.target_id = f'{CLUSTER_ID}/{REPLICATION_TARGET_ID}' + policy.interval_min = 5 + policy.mode = ReplicationPolicy.MODE_FAILOVER + policy.keep_replicated = 3 + policy.status = ReplicationPolicy.STATUS_ACTIVE + return _apply(policy, attrs) diff --git a/tests/unit/web/api/v2/conftest.py b/tests/unit/web/api/v2/conftest.py index 64ba57786..ea214c1a2 100644 --- a/tests/unit/web/api/v2/conftest.py +++ b/tests/unit/web/api/v2/conftest.py @@ -30,11 +30,13 @@ import simplyblock_web.api.v2._dtos as dtos_module import simplyblock_web.api.v2.cluster as cluster_module import simplyblock_web.api.v2.cluster.backup as backup_module +import simplyblock_web.api.v2.cluster.replication as replication_module import simplyblock_web.api.v2.cluster.storage_node as storage_node_module import simplyblock_web.api.v2.cluster.storage_node.device as device_module import simplyblock_web.api.v2.cluster.storage_pool as storage_pool_module import simplyblock_web.api.v2.cluster.storage_pool.snapshot as snapshot_module import simplyblock_web.api.v2.cluster.storage_pool.volume as volume_module +import simplyblock_web.api.v2.cluster.storage_pool.volume.replication as volume_replication_module import simplyblock_web.api.v2.cluster.subsystem.migration as migration_module import simplyblock_web.api.v2.cluster.task as task_module import simplyblock_web.api.v2.management_node as management_node_module @@ -77,6 +79,7 @@ def db(monkeypatch): for module in ( cluster_module, backup_module, + replication_module, storage_node_module, device_module, storage_pool_module, @@ -134,6 +137,7 @@ def lvol_controller(monkeypatch): mock = MagicMock() mock.get_replication_info.return_value = None monkeypatch.setattr(volume_module, 'lvol_controller', mock) + monkeypatch.setattr(volume_replication_module, 'lvol_controller', mock) return mock @@ -153,6 +157,14 @@ def backup_controller(monkeypatch): return mock +@pytest.fixture() +def replication_policy_controller(monkeypatch): + mock = MagicMock() + monkeypatch.setattr(replication_module, 'replication_policy_controller', mock) + monkeypatch.setattr(volume_replication_module, 'replication_policy_controller', mock) + return mock + + @pytest.fixture() def storage_node_ops(monkeypatch): mock = MagicMock() @@ -263,6 +275,22 @@ def backup_policy(db, cluster): return policy +@pytest.fixture() +def replication_target(db, cluster): + target = factories.make_replication_target() + db.get_replication_targets.return_value = [target] + db.get_replication_target_by_id.return_value = target + return target + + +@pytest.fixture() +def replication_policy(db, replication_target): + policy = factories.make_replication_policy() + db.get_replication_policies.return_value = [policy] + db.get_replication_policy_by_id.return_value = policy + return policy + + @pytest.fixture() def migration(db, volume): migration = factories.make_migration() diff --git a/tests/unit/web/api/v2/test_replication_endpoints.py b/tests/unit/web/api/v2/test_replication_endpoints.py new file mode 100644 index 000000000..07b64f542 --- /dev/null +++ b/tests/unit/web/api/v2/test_replication_endpoints.py @@ -0,0 +1,262 @@ +# coding=utf-8 +"""Unit tests for /api/v2/clusters/{id}/replication endpoints.""" + +from simplyblock_core.controllers.replication_policy_controller import ReplicationConfigError + +from tests.unit.web.api.v2._factories import ( + CLUSTER_ID, + REPLICATION_POLICY_ID, + REPLICATION_TARGET_ID, + TARGET_CLUSTER_ID, + TARGET_POOL_ID, + VOLUME_ID, +) + + +TARGETS_URL = f'/api/v2/clusters/{CLUSTER_ID}/replication/targets/' +POLICIES_URL = f'/api/v2/clusters/{CLUSTER_ID}/replication/policies/' + + +class TestListTargets: + + def test_returns_targets_from_controller(self, client, db, cluster, replication_target, + replication_policy_controller): + replication_policy_controller.list_targets.return_value = [replication_target] + + response = client.get(TARGETS_URL) + + assert response.status_code == 200 + (body,) = response.json() + assert body['id'] == REPLICATION_TARGET_ID + assert body['cluster_id'] == CLUSTER_ID + assert body['target_name'] == 'site-b' + assert body['target_cluster_id'] == TARGET_CLUSTER_ID + assert body['target_pool_uuid'] == TARGET_POOL_ID + replication_policy_controller.list_targets.assert_called_once_with(CLUSTER_ID) + + def test_unset_target_pool_serializes_as_null(self, client, db, cluster, replication_target, + replication_policy_controller): + replication_target.target_pool_uuid = '' + replication_policy_controller.list_targets.return_value = [replication_target] + + (body,) = client.get(TARGETS_URL).json() + + assert body['target_pool_uuid'] is None + + +class TestCreateTarget: + + def test_creates_target_and_links_to_it(self, client, db, cluster, replication_target, + replication_policy_controller): + replication_policy_controller.add_target.return_value = \ + f'{CLUSTER_ID}/{REPLICATION_TARGET_ID}' + + response = client.post(TARGETS_URL, json={ + 'target_name': 'site-b', + 'target_cluster_id': TARGET_CLUSTER_ID, + 'target_pool_id': TARGET_POOL_ID, + 'timeout_sec': 600, + }) + + assert response.status_code == 201 + assert response.json()['id'] == REPLICATION_TARGET_ID + assert response.headers['Location'].endswith( + f'/clusters/{CLUSTER_ID}/replication/targets/{REPLICATION_TARGET_ID}/') + args, kwargs = replication_policy_controller.add_target.call_args + assert args == (CLUSTER_ID, 'site-b', TARGET_CLUSTER_ID) + assert kwargs == {'target_pool': TARGET_POOL_ID, 'timeout_sec': 600} + + def test_identifier_response_format(self, client, db, cluster, replication_target, + replication_policy_controller): + replication_policy_controller.add_target.return_value = \ + f'{CLUSTER_ID}/{REPLICATION_TARGET_ID}' + + response = client.post(TARGETS_URL + '?response-format=identifier', json={ + 'target_name': 'site-b', + 'target_cluster_id': TARGET_CLUSTER_ID, + }) + + assert response.status_code == 201 + assert response.json() == REPLICATION_TARGET_ID + + def test_non_uuid_cluster_rejected(self, client, db, cluster, replication_policy_controller): + response = client.post(TARGETS_URL, json={ + 'target_name': 'site-b', + 'target_cluster_id': 'not-a-uuid', + }) + + assert response.status_code == 422 + replication_policy_controller.add_target.assert_not_called() + + def test_config_error_maps_to_400(self, client, db, cluster, replication_policy_controller): + replication_policy_controller.add_target.side_effect = \ + ReplicationConfigError('A cluster cannot replicate to itself') + + response = client.post(TARGETS_URL, json={ + 'target_name': 'site-b', + 'target_cluster_id': TARGET_CLUSTER_ID, + }) + + assert response.status_code == 400 + + def test_unknown_cluster_maps_to_404(self, client, db, cluster, replication_policy_controller): + replication_policy_controller.add_target.side_effect = KeyError('Cluster not found') + + response = client.post(TARGETS_URL, json={ + 'target_name': 'site-b', + 'target_cluster_id': TARGET_CLUSTER_ID, + }) + + assert response.status_code == 404 + + +class TestTargetInstance: + + def test_detail(self, client, db, cluster, replication_target): + response = client.get(TARGETS_URL + f'{REPLICATION_TARGET_ID}/') + + assert response.status_code == 200 + assert response.json()['id'] == REPLICATION_TARGET_ID + + def test_target_of_another_cluster_is_not_found(self, client, db, cluster, replication_target): + replication_target.cluster_id = TARGET_CLUSTER_ID + + response = client.get(TARGETS_URL + f'{REPLICATION_TARGET_ID}/') + + assert response.status_code == 404 + + def test_delete(self, client, db, cluster, replication_target, + replication_policy_controller): + response = client.delete(TARGETS_URL + f'{REPLICATION_TARGET_ID}/') + + assert response.status_code == 204 + replication_policy_controller.remove_target.assert_called_once_with( + f'{CLUSTER_ID}/{REPLICATION_TARGET_ID}') + + def test_delete_of_used_target_maps_to_400(self, client, db, cluster, replication_target, + replication_policy_controller): + replication_policy_controller.remove_target.side_effect = \ + ReplicationConfigError('still used by 1 policy(ies)') + + response = client.delete(TARGETS_URL + f'{REPLICATION_TARGET_ID}/') + + assert response.status_code == 400 + + def test_failover_reports_per_volume_results(self, client, db, cluster, replication_target, + replication_policy_controller): + replication_policy_controller.failover_target.return_value = [ + {'lvol_id': VOLUME_ID, 'status': 'failed_over', + 'target_lvol_id': VOLUME_ID, 'connection_strings': ['nvme connect …']}, + {'lvol_id': VOLUME_ID, 'status': 'skipped', + 'detail': 'already failed_over', 'target_lvol_id': ''}, + ] + + response = client.post(TARGETS_URL + f'{REPLICATION_TARGET_ID}/failover') + + assert response.status_code == 200 + done, skipped = response.json() + assert done['status'] == 'failed_over' + assert done['connection_strings'] == ['nvme connect …'] + assert skipped['status'] == 'skipped' + assert skipped['target_lvol_id'] is None + + +class TestListPolicies: + + def test_returns_policies_from_controller(self, client, db, cluster, replication_policy, + replication_policy_controller): + replication_policy_controller.list_policies.return_value = [replication_policy] + + response = client.get(POLICIES_URL) + + assert response.status_code == 200 + (body,) = response.json() + assert body['id'] == REPLICATION_POLICY_ID + assert body['policy_name'] == 'nightly' + assert body['target_id'] == REPLICATION_TARGET_ID + assert body['mode'] == 'failover' + replication_policy_controller.list_policies.assert_called_once_with(CLUSTER_ID) + + +class TestCreatePolicy: + + def test_creates_policy_and_links_to_it(self, client, db, cluster, replication_policy, + replication_policy_controller): + replication_policy_controller.add_policy.return_value = \ + f'{CLUSTER_ID}/{REPLICATION_POLICY_ID}' + + response = client.post(POLICIES_URL, json={ + 'policy_name': 'nightly', + 'target_id': REPLICATION_TARGET_ID, + 'interval_min': 5, + 'mode': 'failover', + 'keep_replicated': 3, + }) + + assert response.status_code == 201 + assert response.json()['id'] == REPLICATION_POLICY_ID + assert response.headers['Location'].endswith( + f'/clusters/{CLUSTER_ID}/replication/policies/{REPLICATION_POLICY_ID}/') + args, kwargs = replication_policy_controller.add_policy.call_args + assert args == (CLUSTER_ID, 'nightly', REPLICATION_TARGET_ID) + assert kwargs == {'interval_min': 5, 'mode': 'failover', 'keep_replicated': 3} + + def test_unknown_mode_rejected(self, client, db, cluster, replication_policy_controller): + response = client.post(POLICIES_URL, json={ + 'policy_name': 'nightly', + 'target_id': REPLICATION_TARGET_ID, + 'mode': 'sideways', + }) + + assert response.status_code == 422 + replication_policy_controller.add_policy.assert_not_called() + + def test_keep_replicated_below_minimum_rejected(self, client, db, cluster, + replication_policy_controller): + response = client.post(POLICIES_URL, json={ + 'policy_name': 'nightly', + 'target_id': REPLICATION_TARGET_ID, + 'keep_replicated': 1, + }) + + assert response.status_code == 422 + replication_policy_controller.add_policy.assert_not_called() + + +class TestPolicyInstance: + + def test_detail(self, client, db, cluster, replication_policy): + response = client.get(POLICIES_URL + f'{REPLICATION_POLICY_ID}/') + + assert response.status_code == 200 + assert response.json()['id'] == REPLICATION_POLICY_ID + + def test_policy_of_another_cluster_is_not_found(self, client, db, cluster, replication_policy): + replication_policy.cluster_id = TARGET_CLUSTER_ID + + response = client.get(POLICIES_URL + f'{REPLICATION_POLICY_ID}/') + + assert response.status_code == 404 + + def test_delete(self, client, db, cluster, replication_policy, + replication_policy_controller): + response = client.delete(POLICIES_URL + f'{REPLICATION_POLICY_ID}/') + + assert response.status_code == 204 + replication_policy_controller.remove_policy.assert_called_once_with( + f'{CLUSTER_ID}/{REPLICATION_POLICY_ID}') + + def test_failover(self, client, db, cluster, replication_policy, + replication_policy_controller): + replication_policy_controller.failover_policy.return_value = [ + {'lvol_id': VOLUME_ID, 'status': 'failed', 'detail': 'boom'}, + ] + + response = client.post(POLICIES_URL + f'{REPLICATION_POLICY_ID}/failover') + + assert response.status_code == 200 + (body,) = response.json() + assert body['status'] == 'failed' + assert body['detail'] == 'boom' + replication_policy_controller.failover_policy.assert_called_once_with( + f'{CLUSTER_ID}/{REPLICATION_POLICY_ID}') diff --git a/tests/unit/web/api/v2/test_volume_replication_endpoints.py b/tests/unit/web/api/v2/test_volume_replication_endpoints.py new file mode 100644 index 000000000..dc9d4dd5c --- /dev/null +++ b/tests/unit/web/api/v2/test_volume_replication_endpoints.py @@ -0,0 +1,288 @@ +# coding=utf-8 +"""Unit tests for the volume replication endpoints and the policy assignment +folded into the volume PUT.""" + +from simplyblock_core.controllers.replication_policy_controller import ReplicationConfigError + +from tests.unit.web.api.v2 import _factories as factories +from tests.unit.web.api.v2._factories import ( + CLUSTER_ID, + POOL_ID, + REPLICATION_POLICY_ID, + TARGET_CLUSTER_ID, + TASK_ID, + VOLUME_ID, +) + + +VOLUME_URL = f'/api/v2/clusters/{CLUSTER_ID}/storage-pools/{POOL_ID}/volumes/{VOLUME_ID}/' +REPLICATION_URL = VOLUME_URL + 'replication/' +TARGET_VOLUME_ID = '33333333-3333-3333-3333-333333333334' +REPLICATION_ID = 'abababab-abab-abab-abab-abababababab' + + +def _relationship(**overrides): + relationship = { + 'replication_id': REPLICATION_ID, + 'source_lvol_id': VOLUME_ID, + 'target_lvol_id': TARGET_VOLUME_ID, + 'source_cluster_id': CLUSTER_ID, + 'target_cluster_id': TARGET_CLUSTER_ID, + 'mode': 'failover', + 'state': 'replicating', + 'direction': 'to_target', + 'target_nqn': 'nqn.2023-02.io.simplyblock:volume-1', + 'target_ns_id': 1, + 'is_source': True, + } + relationship.update(overrides) + return relationship + + +class TestAssignPolicyThroughUpdate: + + def test_setting_the_policy_attaches_it(self, client, db, volume, lvol_controller, + replication_policy_controller): + response = client.put(VOLUME_URL, json={'replication_policy_id': REPLICATION_POLICY_ID}) + + assert response.status_code == 204 + replication_policy_controller.attach_policy.assert_called_once_with( + VOLUME_ID, REPLICATION_POLICY_ID) + + def test_null_policy_detaches_it(self, client, db, volume, lvol_controller, + replication_policy_controller): + response = client.put(VOLUME_URL, json={'replication_policy_id': None}) + + assert response.status_code == 204 + replication_policy_controller.detach_policy.assert_called_once_with(VOLUME_ID) + + def test_omitting_the_policy_leaves_it_alone(self, client, db, volume, lvol_controller, + replication_policy_controller): + response = client.put(VOLUME_URL, json={'name': 'volume-renamed'}) + + assert response.status_code == 204 + replication_policy_controller.attach_policy.assert_not_called() + replication_policy_controller.detach_policy.assert_not_called() + + def test_non_uuid_policy_rejected(self, client, db, volume, lvol_controller, + replication_policy_controller): + response = client.put(VOLUME_URL, json={'replication_policy_id': 'nightly'}) + + assert response.status_code == 422 + replication_policy_controller.attach_policy.assert_not_called() + + def test_attach_config_error_maps_to_400(self, client, db, volume, lvol_controller, + replication_policy_controller): + replication_policy_controller.attach_policy.side_effect = \ + ReplicationConfigError('Replication policy nightly is not active') + + response = client.put(VOLUME_URL, json={'replication_policy_id': REPLICATION_POLICY_ID}) + + assert response.status_code == 400 + + def test_unknown_policy_maps_to_404(self, client, db, volume, lvol_controller, + replication_policy_controller): + replication_policy_controller.attach_policy.side_effect = KeyError('policy not found') + + response = client.put(VOLUME_URL, json={'replication_policy_id': REPLICATION_POLICY_ID}) + + assert response.status_code == 404 + + def test_detach_during_cutover_maps_to_409(self, client, db, volume, lvol_controller, + replication_policy_controller): + replication_policy_controller.detach_policy.side_effect = \ + ReplicationConfigError('cutover in flight') + + response = client.put(VOLUME_URL, json={'replication_policy_id': None}) + + assert response.status_code == 409 + + def test_size_and_policy_are_applied_together(self, client, db, volume, lvol_controller, + replication_policy_controller): + response = client.put(VOLUME_URL, json={ + 'size': '20G', + 'replication_policy_id': REPLICATION_POLICY_ID, + }) + + assert response.status_code == 204 + lvol_controller.resize_lvol.assert_called_once() + replication_policy_controller.attach_policy.assert_called_once() + + +class TestRelationship: + + def test_returns_the_relationship(self, client, db, volume, replication_policy_controller): + replication_policy_controller.get_relationship.return_value = _relationship() + + response = client.get(REPLICATION_URL) + + assert response.status_code == 200 + body = response.json() + assert body['replication_id'] == REPLICATION_ID + assert body['source_lvol_id'] == VOLUME_ID + assert body['target_lvol_id'] == TARGET_VOLUME_ID + assert body['state'] == 'replicating' + assert body['direction'] == 'to_target' + assert body['is_source'] is True + + def test_blank_counterpart_serializes_as_null(self, client, db, volume, + replication_policy_controller): + replication_policy_controller.get_relationship.return_value = \ + _relationship(target_lvol_id='') + + body = client.get(REPLICATION_URL).json() + + assert body['target_lvol_id'] is None + + def test_missing_relationship_maps_to_404(self, client, db, volume, + replication_policy_controller): + replication_policy_controller.get_relationship.return_value = None + + assert client.get(REPLICATION_URL).status_code == 404 + + +class TestStartStopTrigger: + + def test_start_passes_parameters(self, client, db, volume, lvol_controller): + response = client.post(REPLICATION_URL + 'start', json={ + 'replication_cluster_id': TARGET_CLUSTER_ID, + 'mode': 'migration', + 'interval_min': 15, + }) + + assert response.status_code == 204 + args, kwargs = lvol_controller.replication_start.call_args + assert args == (VOLUME_ID,) + assert kwargs == { + 'replication_cluster_id': TARGET_CLUSTER_ID, + 'mode': 'migration', + 'interval_min': 15, + } + + def test_start_without_body_uses_cluster_default(self, client, db, volume, lvol_controller): + response = client.post(REPLICATION_URL + 'start') + + assert response.status_code == 204 + assert lvol_controller.replication_start.call_args.kwargs == { + 'replication_cluster_id': None, 'mode': None, 'interval_min': None, + } + + def test_unknown_mode_rejected(self, client, db, volume, lvol_controller): + response = client.post(REPLICATION_URL + 'start', json={'mode': 'sideways'}) + + assert response.status_code == 422 + lvol_controller.replication_start.assert_not_called() + + def test_negative_interval_rejected(self, client, db, volume, lvol_controller): + response = client.post(REPLICATION_URL + 'start', json={'interval_min': -1}) + + assert response.status_code == 422 + lvol_controller.replication_start.assert_not_called() + + def test_stop(self, client, db, volume, lvol_controller): + response = client.post(REPLICATION_URL + 'stop') + + assert response.status_code == 204 + lvol_controller.replication_stop.assert_called_once_with(VOLUME_ID) + + def test_trigger(self, client, db, volume, lvol_controller): + response = client.post(REPLICATION_URL + 'trigger') + + assert response.status_code == 204 + lvol_controller.replication_trigger.assert_called_once_with(VOLUME_ID) + + +class TestCutover: + + def test_failover(self, client, db, volume, lvol_controller): + lvol_controller.replicate_lvol_on_target_cluster.return_value = { + 'lvol_id': TARGET_VOLUME_ID, 'nqn': 'nqn.x', 'ns_id': 1, 'connection_strings': [], + } + + response = client.post(REPLICATION_URL + 'failover') + + assert response.status_code == 204 + assert response.content == b'' + lvol_controller.replicate_lvol_on_target_cluster.assert_called_once_with(VOLUME_ID) + + def test_failed_failover_is_an_error(self, client, db, volume, lvol_controller): + lvol_controller.replicate_lvol_on_target_cluster.return_value = (False, 'node is not online') + + response = client.post(REPLICATION_URL + 'failover') + + assert response.status_code == 500 + assert response.json()['detail'] == 'node is not online' + + def test_commit_points_at_the_cutover_task(self, client, db, volume, lvol_controller): + lvol_controller.replication_commit.return_value = { + 'cutover_task_queued': True, 'task_id': TASK_ID, + } + + response = client.post(REPLICATION_URL + 'commit') + + assert response.status_code == 202 + assert response.content == b'' + assert response.headers['Location'].endswith(f'/clusters/{CLUSTER_ID}/tasks/{TASK_ID}/') + lvol_controller.replication_commit.assert_called_once_with(VOLUME_ID) + + def test_unqueued_cutover_is_an_error(self, client, db, volume, lvol_controller): + lvol_controller.replication_commit.return_value = False + + assert client.post(REPLICATION_URL + 'commit').status_code == 500 + + def test_failback(self, client, db, volume, lvol_controller): + lvol_controller.replication_failback.return_value = True + + response = client.post(REPLICATION_URL + 'failback', + json={'source_cluster_id': TARGET_CLUSTER_ID}) + + assert response.status_code == 204 + lvol_controller.replication_failback.assert_called_once_with( + VOLUME_ID, source_cluster_id=TARGET_CLUSTER_ID) + + def test_failback_without_source_cluster(self, client, db, volume, lvol_controller): + lvol_controller.replication_failback.return_value = True + + response = client.post(REPLICATION_URL + 'failback', json={}) + + assert response.status_code == 204 + lvol_controller.replication_failback.assert_called_once_with( + VOLUME_ID, source_cluster_id=None) + + def test_failed_failback_is_an_error(self, client, db, volume, lvol_controller): + lvol_controller.replication_failback.return_value = False + + assert client.post(REPLICATION_URL + 'failback', json={}).status_code == 500 + + +class TestTasks: + + def test_lists_replication_tasks(self, client, db, volume, lvol_controller): + lvol_controller.list_replication_tasks.return_value = [factories.make_task()] + + response = client.get(REPLICATION_URL + 'tasks') + + assert response.status_code == 200 + (body,) = response.json() + assert body['status'] == 'new' + lvol_controller.list_replication_tasks.assert_called_once_with(VOLUME_ID) + + +class TestReplicateOnSourceCluster: + + def test_passes_the_volume_id(self, client, db, pool, lvol_controller): + url = f'/api/v2/clusters/{CLUSTER_ID}/storage-pools/{POOL_ID}/volumes/replicate_lvol_on_source_cluster' + + response = client.post(url, json={'lvol_id': VOLUME_ID}) + + assert response.status_code == 204 + lvol_controller.replicate_lvol_on_source_cluster.assert_called_once_with( + VOLUME_ID, CLUSTER_ID, POOL_ID) + + def test_missing_volume_id_rejected(self, client, db, pool, lvol_controller): + url = f'/api/v2/clusters/{CLUSTER_ID}/storage-pools/{POOL_ID}/volumes/replicate_lvol_on_source_cluster' + + response = client.post(url, json={}) + + assert response.status_code == 422 + lvol_controller.replicate_lvol_on_source_cluster.assert_not_called()