feat: Traffic Influence support

Paris Stentoumisrequested to merge
refactor/traffic-influence-work into develop

Summary

Adds Traffic Influence network-capability support on the activate path, including resolving target.application_server from a deployed workload before TF-SDK activation, and aligns QoD / Traffic Influence instance monitors to emit CAMARA-style capability_status (and related info) in event metadata / result backend.

Changes

  • Platform catalog: bootstrap traffic-influence capability (CapabilityKind.TRAFFIC_INFLUENCE).
  • Activation:
    • TrafficInfluenceActivationSpec — requires device IPv4 + application_reference; validates traffic-influence fields.
    • ApplicationServerResolver — if application_server IP is missing, resolve it from an active workload instance (pinned app_instance_id, or oldest active instance in zone) and take the first endpoint IPv4.
    • Wire resolver into NetworkCapabilityActivate / databus dependencies; persist resolved parameters on the capability instance snapshot.
  • Catalog port: ServiceSpecificationRepository.get_by_ref (+ SQLAlchemy adapter) for app lookup by ref.
  • Instance monitor: for QoD and Traffic Influence, publish capability_status (AVAILABLE / UNAVAILABLE) and capability_status_info when useful (duration_expired, network_terminated); mirror status into result_summary.backend.
  • Tests: unit coverage for resolver, activate, activation specs, and monitor; databus integration coverage for traffic-influence activate with app-server resolution.

Merge request reports

src/srm/adapters/database/repos/catalog.py

+22 −0
Changes for src/srm/adapters/database/repos/catalog.py: 22 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -74,6 +74,28 @@ class SqlServiceSpecificationRepository(ServiceSpecificationRepository): 
        row = await self._session.scalar(stmt)

        return ServiceSpecificationMapper.to_domain(row) if row is not None else None



    async def get_by_ref(self, app_provider_id: str, ref: str) -> ServiceSpecification | None:

        stmt = (

            select(ServiceSpecificationRow)

            .options(

                selectinload(ServiceSpecificationRow.deployment_units),

                selectinload(ServiceSpecificationRow.capability_requirements),

            )

            .where(

                ServiceSpecificationRow.app_provider_id == app_provider_id,

                ServiceSpecificationRow.ref == ref,

                ServiceSpecificationRow.state == ServiceSpecificationState.ACTIVE,

            )

            .order_by(

                ServiceSpecificationRow.created_at.desc(),

                ServiceSpecificationRow.version.desc(),

                ServiceSpecificationRow.id.desc(),

            )

            .limit(1)

        )

        row = await self._session.scalar(stmt)

        return ServiceSpecificationMapper.to_domain(row) if row is not None else None



    async def list_by_app_provider_id(self, app_provider_id: str) -> list[ServiceSpecification]:

        stmt = (

            select(ServiceSpecificationRow)

src/srm/api/databus/dependencies.py

+12 −3
Changes for src/srm/api/databus/dependencies.py: 12 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -27,6 +27,7 @@ from srm.application.command_handlers.network_capability_deactivate import ( 
    NetworkCapabilityDeactivateCommandCoordinator,

)

from srm.application.command_handlers.terminate_service import TerminateServiceCommandCoordinator

from srm.application.services.application_server_resolver import ApplicationServerResolver

from srm.application.services.capability_placement import CapabilityPlacementPlanner

from srm.application.services.device_targeted_control_path import (

    DeviceTargetedControlPathResolver,
@@ -90,12 +91,15 @@ def build_network_capability_activate_use_case( 
    control_paths: ControlPathRegistry,

) -> NetworkCapabilityActivateUseCase:

    zones = SqlZoneRepository(session)

    service_specifications = SqlServiceSpecificationRepository(session)

    service_instances = SqlServiceInstanceRepository(session)

    capability_instances = SqlCapabilityInstanceRepository(session)

    return NetworkCapabilityActivateUseCase(

        service_specifications=SqlServiceSpecificationRepository(session),

        service_specifications=service_specifications,

        service_capability_requirements=SqlServiceCapabilityRequirementRepository(session),

        service_orders=SqlServiceOrderRepository(session),

        service_instances=SqlServiceInstanceRepository(session),

        capability_instances=SqlCapabilityInstanceRepository(session),

        service_instances=service_instances,

        capability_instances=capability_instances,

        zones=zones,

        publisher=publisher,

        resolver=DeviceTargetedControlPathResolver(
@@ -103,6 +107,11 @@ def build_network_capability_activate_use_case( 
            control_paths,

        ),

        control_paths=control_paths,

        application_server_resolver=ApplicationServerResolver(

            service_specifications,

            service_instances,

            capability_instances,

        ),

    )

src/srm/application/bootstrap/platform_catalog.py

+5 −0
Changes for src/srm/application/bootstrap/platform_catalog.py: 5 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -50,6 +50,11 @@ PLATFORM_SPECIFICATIONS = ( 
        name="Location Context",

        capability_kind=CapabilityKind.LOCATION_CONTEXT,

    ),

    PlatformSpecification(

        ref="traffic-influence",

        name="Traffic Influence",

        capability_kind=CapabilityKind.TRAFFIC_INFLUENCE,

    ),

)

src/srm/application/services/application_server_resolver.py

0 → 100644
+198 −0
Changes for src/srm/application/services/application_server_resolver.py: 198 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from __future__ import annotations



from dataclasses import dataclass

from datetime import datetime

from ipaddress import IPv4Address, IPv4Network, ip_address

from uuid import UUID



from srm.domain.models.canonical_parameters.parameters import (

    ApplicationServerTarget,

    Parameters,

)

from srm.domain.models.canonical_parameters.result import ServiceEndpoint

from srm.domain.models.runtime_inventory import ServiceInstance

from srm.domain.models.runtime_inventory.enums import ServiceInstanceState

from srm.domain.models.topology.enums import CapabilityKind

from srm.domain.ports.database.catalog import ServiceSpecificationRepository

from srm.domain.ports.database.runtime_inventory import (

    CapabilityInstanceRepository,

    ServiceInstanceRepository,

)





class ApplicationServerResolutionError(ValueError):

    pass





@dataclass(frozen=True, slots=True)

class ApplicationServerResolution:

    parameters: Parameters

    selected_app_instance_id: UUID | None





class ApplicationServerResolver:

    """Fill `target.application_server` from a deployed workload instance.



    Used by Traffic Influence activate: CAMARA supplies an app identity (and optional

    instance pin), and SRM picks the active instance's reported endpoint IP before the

    shared network activate path runs.

    """



    def __init__(

        self,

        service_specifications: ServiceSpecificationRepository,

        service_instances: ServiceInstanceRepository,

        capability_instances: CapabilityInstanceRepository,

    ) -> None:

        self._service_specifications = service_specifications

        self._service_instances = service_instances

        self._capability_instances = capability_instances



    async def resolve(

        self,

        *,

        app_provider_id: str,

        parameters: Parameters,

        zone_id: UUID | None,

    ) -> ApplicationServerResolution:

        if parameters.target.application_server is not None and (

            parameters.target.application_server.ipv4 is not None

            or parameters.target.application_server.ipv6 is not None

        ):

            selected = (

                parameters.target.application_reference.app_instance_id

                if parameters.target.application_reference is not None

                else None

            )

            return ApplicationServerResolution(

                parameters=parameters,

                selected_app_instance_id=selected,

            )



        application_reference = parameters.target.application_reference

        if application_reference is None:

            raise ApplicationServerResolutionError(

                "traffic_influence requires an application_reference"

            )



        app_spec = await self._service_specifications.get_by_ref(

            app_provider_id,

            application_reference.service_specification_ref,

        )

        if app_spec is None:

            raise ApplicationServerResolutionError("app not onboarded")



        instance = await self._select_instance(

            app_provider_id=app_provider_id,

            service_specification_id=app_spec.id,

            app_instance_id=application_reference.app_instance_id,

            zone_id=zone_id,

        )

        endpoint = await self._select_endpoint(service_instance_id=instance.id)

        ipv4 = _endpoint_ipv4(endpoint)

        if ipv4 is None:

            raise ApplicationServerResolutionError("instance has no reported endpoint address")



        enriched = parameters.model_copy(

            update={

                "target": parameters.target.model_copy(

                    update={

                        "application_server": ApplicationServerTarget(

                            ipv4=IPv4Network(f"{ipv4}/32"),

                        ),

                        "application_reference": application_reference.model_copy(

                            update={"app_instance_id": instance.id}

                        ),

                    }

                )

            }

        )

        return ApplicationServerResolution(

            parameters=enriched,

            selected_app_instance_id=instance.id,

        )



    async def _select_instance(

        self,

        *,

        app_provider_id: str,

        service_specification_id: UUID,

        app_instance_id: UUID | None,

        zone_id: UUID | None,

    ) -> ServiceInstance:

        if app_instance_id is not None:

            instance = await self._service_instances.get_by_id(app_instance_id)

            if (

                instance is None

                or instance.app_provider_id != app_provider_id

                or instance.service_specification_id != service_specification_id

                or instance.state != ServiceInstanceState.ACTIVE

            ):

                raise ApplicationServerResolutionError(

                    "pinned app instance not found or not active"

                )

            if zone_id is not None and instance.zone_id != zone_id:

                raise ApplicationServerResolutionError(

                    "pinned app instance is not in the requested zone"

                )

            return instance



        if zone_id is None:

            raise ApplicationServerResolutionError("zone required")



        candidates = await self._service_instances.list_by_app_provider_id(

            app_provider_id,

            service_specification_id=service_specification_id,

        )

        active_in_zone = [

            candidate

            for candidate in candidates

            if candidate.state == ServiceInstanceState.ACTIVE and candidate.zone_id == zone_id

        ]

        if not active_in_zone:

            raise ApplicationServerResolutionError("no active instance of app in zone")



        return min(active_in_zone, key=_instance_created_at)



    async def _select_endpoint(self, *, service_instance_id: UUID) -> ServiceEndpoint:

        rows = await self._capability_instances.get_by_service_instance_id(service_instance_id)

        workload_rows = [

            capability for capability, kind, _zone_id in rows if kind == CapabilityKind.WORKLOAD

        ]

        if not workload_rows:

            raise ApplicationServerResolutionError("instance has no reported endpoint address")



        endpoints = [

            endpoint

            for capability in workload_rows

            for endpoint in capability.result_summary.endpoints

        ]

        if not endpoints:

            raise ApplicationServerResolutionError("instance has no reported endpoint address")



        # TF-SDK uses the host IP only (appInstanceId := application_server IPv4).

        return endpoints[0]





def _instance_created_at(instance: ServiceInstance) -> datetime:

    if instance.created_at is not None:

        return instance.created_at

    return datetime.min





def _endpoint_ipv4(endpoint: ServiceEndpoint) -> IPv4Address | None:

    if endpoint.ipv4 is not None:

        try:

            address = ip_address(endpoint.ipv4)

        except ValueError:

            return None

        if isinstance(address, IPv4Address):

            return address

        return None

    try:

        address = ip_address(endpoint.fqdn)

    except ValueError:

        return None

    if isinstance(address, IPv4Address):

        return address

    return None

src/srm/application/services/network_capability_activation_specs.py

+16 −9
Changes for src/srm/application/services/network_capability_activation_specs.py: 16 added lines, 9 removed lines.
Original line number Diff line number Diff line
@@ -66,9 +66,8 @@ class QodSessionFields(DomainModel): 




class TrafficInfluenceFields(DomainModel):

    profile_ref: str | None = None

    duration_seconds: int | None = Field(default=None, gt=0)

    traffic_filters: TrafficFilters | None = None

    device: DeviceTarget | None = None

    application_reference: ApplicationReferenceTarget | None = None

    application_server: ApplicationServerTarget | None = None
@@ -118,9 +117,8 @@ class TrafficInfluenceActivationSpec: 
    ) -> NormalizedCapabilityActivation:

        try:

            fields = TrafficInfluenceFields(

                profile_ref=parameters.profile_ref,

                duration_seconds=parameters.duration_seconds,

                traffic_filters=parameters.traffic_filters,

                device=target.device,

                application_reference=target.application_reference,

                application_server=target.application_server,

            )
@@ -129,13 +127,20 @@ class TrafficInfluenceActivationSpec: 
                _validation_error_detail("traffic_influence", exc)

            ) from exc



        if fields.device is None:

            raise CapabilityActivationValidationError("traffic_influence requires a device target")

        if fields.device.ipv4 is None and fields.device.ipv4_private is None:

            raise CapabilityActivationValidationError(

                "traffic_influence requires a device IPv4 address"

            )

        if fields.application_reference is None:

            raise CapabilityActivationValidationError(

                "traffic_influence requires an application_reference"

            )



        return NormalizedCapabilityActivation(

            target=target,

            parameters=CapabilityParameters(

                profile_ref=fields.profile_ref,

                duration_seconds=fields.duration_seconds,

                traffic_filters=fields.traffic_filters,

            ),

            parameters=CapabilityParameters(traffic_filters=fields.traffic_filters),

        )
@@ -144,6 +149,8 @@ CAPABILITY_ACTIVATION_SPECS: Mapping[CapabilityKind, CapabilityActivationSpec] = 
    CapabilityKind.TRAFFIC_INFLUENCE: TrafficInfluenceActivationSpec(),

}



APPLICATION_SERVER_RESOLVED_KINDS = frozenset({CapabilityKind.TRAFFIC_INFLUENCE})





def normalize_network_capability_activation(

    payload: NetworkCapabilityPayloadLike,