SRM Refactor: Network Capability Activation

Paris Stentoumisrequested to merge
refactor/network-capability-deployment into refactor/tf-sdk-ports

Summary

Adds handling for command.srm.network.capability.activate and wires it into the NATS DataBus subscription flow.

The new activation flow validates the command payload, resolves the requested network capability against active service specification requirements, performs naive placement using zone/domain constraints, creates the accepted service order and service instance records, and publishes SRM operation status/completed events according to the current architecture.

Changes

  • Added SrmNetworkCapabilityActivateV1 command schema with service instance, service specification, placement pin, and network capability payload validation.
  • Added NetworkCapabilityActivateCommandCoordinator to parse commands, commit acceptance separately, publish accepted status, complete the operation, and handle before/after acceptance failures.
  • Added NetworkCapabilityActivateUseCase for:
    • duplicate operation replay
    • supported capability validation
    • active service specification and requirement resolution
    • zone/domain placement
    • service order and service instance creation
    • capability instance creation
    • operation completed/failed event generation
  • Subscribed SRM to command.srm.network.capability.activate.
  • Added temporary default QoD service specification bootstrap until catalog migrations/Alembic are introduced.
  • Added bootstrap settings for the default QoD service specification and platform app provider.
  • Updated canonical parameter/domain exports to support the new flow.
  • Added unit, application, API, and integration coverage for the network capability activation path and DataBus command flow.

Notes

  • Capability execution is currently a placeholder success path. The use case includes a TODO to replace it with a TF-SDK capability activation adapter call.
  • The default QoD service specification bootstrap is temporary and should be removed once catalog data is managed through migrations.

Merge request reports

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/adapters/transformation_functions/registry.py

0 → 100644
+46 −0
Changes for src/srm/adapters/transformation_functions/registry.py: 46 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from collections.abc import Mapping



from srm.config import ControlPathSettings

from srm.domain.ports.transformation_functions import (

    ControlPathRegistry,

    TransformationFunctionAdapterFactory,

    TransformationFunctionExecutionPort,

    UnknownControlPathRef,

)





class NoAdapterFactoryInstalled(RuntimeError):

    pass





class ConfiguredControlPathRegistry(ControlPathRegistry):

    def __init__(self, adapters: Mapping[str, TransformationFunctionExecutionPort]) -> None:

        self._adapters = dict(adapters)



    def get_adapter(self, control_path_ref: str) -> TransformationFunctionExecutionPort:

        try:

            return self._adapters[control_path_ref]

        except KeyError as exc:

            raise UnknownControlPathRef(control_path_ref) from exc



    def control_path_refs(self) -> list[str]:

        return sorted(self._adapters)





def build_control_path_registry(

    settings: ControlPathSettings,

    factory: TransformationFunctionAdapterFactory | None,

) -> ConfiguredControlPathRegistry:

    if not settings.entries:

        return ConfiguredControlPathRegistry({})

    if factory is None:

        raise NoAdapterFactoryInstalled(

            "control paths are configured but no TF-SDK adapter factory is installed: "

            + ", ".join(entry.control_path_ref for entry in settings.entries)

        )

    return ConfiguredControlPathRegistry(

        {

            entry.control_path_ref: factory.create_adapter(entry.adapter, entry.config)

            for entry in settings.entries

        }

    )

src/srm/api/databus/dependencies.py

+51 −0
Changes for src/srm/api/databus/dependencies.py: 51 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -14,10 +14,21 @@ 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.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(
@@ -54,6 +65,46 @@ 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=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=use_case_factory,

        publisher=NatsPublisher(connection_manager),

    )





def get_terminate_service_use_case(

    session: AsyncSession,

    connection_manager: NatsConnectionManager,

src/srm/api/databus/nats_subscriber.py

+10 −1
Changes for src/srm/api/databus/nats_subscriber.py: 10 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -9,9 +9,11 @@ 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_network_capability_activate_command_coordinator,

    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__)
@@ -30,13 +32,13 @@ 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,

}



COMMAND_SUBJECTS = (

    "command.srm.service.deploy",

    "command.srm.network.capability.activate",

    "command.srm.service.terminate",

    *COMMAND_HANDLERS.keys(),

)
@@ -45,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,
@@ -54,9 +57,15 @@ async def subscribe_to_subjects( 
        session_maker,

        connection_manager,

    )

    network_activate_coordinator = get_network_capability_activate_command_coordinator(

        session_maker,

        connection_manager,

        control_paths,

    )



    command_handlers = {

        "command.srm.service.deploy": deploy_coordinator.handle,

        "command.srm.network.capability.activate": network_activate_coordinator.handle,

        "command.srm.service.terminate": terminate_coordinator.handle,

        **COMMAND_HANDLERS,

    }

src/srm/api/databus/schemas.py

+4 −0
Changes for src/srm/api/databus/schemas.py: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -7,6 +7,7 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator 
from srm.domain.models.canonical_parameters.parameters import (

    CapabilityParameters,

    CapabilityTarget,

    SourceSpecification,

)
@@ -96,9 +97,11 @@ class NetworkCapabilityPayloadV1(CommandSchema): 
    target: CapabilityTarget

    profile_ref: str | None = None

    parameters: CapabilityParameters

    source_spec: SourceSpecification | None = None





class SrmNetworkCapabilityActivateV1(CommandPayloadV1):

    service_instance_id: UUID

    service_specification_id: UUID

    zone_id: UUID | None = None

    domain_id: UUID | None = None
@@ -135,6 +138,7 @@ class NetworkCapabilityUpdatePayloadV1(NetworkCapabilityRealizationRefV1): 
    target: CapabilityTarget | None = None

    profile_ref: str | None = None

    parameters: CapabilityParameters

    source_spec: SourceSpecification | None = None





class SrmNetworkCapabilityUpdateV1(CommandPayloadV1):