feat: scale service message handling (57e1a74f) · Commits · OOP / code / service-resource-manager

src/srm/api/databus/dependencies.py

+32 −0

Changes for src/srm/api/databus/dependencies.py: 32 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -14,8 +14,10 @@ from srm.adapters.database.repos.topology import SqlZoneRepository
		from srm.adapters.databus.nats_connection_manager import NatsConnectionManager
		from srm.adapters.databus.nats_publisher import NatsPublisher
		from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator
		from srm.application.command_handlers.scale_service import ScaleServiceCommandCoordinator
		from srm.application.command_handlers.terminate_service import TerminateServiceCommandCoordinator
		from srm.application.use_cases.deploy_service import DeployServiceUseCase
		from srm.application.use_cases.scale_service import ScaleServiceUseCase
		from srm.application.use_cases.terminate_service import TerminateServiceUseCase
		from srm.domain.ports.databus.publisher import DataBusPublisher

		@@ -54,6 +56,36 @@ def get_deploy_service_command_coordinator(
		)


		def get_scale_service_use_case(
		session: AsyncSession,
		connection_manager: NatsConnectionManager,
		) -> ScaleServiceUseCase:
		return build_scale_service_use_case(session, NatsPublisher(connection_manager))


		def build_scale_service_use_case(
		session: AsyncSession,
		publisher: DataBusPublisher,
		) -> ScaleServiceUseCase:
		return ScaleServiceUseCase(
		service_orders=SqlServiceOrderRepository(session),
		service_instances=SqlServiceInstanceRepository(session),
		capability_instances=SqlCapabilityInstanceRepository(session),
		publisher=publisher,
		)


		def get_scale_service_command_coordinator(
		session_maker: async_sessionmaker[AsyncSession],
		connection_manager: NatsConnectionManager,
		) -> ScaleServiceCommandCoordinator:
		return ScaleServiceCommandCoordinator(
		session_maker=session_maker,
		use_case_factory=build_scale_service_use_case,
		publisher=NatsPublisher(connection_manager),
		)


		def get_terminate_service_use_case(
		session: AsyncSession,
		connection_manager: NatsConnectionManager,

src/srm/api/databus/nats_subscriber.py

+7 −1

Changes for src/srm/api/databus/nats_subscriber.py: 7 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -9,6 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
		from srm.adapters.databus.nats_connection_manager import NatsConnectionManager
		from srm.api.databus.dependencies import (
		get_deploy_service_command_coordinator,
		get_scale_service_command_coordinator,
		get_terminate_service_command_coordinator,
		)
		from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage
		@@ -29,7 +30,6 @@ async def _noop_callback(envelope: AnswerableCommandV1) -> None:


		COMMAND_HANDLERS: dict[str, MessageHandler] = {
		"command.srm.service.scale": _noop_callback,
		"command.srm.network.capability.activate": _noop_callback,
		"command.srm.network.capability.update": _noop_callback,
		"command.srm.network.capability.deactivate": _noop_callback,
		@@ -37,6 +37,7 @@ COMMAND_HANDLERS: dict[str, MessageHandler] = {

		COMMAND_SUBJECTS = (
		"command.srm.service.deploy",
		"command.srm.service.scale",
		"command.srm.service.terminate",
		*COMMAND_HANDLERS.keys(),
		)
		@@ -50,6 +51,10 @@ async def subscribe_to_subjects(
		session_maker,
		connection_manager,
		)
		scale_coordinator = get_scale_service_command_coordinator(
		session_maker,
		connection_manager,
		)
		terminate_coordinator = get_terminate_service_command_coordinator(
		session_maker,
		connection_manager,
		@@ -57,6 +62,7 @@ async def subscribe_to_subjects(

		command_handlers = {
		"command.srm.service.deploy": deploy_coordinator.handle,
		"command.srm.service.scale": scale_coordinator.handle,
		"command.srm.service.terminate": terminate_coordinator.handle,
		**COMMAND_HANDLERS,
		}

src/srm/application/command_handlers/scale_service.py

0 → 100644

+223 −0

Changes for src/srm/application/command_handlers/scale_service.py: 223 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from __future__ import annotations

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

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

		from srm.api.databus.schemas import AnswerableCommandV1, SrmServiceScaleV1
		from srm.application.use_cases.scale_service import (
		ScalePayloadCommand,
		ScaleServiceCommand,
		ScaleServiceResult,
		ScaleServiceUseCase,
		)
		from srm.domain.ports.databus.events import (
		OperationCompletedStatus,
		OperationStatusState,
		SrmOperationCompletedV1,
		SrmOperationStatusV1,
		)
		from srm.domain.ports.databus.publisher import DataBusPublisher

		ScaleUseCaseFactory = Callable[[AsyncSession, DataBusPublisher], ScaleServiceUseCase]


		class ScaleServiceCommandCoordinator:
		def __init__(
		self,
		session_maker: async_sessionmaker[AsyncSession],
		use_case_factory: ScaleUseCaseFactory,
		publisher: DataBusPublisher,
		) -> None:
		self._session_maker = session_maker
		self._use_case_factory = use_case_factory
		self._publisher = publisher

		async def handle(self, envelope: AnswerableCommandV1) -> None:
		command = await self._parse_command(envelope)
		if command is None:
		return

		accepted = await self._accept(command)
		if accepted.completed_event is not None:
		await self._publisher.publish(
		"event.srm.operation.completed",
		accepted.completed_event.model_dump(mode="json"),
		)
		return
		if accepted.duplicate_order_id is not None:
		await self._publish_accepted(envelope, accepted.duplicate_order_id)
		return
		if accepted.accepted_order_id is None:
		return

		await self._publish_accepted(envelope, accepted.accepted_order_id)

		try:
		completed_event = await self._complete(command, accepted)
		except Exception as exc:
		completed_event = await self._fail_after_accept(command, accepted, exc)
		if completed_event is not None:
		await self._publisher.publish(
		"event.srm.operation.completed",
		completed_event.model_dump(mode="json"),
		)

		async def _parse_command(
		self,
		envelope: AnswerableCommandV1,
		) -> ScaleServiceCommand \| None:
		try:
		scale_message = SrmServiceScaleV1.model_validate(envelope.model_dump(mode="python"))
		except ValidationError:
		await self._publish_failed_before_start_for_invalid_command(envelope)
		return None

		return ScaleServiceCommand(
		operation_id=scale_message.operation_id,
		correlation_id=scale_message.correlation_id,
		requested_at=scale_message.requested_at,
		app_provider_id=scale_message.app_provider_id,
		federation_partner_ref=scale_message.federation_partner_ref,
		source=scale_message.source,
		service_instance_id=scale_message.service_instance_id,
		service_specification_id=scale_message.service_specification_id,
		scale=ScalePayloadCommand(replicas=scale_message.scale.replicas),
		payload_snapshot=scale_message.model_dump(mode="json"),
		)

		async def _accept(self, command: ScaleServiceCommand) -> ScaleServiceResult:
		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher)
		accepted = await use_case.accept(command)
		await session.commit()
		return accepted
		except Exception:
		await session.rollback()
		raise

		async def _fail_after_accept(
		self,
		command: ScaleServiceCommand,
		accepted: ScaleServiceResult,
		error: Exception,
		) -> SrmOperationCompletedV1:
		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher)
		completed_event = await use_case.fail_after_accept(command, accepted, error)
		await session.commit()
		return completed_event
		except Exception:
		await session.rollback()
		return self._failed_after_accept_event(command, accepted, error)

		async def _complete(
		self,
		command: ScaleServiceCommand,
		accepted: ScaleServiceResult,
		) -> SrmOperationCompletedV1 \| None:
		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session, self._publisher)
		completed_event = await use_case.complete(command, accepted)
		await session.commit()
		return completed_event
		except Exception:
		await session.rollback()
		raise

		async def _publish_accepted(
		self,
		envelope: AnswerableCommandV1,
		service_order_id: UUID,
		) -> None:
		model_extra = envelope.model_extra or {}
		status_event = SrmOperationStatusV1(
		schema_version="1.0",
		operation_id=envelope.operation_id,
		service_order_id=service_order_id,
		service_instance_id=model_extra.get("service_instance_id"),
		capability="scale",
		state=OperationStatusState.ACCEPTED,
		correlation_id=envelope.correlation_id,
		emitted_at=datetime.now(UTC),
		)
		await self._publisher.publish(
		"event.srm.operation.status",
		status_event.model_dump(mode="json"),
		)

		async def _publish_failed_before_start_for_invalid_command(
		self,
		envelope: AnswerableCommandV1,
		) -> None:
		now = datetime.now(UTC)
		status_event = SrmOperationStatusV1(
		schema_version="1.0",
		operation_id=envelope.operation_id,
		service_order_id=None,
		service_instance_id=None,
		capability="scale",
		state=OperationStatusState.FAILED_BEFORE_START,
		correlation_id=envelope.correlation_id,
		emitted_at=now,
		)
		completed_event = SrmOperationCompletedV1(
		schema_version="1.0",
		operation_id=envelope.operation_id,
		status=OperationCompletedStatus.FAILED,
		service_order_id=None,
		error={
		"type": "about:blank",
		"title": "Scale service command is invalid.",
		"status": 400,
		},
		correlation_id=envelope.correlation_id,
		completed_at=now,
		)
		await self._publisher.publish(
		"event.srm.operation.status",
		status_event.model_dump(mode="json"),
		)
		await self._publisher.publish(
		"event.srm.operation.completed",
		completed_event.model_dump(mode="json"),
		)

		def _failed_after_accept_event(
		self,
		command: ScaleServiceCommand,
		accepted: ScaleServiceResult,
		error: Exception,
		) -> SrmOperationCompletedV1:
		now = datetime.now(UTC)
		problem = {
		"type": "about:blank",
		"title": "Scale service failed after acceptance.",
		"status": 500,
		"detail": str(error),
		}
		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,
		correlation_id=command.correlation_id,
		completed_at=now,
		)

src/srm/application/use_cases/scale_service.py

0 → 100644

+338 −0

Changes for src/srm/application/use_cases/scale_service.py: 338 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from __future__ import annotations

		from dataclasses import dataclass, field
		from datetime import UTC, datetime
		from typing import Any, Literal
		from uuid import UUID

		import structlog

		from srm.domain.models.runtime_inventory import ServiceInstance, ServiceOrder
		from srm.domain.models.runtime_inventory.enums import (
		CapabilityInstanceState,
		ServiceInstanceState,
		ServiceOrderState,
		ServiceOrderType,
		)
		from srm.domain.ports.database.runtime_inventory import (
		CapabilityInstanceRepository,
		ServiceInstanceRepository,
		ServiceOrderRepository,
		)
		from srm.domain.ports.databus.events import (
		OperationCompletedStatus,
		OperationStatusState,
		SrmOperationCompletedV1,
		SrmOperationStatusV1,
		)
		from srm.domain.ports.databus.publisher import DataBusPublisher


		@dataclass(frozen=True, slots=True)
		class ScalePayloadCommand:
		replicas: int


		@dataclass(frozen=True, slots=True)
		class ScaleServiceCommand:
		operation_id: UUID
		correlation_id: str
		requested_at: datetime
		app_provider_id: str
		source: Literal["nbi_camara", "nbi_tmf", "operator_portal", "federation"]
		service_instance_id: UUID
		scale: ScalePayloadCommand
		service_specification_id: UUID \| None = None
		federation_partner_ref: str \| None = None
		payload_snapshot: dict[str, Any] = field(default_factory=dict)


		@dataclass(frozen=True, slots=True)
		class ScaleServiceResult:
		accepted_order_id: UUID \| None = None
		duplicate_order_id: UUID \| None = None
		target_service_instance_id: UUID \| None = None
		target_zone_id: UUID \| None = None
		completed_event: SrmOperationCompletedV1 \| None = None


		class ScaleServiceUseCase:
		def __init__(
		self,
		service_orders: ServiceOrderRepository,
		service_instances: ServiceInstanceRepository,
		capability_instances: CapabilityInstanceRepository,
		publisher: DataBusPublisher,
		) -> None:
		self._service_orders = service_orders
		self._service_instances = service_instances
		self._capability_instances = capability_instances
		self._publisher = publisher

		async def accept(self, command: ScaleServiceCommand) -> ScaleServiceResult:
		existing_order = await self._service_orders.get_by_operation_id(command.operation_id)
		if existing_order is not None:
		return await self._replay_existing_order(command, existing_order)

		service_instance = await self._service_instances.get_by_id(command.service_instance_id)
		if service_instance is None:
		await self._publish_failed_before_start(command, service_order_id=None)
		return ScaleServiceResult()

		service_order = await self._service_orders.create(
		ServiceOrder(
		operation_id=command.operation_id,
		correlation_id=command.correlation_id,
		order_type=ServiceOrderType.SCALE_SERVICE,
		service_specification_id=(
		command.service_specification_id or service_instance.service_specification_id
		),
		target_service_instance_id=service_instance.id,
		state=ServiceOrderState.ACCEPTED,
		payload_snapshot=command.payload_snapshot,
		app_provider_id=command.app_provider_id,
		federation_partner_ref=command.federation_partner_ref,
		)
		)
		await self._mark_runtime_updating(service_instance)

		structlog.get_logger(__name__).info(
		"service_scale_accepted",
		operation_id=str(command.operation_id),
		service_instance_id=str(command.service_instance_id),
		replicas=command.scale.replicas,
		)
		return ScaleServiceResult(
		accepted_order_id=service_order.id,
		target_service_instance_id=service_instance.id,
		target_zone_id=service_instance.zone_id,
		)

		async def complete(
		self,
		command: ScaleServiceCommand,
		accepted: ScaleServiceResult,
		) -> SrmOperationCompletedV1 \| None:
		if accepted.accepted_order_id is None:
		return None

		service_order = await self._service_orders.get_by_id(accepted.accepted_order_id)
		if service_order is None:
		raise RuntimeError("Accepted scale service order could not be reloaded")

		service_instance = await self._service_instances.get_by_id(command.service_instance_id)
		if service_instance is None:
		raise RuntimeError("Accepted scale service instance could not be reloaded")

		completed_at = datetime.now(UTC)
		await self._scale_child_capabilities(command, service_instance)
		# Leave runtime rows in UPDATING here. A later polling/reconciliation
		# mechanism moves them back to ACTIVE after backend readiness is confirmed.
		await self._service_orders.update_state(
		service_order.id,
		ServiceOrderState.COMPLETED,
		completed_at=completed_at,
		)

		structlog.get_logger(__name__).info(
		"service_scale_completed",
		operation_id=str(command.operation_id),
		service_instance_id=str(command.service_instance_id),
		replicas=command.scale.replicas,
		)
		return SrmOperationCompletedV1(
		schema_version="1.0",
		operation_id=command.operation_id,
		status=OperationCompletedStatus.COMPLETED,
		service_order_id=service_order.id,
		instances=[
		{
		"service_instance_id": str(service_instance.id),
		"zone_id": str(service_instance.zone_id),
		"status": "completed",
		}
		],
		correlation_id=command.correlation_id,
		completed_at=completed_at,
		)

		async def fail_after_accept(
		self,
		command: ScaleServiceCommand,
		accepted: ScaleServiceResult,
		error: Exception,
		) -> SrmOperationCompletedV1:
		if accepted.accepted_order_id is None:
		raise RuntimeError("Cannot fail scale after accept without a service order")

		service_order = await self._service_orders.get_by_id(accepted.accepted_order_id)
		if service_order is None:
		raise RuntimeError("Accepted scale service order could not be reloaded")

		completed_at = datetime.now(UTC)
		problem = {
		"type": "about:blank",
		"title": "Scale service failed after acceptance.",
		"status": 500,
		"detail": str(error),
		}
		await self._service_orders.update_state(
		service_order.id,
		ServiceOrderState.FAILED,
		completed_at=completed_at,
		failure_detail=problem,
		)

		return SrmOperationCompletedV1(
		schema_version="1.0",
		operation_id=command.operation_id,
		status=OperationCompletedStatus.FAILED,
		service_order_id=service_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,
		correlation_id=command.correlation_id,
		completed_at=completed_at,
		)

		async def _replay_existing_order(
		self,
		command: ScaleServiceCommand,
		service_order: ServiceOrder,
		) -> ScaleServiceResult:
		if service_order.state == ServiceOrderState.COMPLETED:
		return ScaleServiceResult(
		duplicate_order_id=service_order.id,
		completed_event=await self._completed_event_from_order(
		command,
		service_order,
		status=OperationCompletedStatus.COMPLETED,
		),
		)
		if service_order.state == ServiceOrderState.FAILED:
		return ScaleServiceResult(
		duplicate_order_id=service_order.id,
		completed_event=await self._completed_event_from_order(
		command,
		service_order,
		status=OperationCompletedStatus.FAILED,
		),
		)

		return ScaleServiceResult(
		duplicate_order_id=service_order.id,
		target_service_instance_id=service_order.target_service_instance_id,
		)

		async def _completed_event_from_order(
		self,
		command: ScaleServiceCommand,
		service_order: ServiceOrder,
		*,
		status: OperationCompletedStatus,
		) -> SrmOperationCompletedV1:
		service_instance = (
		await self._service_instances.get_by_id(service_order.target_service_instance_id)
		if service_order.target_service_instance_id is not None
		else None
		)
		instance_status = "completed" if status == OperationCompletedStatus.COMPLETED else "failed"
		problem = service_order.failure_detail or {
		"type": "about:blank",
		"title": "Scale service failed.",
		"status": 500,
		}

		return SrmOperationCompletedV1(
		schema_version="1.0",
		operation_id=command.operation_id,
		status=status,
		service_order_id=service_order.id,
		instances=[
		{
		"service_instance_id": str(service_instance.id),
		"zone_id": str(service_instance.zone_id),
		"status": instance_status,
		**({"error": problem} if status == OperationCompletedStatus.FAILED else {}),
		}
		]
		if service_instance is not None
		else [],
		error=(
		problem
		if status == OperationCompletedStatus.FAILED and service_instance is None
		else None
		),
		correlation_id=command.correlation_id,
		completed_at=service_order.completed_at or datetime.now(UTC),
		)

		async def _mark_runtime_updating(self, service_instance: ServiceInstance) -> None:
		for capability in service_instance.capabilities:
		await self._capability_instances.update_state(
		capability.id,
		CapabilityInstanceState.UPDATING,
		)
		await self._service_instances.update_state(
		service_instance.id,
		ServiceInstanceState.UPDATING,
		)

		async def _scale_child_capabilities(
		self,
		command: ScaleServiceCommand,
		service_instance: ServiceInstance,
		) -> None:
		for capability in service_instance.capabilities:
		# TODO: TF-SDK CapabilityExecutionAdapter.update_capability/scale(...)
		# call using capability.external_id, command.scale.replicas, and
		# capability.control_path_ref_snapshot.
		pass

		async def _publish_failed_before_start(
		self,
		command: ScaleServiceCommand,
		*,
		service_order_id: UUID \| None,
		) -> None:
		now = datetime.now(UTC)
		problem = {
		"type": "about:blank",
		"title": "Scale service failed before start.",
		"status": 400,
		}
		status_event = SrmOperationStatusV1(
		schema_version="1.0",
		operation_id=command.operation_id,
		service_order_id=service_order_id,
		service_instance_id=command.service_instance_id,
		capability="scale",
		state=OperationStatusState.FAILED_BEFORE_START,
		correlation_id=command.correlation_id,
		emitted_at=now,
		)
		completed_event = SrmOperationCompletedV1(
		schema_version="1.0",
		operation_id=command.operation_id,
		status=OperationCompletedStatus.FAILED,
		service_order_id=service_order_id,
		error=problem,
		correlation_id=command.correlation_id,
		completed_at=now,
		)
		await self._publisher.publish(
		"event.srm.operation.status",
		status_event.model_dump(mode="json"),
		)
		await self._publisher.publish(
		"event.srm.operation.completed",
		completed_event.model_dump(mode="json"),
		)

tests/api/databus/test_dependencies.py

+27 −0

File changed.

Preview size limit exceeded, changes collapsed.