feat: Traffic Influence support (ed32bd28) · Commits · OOP / code / service-resource-manager

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,