Commit 639a462f authored by Paris Stentoumis's avatar Paris Stentoumis
Browse files

Merge branch 'fix/partner-instance-lifecycle' into 'refactor/tf-sdk-wiring'

fix(federation): complete partner instance lifecycle correctly

See merge request !37
parents 65b61ae5 76b109b6
Loading
Loading
Loading
Loading
Loading
+6 −1
Changes for src/srm/application/use_cases/deploy_service.py: 6 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -307,7 +307,12 @@ class DeployServiceUseCase:
                    "service_instance_id": str(target["app_instance_id"]),
                    "zone_id": str(target["zone_id"]),
                    "status": "completed" if federated_status == "completed" else "failed",
                    **({"error": problem} if status == OperationCompletedStatus.FAILED else {}),
                    # a failed instance carries an error even when the order completed partly
                    **(
                        {"error": target.get("error") or problem}
                        if federated_status != "completed"
                        else {}
                    ),
                }
            )

+23 −3
Changes for src/srm/application/use_cases/federated_deploy_outcome.py: 23 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -20,6 +20,12 @@ logger: structlog.BoundLogger = structlog.get_logger(__name__)
_TERMINAL = {ServiceOrderState.COMPLETED, ServiceOrderState.FAILED, ServiceOrderState.CANCELLED}

_HANDOFF_TARGETS_KEY = "targets"
# SRM interface contract C.2: a failed instance carries an error; this when nobody said why
_UNEXPLAINED_FAILURE: dict[str, Any] = {
    "type": "about:blank",
    "title": "The instance failed.",
    "status": 500,
}


def build_federated_handoff(federated_targets: list[tuple[UUID, UUID]]) -> dict[str, Any]:
@@ -37,11 +43,15 @@ def build_federated_handoff(federated_targets: list[tuple[UUID, UUID]]) -> dict[
    }


def build_terminate_federated_handoff(app_instance_id: UUID) -> dict[str, Any]:
def build_terminate_federated_handoff(
    app_instance_id: UUID, zone_id: UUID | None = None
) -> dict[str, Any]:
    return {
        _HANDOFF_TARGETS_KEY: [
            {
                "app_instance_id": str(app_instance_id),
                # contract C.2: the completion names the instance's zone
                **({"zone_id": str(zone_id)} if zone_id is not None else {}),
                "status": "pending",
            }
        ]
@@ -59,6 +69,8 @@ def _handoff_instance_entry(target: dict[str, Any]) -> dict[str, Any]:
    }
    if target.get("zone_id") is not None:
        entry["zone_id"] = str(target["zone_id"])
    if entry["status"] == "failed":
        entry["error"] = target.get("error") or _UNEXPLAINED_FAILURE
    return entry


@@ -94,6 +106,9 @@ class FederatedDeployOutcomeUseCase:
            ):
                continue
            target["status"] = target_status
            if target_status == "failed":
                # kept with the target: the completion event needs it per instance
                target["error"] = outcome.error or _UNEXPLAINED_FAILURE
            matched = True
            if outcome.service_instance_id is not None:
                break
@@ -116,9 +131,14 @@ class FederatedDeployOutcomeUseCase:
            {
                "service_instance_id": str(instance.id),
                "zone_id": str(instance.zone_id),
                "status": "completed"
                "status": "completed",
            }
            if instance.state == ServiceInstanceState.ACTIVE
                else "failed",
            else {
                "service_instance_id": str(instance.id),
                "zone_id": str(instance.zone_id),
                "status": "failed",
                "error": _UNEXPLAINED_FAILURE,
            }
            for instance in local_instances
        ]
+28 −3
Changes for src/srm/application/use_cases/terminate_service.py: 28 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -9,6 +9,7 @@ import structlog
from anyio import to_thread
from fastapi.encoders import jsonable_encoder

from srm.adapters.errors import DownstreamServiceException, NotFoundException
from srm.adapters.federation_manager.fm_client import FederationManager
from srm.application.use_cases.federated_deploy_outcome import (
    build_terminate_federated_handoff,
@@ -139,17 +140,23 @@ class TerminateServiceUseCase:

        service_instance = await self._service_instances.get_by_id(command.service_instance_id)
        if service_instance is None:
            zone_id = await self._partner_hosted_zone(command.service_instance_id)
            if zone_id is None:
                await self._publish_failed_before_start(command, service_order_id=None)
                return TerminateServiceResult()
            service_order = await self._service_orders.create(
                ServiceOrder(
                    operation_id=command.operation_id,
                    correlation_id=command.correlation_id,
                    order_type=ServiceOrderType.TERMINATE_SERVICE,
                    service_specification_id=command.service_specification_id,
                    target_service_instance_id=command.service_instance_id,
                    # SRM keeps no row for a partner-hosted instance, so this can't point at
                    # one; the instance is the federated hand-off's target, as on deploy
                    target_service_instance_id=None,
                    state=ServiceOrderState.ACCEPTED,
                    payload_snapshot=command.payload_snapshot,
                    federated_handoff=build_terminate_federated_handoff(
                        command.service_instance_id
                        command.service_instance_id, zone_id
                    ),
                    app_provider_id=command.app_provider_id,
                    federation_partner_ref=command.federation_partner_ref,
@@ -206,6 +213,19 @@ class TerminateServiceUseCase:
            target_zone_id=service_instance.zone_id,
        )

    async def _partner_hosted_zone(self, service_instance_id: UUID) -> UUID | None:
        """The zone of an instance a partner hosts; None when FM doesn't know it either."""
        try:
            instance = await self._fm_client.get_service_instance(service_instance_id)
        except (NotFoundException, DownstreamServiceException) as error:
            logger.info(
                "service_terminate_instance_unknown",
                service_instance_id=str(service_instance_id),
                reason=type(error).__name__,
            )
            return None
        return instance.zone_id

    async def hand_off_to_partner(
        self,
        command: TerminateServiceCommand,
@@ -568,7 +588,12 @@ class TerminateServiceUseCase:
                entry: dict[str, Any] = {
                    "service_instance_id": str(target["app_instance_id"]),
                    "status": "completed" if federated_status == "completed" else "failed",
                    **({"error": problem} if status == OperationCompletedStatus.FAILED else {}),
                    # a failed instance carries an error even when the order completed partly
                    **(
                        {"error": target.get("error") or problem}
                        if federated_status != "completed"
                        else {}
                    ),
                }
                if target.get("zone_id") is not None:
                    entry["zone_id"] = str(target["zone_id"])
+50 −0
Changes for tests/integration/test_federated_terminate.py: 50 added lines, 0 removed lines.
Original line number Diff line number Diff line
"""Terminating a partner-hosted instance against the real schema, foreign keys included."""

from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
from uuid import uuid4

from sqlalchemy.ext.asyncio import AsyncSession

from srm.adapters.database.repos.runtime_inventory import (
    SqlServiceInstanceRepository,
    SqlServiceOrderRepository,
)
from srm.application.use_cases.terminate_service import TerminateServiceUseCase
from tests.unit.test_terminate_service_use_case import _command


async def test_terminating_a_partner_hosted_instance_stores_its_order(
    db_session: AsyncSession,
) -> None:
    # SRM keeps no service_instance row for an instance a partner hosts; FM knows its zone
    instance_id = uuid4()
    zone_id = uuid4()
    fm_client = AsyncMock()
    fm_client.get_service_instance.return_value = SimpleNamespace(zone_id=zone_id)
    orders = SqlServiceOrderRepository(db_session)
    use_case = TerminateServiceUseCase(
        service_orders=orders,
        service_instances=SqlServiceInstanceRepository(db_session),
        capability_instances=AsyncMock(),
        capabilities=AsyncMock(),
        domains=AsyncMock(),
        zones=AsyncMock(),
        publisher=AsyncMock(),
        fm_client=fm_client,
        control_paths=MagicMock(),
    )
    operation_id = uuid4()

    result = await use_case.accept(_command(operation_id, service_instance_id=instance_id))

    assert result.is_federated is True
    stored = await orders.get_by_operation_id(operation_id)
    assert stored is not None
    assert stored.federated_handoff == {
        "targets": [
            {"app_instance_id": str(instance_id), "zone_id": str(zone_id), "status": "pending"}
        ]
    }
+16 −0
Changes for tests/unit/test_federated_deploy_outcome.py: 16 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -104,6 +104,22 @@ async def test_a_failed_partner_deploy_fails_the_order() -> None:
    assert event is not None and event.status == OperationCompletedStatus.FAILED
    assert event.error == {"type": "x", "title": "partner failed"}
    assert orders.update_state.await_args.args[1] == ServiceOrderState.FAILED
    # contract C.2: OEG drops a completion whose failed instance has no error
    [entry] = event.instances
    assert entry["status"] == "failed"
    assert entry["error"] == {"type": "x", "title": "partner failed"}


async def test_a_failed_target_without_a_reason_still_carries_an_error() -> None:
    instance = uuid4()
    use_case, _, _ = await _setup(_order(federated_targets=[_pending(instance)]))
    outcome = _outcome("failed", instance).model_copy(update={"error": None})

    event = await use_case.execute(outcome)

    assert event is not None
    [entry] = event.instances
    assert entry["error"]["status"] == 500


async def test_an_outcome_for_a_finished_or_unknown_order_is_ignored() -> None:
Loading