Commit 42a0dc0f authored by Paris Stentoumis's avatar Paris Stentoumis
Browse files

fix: wire device-targeted placement into network capability activate 

Share the control-path registry from lifespan into databus subscribe,
map TargetResolutionFailure into distinct failed-before-start details,
and dump Parameters/Result JSONB with mode=json so IPv4Network fields serialize.
parent fd973f0e

src/srm/adapters/database/mappers.py

+3 −3
Changes for src/srm/adapters/database/mappers.py: 3 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -295,7 +295,7 @@ class ServiceCapabilityRequirementMapper: 
                domain_kind=domain.domain_kind,

                is_required=domain.is_required,

                selector=domain.selector,

                parameters=domain.parameters.model_dump(mode="python"),

                parameters=domain.parameters.model_dump(mode="json"),

                policy=domain.policy,

                capability_requirement_metadata=domain.metadata,

                created_at=domain.created_at,
@@ -436,8 +436,8 @@ class CapabilityInstanceMapper: 
                state=domain.state,

                external_id=domain.external_id,

                external_ref=domain.external_ref,

                parameters_snapshot=domain.parameters_snapshot.model_dump(mode="python"),

                result_summary=domain.result_summary.model_dump(mode="python"),

                parameters_snapshot=domain.parameters_snapshot.model_dump(mode="json"),

                result_summary=domain.result_summary.model_dump(mode="json"),

                failure_detail=domain.failure_detail,

                capability_instance_metadata=domain.metadata,

                app_provider_id=domain.app_provider_id,

src/srm/api/databus/dependencies.py

+21 −2
Changes for src/srm/api/databus/dependencies.py: 21 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -18,12 +18,17 @@ from srm.application.command_handlers.network_capability_activate import ( 
    NetworkCapabilityActivateCommandCoordinator,

)

from srm.application.command_handlers.terminate_service import TerminateServiceCommandCoordinator

from srm.application.services.capability_placement import CapabilityPlacementPlanner

from srm.application.services.device_targeted_control_path import (

    DeviceTargetedControlPathResolver,

)

from srm.application.use_cases.deploy_service import DeployServiceUseCase

from srm.application.use_cases.network_capability_activate import (

    NetworkCapabilityActivateUseCase,

)

from srm.application.use_cases.terminate_service import TerminateServiceUseCase

from srm.domain.ports.databus.publisher import DataBusPublisher

from srm.domain.ports.transformation_functions import ControlPathRegistry





def get_deploy_service_use_case(
@@ -63,25 +68,39 @@ def get_deploy_service_command_coordinator( 
def build_network_capability_activate_use_case(

    session: AsyncSession,

    publisher: DataBusPublisher,

    control_paths: ControlPathRegistry,

) -> NetworkCapabilityActivateUseCase:

    zones = SqlZoneRepository(session)

    return NetworkCapabilityActivateUseCase(

        service_specifications=SqlServiceSpecificationRepository(session),

        service_capability_requirements=SqlServiceCapabilityRequirementRepository(session),

        service_orders=SqlServiceOrderRepository(session),

        service_instances=SqlServiceInstanceRepository(session),

        capability_instances=SqlCapabilityInstanceRepository(session),

        zones=SqlZoneRepository(session),

        zones=zones,

        publisher=publisher,

        resolver=DeviceTargetedControlPathResolver(

            CapabilityPlacementPlanner(zones),

            control_paths,

        ),

        control_paths=control_paths,

    )





def get_network_capability_activate_command_coordinator(

    session_maker: async_sessionmaker[AsyncSession],

    connection_manager: NatsConnectionManager,

    control_paths: ControlPathRegistry,

) -> NetworkCapabilityActivateCommandCoordinator:

    def use_case_factory(

        session: AsyncSession,

        publisher: DataBusPublisher,

    ) -> NetworkCapabilityActivateUseCase:

        return build_network_capability_activate_use_case(session, publisher, control_paths)



    return NetworkCapabilityActivateCommandCoordinator(

        session_maker=session_maker,

        use_case_factory=build_network_capability_activate_use_case,

        use_case_factory=use_case_factory,

        publisher=NatsPublisher(connection_manager),

    )

src/srm/api/databus/nats_subscriber.py

+3 −0
Changes for src/srm/api/databus/nats_subscriber.py: 3 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -13,6 +13,7 @@ from srm.api.databus.dependencies import ( 
    get_terminate_service_command_coordinator,

)

from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage

from srm.domain.ports.transformation_functions import ControlPathRegistry



logger: structlog.BoundLogger = structlog.get_logger(__name__)
@@ -46,6 +47,7 @@ COMMAND_SUBJECTS = ( 
async def subscribe_to_subjects(

    connection_manager: NatsConnectionManager,

    session_maker: async_sessionmaker[AsyncSession],

    control_paths: ControlPathRegistry,

) -> list["NatsSubscriber"]:

    deploy_coordinator = get_deploy_service_command_coordinator(

        session_maker,
@@ -58,6 +60,7 @@ async def subscribe_to_subjects( 
    network_activate_coordinator = get_network_capability_activate_command_coordinator(

        session_maker,

        connection_manager,

        control_paths,

    )



    command_handlers = {

src/srm/application/use_cases/network_capability_activate.py

+53 −13
Changes for src/srm/application/use_cases/network_capability_activate.py: 53 added lines, 13 removed lines.
Original line number Diff line number Diff line
@@ -2,15 +2,19 @@ from __future__ import annotations 


from dataclasses import dataclass, field

from datetime import UTC, datetime

from enum import StrEnum

from typing import Any, Literal

from uuid import UUID



from srm.application.services.capability_placement import (

    CapabilityPlacement,

    CapabilityPlacementPlanner,

    CapabilityPlacementRequest,

    PlacementPin,

)

from srm.application.services.device_targeted_control_path import (

    DeviceTargetedControlPathResolver,

    DeviceTargetedResolutionRequest,

    TargetResolutionFailure,

)

from srm.application.services.network_capability_activation_specs import (

    CapabilityActivationValidationError,

    normalize_network_capability_activation,
@@ -53,6 +57,7 @@ from srm.domain.ports.databus.events import ( 
    SrmOperationStatusV1,

)

from srm.domain.ports.databus.publisher import DataBusPublisher

from srm.domain.ports.transformation_functions.registry import ControlPathRegistry





@dataclass(frozen=True, slots=True)
@@ -89,6 +94,36 @@ class NetworkCapabilityActivateResult: 
    parameters_snapshot: Parameters | None = None





class NetworkCapabilityActivateFailure(StrEnum):

    SERVICE_NOT_APPLICABLE = "service_not_applicable"

    SUBSCRIBER_NOT_FOUND = "subscriber_not_found"

    UNSUPPORTED_IDENTIFIER = "unsupported_identifier"

    CAPABILITY_EXECUTION_FAILED = "capability_execution_failed"





_UNRESOLVED: dict[TargetResolutionFailure, tuple[NetworkCapabilityActivateFailure, str]] = {

    TargetResolutionFailure.NO_CANDIDATE: (

        NetworkCapabilityActivateFailure.SERVICE_NOT_APPLICABLE,

        "no control path can serve this target",

    ),

    TargetResolutionFailure.UNKNOWN_SUBSCRIBER: (

        NetworkCapabilityActivateFailure.SUBSCRIBER_NOT_FOUND,

        "no control path can serve this target",

    ),

    TargetResolutionFailure.UNSUPPORTED_IDENTIFIER: (

        NetworkCapabilityActivateFailure.UNSUPPORTED_IDENTIFIER,

        "no control path can serve this target",

    ),

    TargetResolutionFailure.CONTROL_PATH_UNAVAILABLE: (

        NetworkCapabilityActivateFailure.CAPABILITY_EXECUTION_FAILED,

        "a candidate control path has no adapter configured",

    ),

}





ACTIVATE_ACTION = "activate"





class NetworkCapabilityActivateUseCase:

    def __init__(

        self,
@@ -99,7 +134,8 @@ class NetworkCapabilityActivateUseCase: 
        capability_instances: CapabilityInstanceRepository,

        zones: ZoneRepository,

        publisher: DataBusPublisher,

        placement_planner: CapabilityPlacementPlanner | None = None,

        resolver: DeviceTargetedControlPathResolver,

        control_paths: ControlPathRegistry,

    ) -> None:

        self._service_specifications = service_specifications

        self._service_capability_requirements = service_capability_requirements
@@ -108,7 +144,8 @@ class NetworkCapabilityActivateUseCase: 
        self._capability_instances = capability_instances

        self._zones = zones

        self._publisher = publisher

        self._placement_planner = placement_planner or CapabilityPlacementPlanner(zones)

        self._resolver = resolver

        self._control_paths = control_paths



    async def accept(

        self,
@@ -177,24 +214,27 @@ class NetworkCapabilityActivateUseCase: 
            )

            return NetworkCapabilityActivateResult()



        placement = await self._placement_planner.place(

            CapabilityPlacementRequest(

        resolution = await self._resolver.resolve(

            DeviceTargetedResolutionRequest(

                capability_kind=capability_kind,

                domain_kind=requirement.domain_kind or DomainKind.NETWORK,

                action="activate",

                action=ACTIVATE_ACTION,

                parameters=parameters_snapshot,

                pin=PlacementPin(zone_id=command.zone_id, domain_id=command.domain_id),

                domain_kind=requirement.domain_kind or DomainKind.NETWORK,

            )

        )

        if placement is None:

        if resolution.placement is None:

            failure, detail = _UNRESOLVED[

                resolution.failure or TargetResolutionFailure.NO_CANDIDATE

            ]

            await self._publish_failed_before_start(

                command,

                detail=(

                    "Network capability placement could not be resolved for the requested zone "

                    "or domain."

                ),

                detail=f"{failure.value}: {detail}",

            )

            return NetworkCapabilityActivateResult()



        placement = resolution.placement



        service_order = await self._service_orders.create(

            ServiceOrder(

                operation_id=command.operation_id,

src/srm/main.py

+5 −3
Changes for src/srm/main.py: 5 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -83,9 +83,14 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: 
        raise



    try:

        control_path_registry = build_control_path_registry(

            settings.control_path_settings, factory=None

        )

        app.state.control_path_registry = control_path_registry

        databus_subscribers: list[NatsSubscriber] = await subscribe_to_subjects(

            connection_manager=databus_manager,

            session_maker=session_maker,

            control_paths=control_path_registry,

        )

    except Exception as e:

        logger.error("Databus subscription failed!", error=str(e))
@@ -93,9 +98,6 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: 
        await engine.dispose()

        raise



    app.state.control_path_registry = build_control_path_registry(

        settings.control_path_settings, factory=None

    )

    app.state.db_engine = engine

    app.state.session_maker = session_maker

    app.state.databus_connection_manager = databus_manager