feat: terminate non-local service instances through fm (0623ae40) · Commits · OOP / code / service-resource-manager

src/srm/adapters/federation_manager/fm_client.py

+7 −0

Changes for src/srm/adapters/federation_manager/fm_client.py: 7 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -167,3 +167,10 @@ class FederationManager(FederationManagerPort):
		command=command,
		)
		await self._publisher.publish("command.fm.service.deploy", command)

		async def request_terminate_service(self, command: dict[str, Any]) -> None:
		logger.info(
		"terminate_service_on_partner",
		command=command,
		)
		await self._publisher.publish("command.fm.service.terminate", command)

src/srm/api/databus/dependencies.py

+6 −1

Changes for src/srm/api/databus/dependencies.py: 6 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -119,30 +119,35 @@ def get_network_capability_activate_command_coordinator(
		def get_terminate_service_use_case(
		session: AsyncSession,
		connection_manager: NatsConnectionManager,
		fm_client: FederationManager,
		) -> TerminateServiceUseCase:
		return build_terminate_service_use_case(session, NatsPublisher(connection_manager))
		return build_terminate_service_use_case(session, NatsPublisher(connection_manager), fm_client)


		def build_terminate_service_use_case(
		session: AsyncSession,
		publisher: DataBusPublisher,
		fm_client: FederationManager,
		) -> TerminateServiceUseCase:
		return TerminateServiceUseCase(
		service_orders=SqlServiceOrderRepository(session),
		service_instances=SqlServiceInstanceRepository(session),
		capability_instances=SqlCapabilityInstanceRepository(session),
		publisher=publisher,
		fm_client=fm_client,
		)


		def get_terminate_service_command_coordinator(
		session_maker: async_sessionmaker[AsyncSession],
		connection_manager: NatsConnectionManager,
		fm_client: FederationManager,
		) -> TerminateServiceCommandCoordinator:
		return TerminateServiceCommandCoordinator(
		session_maker=session_maker,
		use_case_factory=build_terminate_service_use_case,
		publisher=NatsPublisher(connection_manager),
		fm_client=fm_client,
		)

src/srm/api/databus/nats_subscriber.py

+1 −0

Changes for src/srm/api/databus/nats_subscriber.py: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -64,6 +64,7 @@ async def subscribe_to_subjects(
		terminate_coordinator = get_terminate_service_command_coordinator(
		session_maker,
		connection_manager,
		fm_client,
		)
		network_activate_coordinator = get_network_capability_activate_command_coordinator(
		session_maker,

src/srm/application/command_handlers/terminate_service.py

+31 −17

Changes for src/srm/application/command_handlers/terminate_service.py: 31 added lines, 17 removed lines.
Original line number	Diff line number	Diff line
		@@ -2,11 +2,13 @@ from __future__ import annotations

		from collections.abc import Callable
		from datetime import UTC, datetime
		from typing import Any
		from uuid import UUID

		from pydantic import ValidationError
		from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

		from srm.adapters.federation_manager.fm_client import FederationManager
		from srm.api.databus.schemas import AnswerableCommandV1, SrmServiceTerminateV1
		from srm.application.use_cases.terminate_service import (
		TerminatePayloadCommand,
		@@ -22,7 +24,9 @@ from srm.domain.ports.databus.events import (
		)
		from srm.domain.ports.databus.publisher import DataBusPublisher

		TerminateUseCaseFactory = Callable[[AsyncSession, DataBusPublisher], TerminateServiceUseCase]
		TerminateUseCaseFactory = Callable[
		[AsyncSession, DataBusPublisher, FederationManager], TerminateServiceUseCase
		]


		class TerminateServiceCommandCoordinator:
		@@ -31,10 +35,12 @@ class TerminateServiceCommandCoordinator:
		session_maker: async_sessionmaker[AsyncSession],
		use_case_factory: TerminateUseCaseFactory,
		publisher: DataBusPublisher,
		fm_client: FederationManager,
		) -> None:
		self._session_maker = session_maker
		self._use_case_factory = use_case_factory
		self._publisher = publisher
		self._fm_client = fm_client

		async def handle(self, envelope: AnswerableCommandV1) -> None:
		command = await self._parse_command(envelope)
		@@ -96,7 +102,7 @@ class TerminateServiceCommandCoordinator:
		async def _accept(self, command: TerminateServiceCommand) -> TerminateServiceResult:
		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher)
		use_case = self._use_case_factory(session, self._publisher, self._fm_client)
		accepted = await use_case.accept(command)
		await session.commit()
		return accepted
		@@ -109,8 +115,19 @@ class TerminateServiceCommandCoordinator:
		command: TerminateServiceCommand,
		accepted: TerminateServiceResult,
		) -> SrmOperationCompletedV1 \| None:
		if accepted.is_federated:
		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher, self._fm_client)
		completed_event = await use_case.hand_off_to_partner(command, accepted)
		await session.commit()
		return completed_event
		except Exception:
		await session.rollback()
		raise

		async with self._session_maker() as session:
		use_case = self._use_case_factory(session, self._publisher)
		use_case = self._use_case_factory(session, self._publisher, self._fm_client)
		plan = await use_case.prepare_fulfill(command, accepted)

		if plan is None:
		@@ -125,7 +142,7 @@ class TerminateServiceCommandCoordinator:

		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher)
		use_case = self._use_case_factory(session, self._publisher, self._fm_client)
		await use_case.persist_capability_terminated(
		target.capability_instance_id,
		terminated_at,
		@@ -137,7 +154,7 @@ class TerminateServiceCommandCoordinator:

		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher)
		use_case = self._use_case_factory(session, self._publisher, self._fm_client)
		completed_event = await use_case.persist_completed(
		command,
		accepted,
		@@ -157,7 +174,7 @@ class TerminateServiceCommandCoordinator:
		) -> SrmOperationCompletedV1:
		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher)
		use_case = self._use_case_factory(session, self._publisher, self._fm_client)
		completed_event = await use_case.fail_after_accept(command, accepted, error)
		await session.commit()
		return completed_event
		@@ -236,22 +253,19 @@ class TerminateServiceCommandCoordinator:
		"status": 500,
		"detail": str(error),
		}
		instance_entry: dict[str, Any] = {
		"service_instance_id": str(command.service_instance_id),
		"status": "failed",
		"error": problem,
		}
		if accepted.target_zone_id is not None:
		instance_entry["zone_id"] = str(accepted.target_zone_id)
		return SrmOperationCompletedV1(
		schema_version="1.0",
		operation_id=command.operation_id,
		status=OperationCompletedStatus.FAILED,
		service_order_id=accepted.accepted_order_id,
		instances=[
		{
		"service_instance_id": str(command.service_instance_id),
		"zone_id": str(accepted.target_zone_id),
		"status": "failed",
		"error": problem,
		}
		]
		if accepted.target_zone_id is not None
		else [],
		error=None if accepted.target_zone_id is not None else problem,
		instances=[instance_entry],
		correlation_id=command.correlation_id,
		completed_at=now,
		)

src/srm/application/use_cases/federated_deploy_outcome.py

+30 −11

Changes for src/srm/application/use_cases/federated_deploy_outcome.py: 30 added lines, 11 removed lines.
Original line number	Diff line number	Diff line
		@@ -37,11 +37,32 @@ def build_federated_handoff(federated_targets: list[tuple[UUID, UUID]]) -> dict[
		}


		def build_terminate_federated_handoff(app_instance_id: UUID) -> dict[str, Any]:
		return {
		_HANDOFF_TARGETS_KEY: [
		{
		"app_instance_id": str(app_instance_id),
		"status": "pending",
		}
		]
		}


		def federated_handoff_targets(federated_handoff: dict[str, Any]) -> list[dict[str, Any]]:
		return list(federated_handoff.get(_HANDOFF_TARGETS_KEY, []))


		# Finishes a deploy SRM handed to FM, once FM reports what the partner did.
		def _handoff_instance_entry(target: dict[str, Any]) -> dict[str, Any]:
		entry: dict[str, Any] = {
		"service_instance_id": str(target["app_instance_id"]),
		"status": "completed" if target.get("status") == "completed" else "failed",
		}
		if target.get("zone_id") is not None:
		entry["zone_id"] = str(target["zone_id"])
		return entry


		# Finishes a deploy/terminate SRM handed to FM, once FM reports what the partner did.
		class FederatedDeployOutcomeUseCase:
		def __init__(
		self,
		@@ -52,7 +73,7 @@ class FederatedDeployOutcomeUseCase:
		self._service_instances = service_instances

		async def execute(self, outcome: FmOperationCompletedV1) -> SrmOperationCompletedV1 \| None:
		if outcome.operation != "deploy":
		if outcome.operation not in ("deploy", "terminate"):
		return None
		order = await self._service_orders.get_by_operation_id(outcome.operation_id)
		if order is None or order.state in _TERMINAL:
		@@ -101,14 +122,7 @@ class FederatedDeployOutcomeUseCase:
		}
		for instance in local_instances
		]
		instances.extend(
		{
		"service_instance_id": str(target["app_instance_id"]),
		"zone_id": str(target["zone_id"]),
		"status": "completed" if target.get("status") == "completed" else "failed",
		}
		for target in federated_targets
		)
		instances.extend(_handoff_instance_entry(target) for target in federated_targets)

		succeeded = [entry for entry in instances if entry["status"] == "completed"]
		if len(succeeded) == len(instances):
		@@ -127,8 +141,12 @@ class FederatedDeployOutcomeUseCase:
		completed_at=completed_at,
		failure_detail=outcome.error if status == OperationCompletedStatus.FAILED else None,
		)
		metadata: dict[str, Any] = {}
		if "completion_basis" in outcome.metadata:
		metadata["completion_basis"] = outcome.metadata["completion_basis"]
		logger.info(
		"federated_deploy_completed",
		"federated_lifecycle_completed",
		operation=outcome.operation,
		operation_id=str(outcome.operation_id),
		status=status.value,
		)
		@@ -138,6 +156,7 @@ class FederatedDeployOutcomeUseCase:
		status=status,
		service_order_id=order.id,
		instances=instances,
		metadata=metadata,
		error=outcome.error if status == OperationCompletedStatus.FAILED else None,
		correlation_id=outcome.correlation_id,
		completed_at=completed_at,