feat: call TF-SDK activate_capability when completing service deploy. removed... (65a2c619) · Commits · OOP / code / service-resource-manager

src/srm/api/databus/dependencies.py

+11 −13

Changes for src/srm/api/databus/dependencies.py: 11 added lines, 13 removed lines.
Original line number	Diff line number	Diff line
		@@ -11,8 +11,6 @@ from srm.adapters.database.repos.runtime_inventory import (
		SqlServiceOrderRepository,
		)
		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.adapters.federation_manager.fm_client import FederationManager
		from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator
		from srm.application.command_handlers.federated_deploy_outcome import (
		@@ -35,20 +33,11 @@ from srm.domain.ports.databus.publisher import DataBusPublisher
		from srm.domain.ports.transformation_functions import ControlPathRegistry


		def get_deploy_service_use_case(
		session: AsyncSession, connection_manager: NatsConnectionManager, fm_client: FederationManager
		) -> DeployServiceUseCase:
		return build_deploy_service_use_case(
		session,
		NatsPublisher(connection_manager),
		fm_client=fm_client,
		)


		def build_deploy_service_use_case(
		session: AsyncSession,
		publisher: DataBusPublisher,
		fm_client: FederationManager,
		control_paths: ControlPathRegistry,
		) -> DeployServiceUseCase:
		return DeployServiceUseCase(
		service_specifications=SqlServiceSpecificationRepository(session),
		@@ -60,6 +49,7 @@ def build_deploy_service_use_case(
		zones=SqlZoneRepository(session),
		publisher=publisher,
		fm_client=fm_client,
		control_paths=control_paths,
		)


		@@ -67,10 +57,18 @@ def get_deploy_service_command_coordinator(
		session_maker: async_sessionmaker[AsyncSession],
		pub: DataBusPublisher,
		fm_client: FederationManager,
		control_paths: ControlPathRegistry,
		) -> DeployServiceCommandCoordinator:
		def use_case_factory(
		session: AsyncSession,
		publisher: DataBusPublisher,
		fm_client: FederationManager,
		) -> DeployServiceUseCase:
		return build_deploy_service_use_case(session, publisher, fm_client, control_paths)

		return DeployServiceCommandCoordinator(
		session_maker=session_maker,
		use_case_factory=build_deploy_service_use_case,
		use_case_factory=use_case_factory,
		publisher=pub,
		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
		@@ -59,6 +59,7 @@ async def subscribe_to_subjects(
		session_maker,
		pub=publisher,
		fm_client=fm_client,
		control_paths=control_paths,
		)
		terminate_coordinator = get_terminate_service_command_coordinator(
		session_maker,

src/srm/application/use_cases/deploy_service.py

+132 −6

Changes for src/srm/application/use_cases/deploy_service.py: 132 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -6,6 +6,7 @@ from typing import Any, Literal
		from uuid import UUID

		import structlog
		from anyio import to_thread
		from fastapi.encoders import jsonable_encoder

		from srm.adapters.federation_manager.fm_client import FederationManager
		@@ -23,6 +24,7 @@ from srm.domain.models.canonical_parameters.result import Result, ResultStatus
		from srm.domain.models.catalog import (
		RuntimeKind,
		ServiceCapabilityRequirement,
		ServiceDeploymentUnit,
		ServiceSpecificationState,
		)
		from srm.domain.models.runtime_inventory import CapabilityInstance, ServiceInstance, ServiceOrder
		@@ -52,9 +54,46 @@ from srm.domain.ports.databus.events import (
		SrmOperationStatusV1,
		)
		from srm.domain.ports.databus.publisher import DataBusPublisher
		from srm.domain.ports.transformation_functions import (
		CapabilityLifecycleRequest,
		ControlPathRegistry,
		DeploymentUnit,
		DeploymentUnitType,
		DomainExecutionFailed,
		ResolvedCapabilityTarget,
		)

		logger: structlog.BoundLogger = structlog.get_logger(__name__)

		_ACTIVATE_ACTION = "activate"


		def _deployment_unit_type(runtime_kind: RuntimeKind) -> DeploymentUnitType:
		if runtime_kind is RuntimeKind.HELM:
		return DeploymentUnitType.HELM_CHART
		if runtime_kind is RuntimeKind.K8S_MANIFEST:
		return DeploymentUnitType.MANIFEST_BUNDLE
		return DeploymentUnitType.COMPONENT


		def _resolved_target(
		placement: CapabilityPlacement,
		*,
		capability_kind: CapabilityKind,
		runtime_kind: RuntimeKind,
		) -> ResolvedCapabilityTarget:
		return ResolvedCapabilityTarget(
		zone_ref=placement.zone_ref,
		zone_kind=placement.zone_kind,
		domain_ref=placement.domain_ref,
		domain_kind=placement.domain_kind,
		capability_ref=placement.capability_ref,
		capability_kind=capability_kind,
		control_path_ref=placement.control_path_ref,
		action=_ACTIVATE_ACTION,
		runtime_kind=runtime_kind,
		)


		@dataclass(frozen=True, slots=True)
		class DeployTargetCommand:
		@@ -120,6 +159,7 @@ class DeployServiceUseCase:
		zones: ZoneRepository,
		publisher: DataBusPublisher,
		fm_client: FederationManager,
		control_paths: ControlPathRegistry,
		placement_planner: CapabilityPlacementPlanner \| None = None,
		) -> None:
		self._service_specifications = service_specifications
		@@ -131,6 +171,7 @@ class DeployServiceUseCase:
		self._zones = zones
		self._publisher = publisher
		self._fm_client = fm_client
		self._control_paths = control_paths
		self._placement_planner = placement_planner or CapabilityPlacementPlanner(zones)

		async def accept(self, command: DeployServiceCommand) -> DeployServiceResult:
		@@ -427,12 +468,76 @@ class DeployServiceUseCase:
		target_placement: DeployTargetPlacement,
		deploy_material: list[tuple[ServiceCapabilityRequirement, RuntimeKind]],
		) -> None:
		# TODO: TF-SDK CapabilityExecutionAdapter.activate_capability(...) call using
		# capability_kind="workload" and placement.control_path_ref.
		requirements_by_id = {requirement.id: requirement for requirement, _ in deploy_material}
		runtime_kind_by_requirement_id = {
		requirement.id: runtime_kind for requirement, runtime_kind in deploy_material
		}
		units_by_id = {
		unit.id: unit
		for unit in await self._service_deployment_units.list_by_service_specification_id(
		command.service_specification_id
		)
		}
		release_name = command.deploy.instance_name or str(service_instance.id)

		for requirement_placement in target_placement.requirements:
		requirement = requirements_by_id[requirement_placement.requirement_id]
		placement = requirement_placement.placement
		runtime_kind = runtime_kind_by_requirement_id[requirement.id]
		catalog_unit = self._catalog_unit_for_requirement(requirement, units_by_id)
		adapter = self._control_paths.get_adapter(placement.control_path_ref)
		logger.info(
		"workload_adapter_activate",
		operation_id=str(command.operation_id),
		correlation_id=command.correlation_id,
		control_path_ref=placement.control_path_ref,
		adapter=type(adapter).__name__,
		deployment_unit_ref=release_name,
		artifact_ref=catalog_unit.artifact_ref,
		capability_kind=requirement.capability_kind,
		)
		result = await to_thread.run_sync(
		adapter.activate_capability,
		CapabilityLifecycleRequest(
		operation_id=str(command.operation_id),
		tenant_id=command.app_provider_id,
		capability_kind=requirement.capability_kind,
		target=_resolved_target(
		placement,
		capability_kind=requirement.capability_kind,
		runtime_kind=runtime_kind,
		),
		deployment_unit=DeploymentUnit(
		deployment_unit_ref=release_name,
		deployment_unit_type=_deployment_unit_type(runtime_kind),
		runtime_type=runtime_kind,
		artifact_ref=catalog_unit.artifact_ref,
		resource_requirements=catalog_unit.resource_requirements,
		),
		parameters=requirement.parameters,
		),
		)
		logger.info(
		"workload_adapter_activate_result",
		operation_id=str(command.operation_id),
		correlation_id=command.correlation_id,
		control_path_ref=placement.control_path_ref,
		adapter=type(adapter).__name__,
		deployment_unit_ref=release_name,
		state=result.state,
		external_id=result.external_id,
		external_ref=result.external_ref,
		details=result.details,
		)
		if result.state is not CapabilityInstanceState.ACTIVE:
		raise DomainExecutionFailed(
		error_code="domain_execution_failed",
		message=(
		"workload activate did not become active on "
		f"{placement.control_path_ref}: {result.state}"
		),
		details=result.details,
		)
		await self._capability_instances.create(
		CapabilityInstance(
		capability_id=placement.capability_id,
		@@ -443,16 +548,37 @@ class DeployServiceUseCase:
		control_path_ref_snapshot=placement.control_path_ref,
		ref=requirement.ref,
		kind=CapabilityInstanceKind.ACTIVATION,
		state=CapabilityInstanceState.ACTIVE,
		external_id=None,
		external_ref=None,
		state=result.state,
		external_id=result.external_id,
		external_ref=result.external_ref,
		parameters_snapshot=requirement.parameters,
		result_summary=Result(status=ResultStatus.ACTIVE),
		result_summary=result.result_summary
		or Result(
		status=ResultStatus.ACTIVE,
		endpoints=result.endpoints,
		backend=result.details,
		),
		app_provider_id=command.app_provider_id,
		federation_partner_ref=command.federation_partner_ref,
		)
		)

		def _catalog_unit_for_requirement(
		self,
		requirement: ServiceCapabilityRequirement,
		units_by_id: dict[UUID, ServiceDeploymentUnit],
		) -> ServiceDeploymentUnit:
		if requirement.deployment_unit_id is not None:
		try:
		return units_by_id[requirement.deployment_unit_id]
		except KeyError as exc:
		raise RuntimeError(
		"Deploy requirement references a missing deployment unit"
		) from exc
		if not units_by_id:
		raise RuntimeError("No deployment unit available for workload activate")
		return next(iter(units_by_id.values()))

		async def _resolve_deploy_requirements(
		self,
		service_specification_id: UUID,

tests/api/databus/test_dependencies.py

+8 −8

Changes for tests/api/databus/test_dependencies.py: 8 added lines, 8 removed lines.
Original line number	Diff line number	Diff line
		from unittest.mock import MagicMock, patch

		from srm.api.databus.dependencies import (
		build_deploy_service_use_case,
		build_network_capability_activate_use_case,
		get_deploy_service_use_case,
		get_terminate_service_use_case,
		)


		def test_get_deploy_service_use_case_wires_required_repositories() -> None:
		def test_build_deploy_service_use_case_wires_required_repositories() -> None:
		session = MagicMock()
		connection_manager = MagicMock()
		publisher = MagicMock()
		fm_client = MagicMock()
		control_paths = MagicMock()

		with (
		patch("srm.api.databus.dependencies.SqlServiceSpecificationRepository") as specs,
		@@ -21,11 +23,9 @@ def test_get_deploy_service_use_case_wires_required_repositories() -> None:
		patch("srm.api.databus.dependencies.SqlServiceInstanceRepository") as instances,
		patch("srm.api.databus.dependencies.SqlCapabilityInstanceRepository") as capabilities,
		patch("srm.api.databus.dependencies.SqlZoneRepository") as zones,
		patch("srm.api.databus.dependencies.NatsPublisher") as publisher,
		patch("srm.api.databus.dependencies.FederationManager") as fm_client,
		patch("srm.api.databus.dependencies.DeployServiceUseCase") as use_case,
		):
		result = get_deploy_service_use_case(session, connection_manager, fm_client)
		result = build_deploy_service_use_case(session, publisher, fm_client, control_paths)

		specs.assert_called_once_with(session)
		units.assert_called_once_with(session)
		@@ -34,7 +34,6 @@ def test_get_deploy_service_use_case_wires_required_repositories() -> None:
		instances.assert_called_once_with(session)
		capabilities.assert_called_once_with(session)
		zones.assert_called_once_with(session)
		publisher.assert_called_once_with(connection_manager)
		use_case.assert_called_once_with(
		service_specifications=specs.return_value,
		service_deployment_units=units.return_value,
		@@ -44,7 +43,8 @@ def test_get_deploy_service_use_case_wires_required_repositories() -> None:
		capability_instances=capabilities.return_value,
		zones=zones.return_value,
		fm_client=fm_client,
		publisher=publisher.return_value,
		publisher=publisher,
		control_paths=control_paths,
		)
		assert result is use_case.return_value

tests/api/databus/test_nats_subscriber.py

+3 −1

Changes for tests/api/databus/test_nats_subscriber.py: 3 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -325,7 +325,9 @@ async def test_deploy_subscription_callback_delegates_to_command_coordinator(
		)
		await deploy_subscriber._callback(envelope)

		coordinator_factory.assert_called_once_with(session_maker, pub=publisher, fm_client=fm_client)
		coordinator_factory.assert_called_once_with(
		session_maker, pub=publisher, fm_client=fm_client, control_paths=control_paths
		)
		coordinator.handle.assert_awaited_once_with(envelope)
		assert session_maker.contexts == []