SRM Refactor: Wire OpenOp TF-SDK as the control-path adapter factory and use it for service deploy and network capability activation. (!33) · Merge requests · OOP / code / service-resource-manager

src/srm/adapters/background_tasks/instances_monitor.py

0 → 100644

+67 −0

Changes for src/srm/adapters/background_tasks/instances_monitor.py: 67 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""Implementation of Service and Capability Instances monitoring adapter
		using APScheduler package.
		"""

		import structlog
		from apscheduler.schedulers.asyncio import AsyncIOScheduler
		from apscheduler.triggers.interval import IntervalTrigger
		from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

		from srm.adapters.database.repos.runtime_inventory import (
		SqlCapabilityInstanceRepository,
		SqlServiceInstanceRepository,
		SqlServiceOrderRepository,
		)
		from srm.application.use_cases.instance_monitor import CapabilityInstanceMonitorUseCase
		from srm.domain.ports.background_tasks.instances_monitor import InstancesMonitorPort
		from srm.domain.ports.databus.publisher import DataBusPublisher
		from srm.domain.ports.transformation_functions import ControlPathRegistry

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


		class APSchedulerInstancesMonitor(InstancesMonitorPort):
		def __init__(
		self,
		session_maker: async_sessionmaker[AsyncSession],
		control_path_registry: ControlPathRegistry,
		publisher: DataBusPublisher,
		) -> None:
		self._session_maker = session_maker
		self._control_path_registry = control_path_registry
		self._publisher = publisher
		self._scheduler = AsyncIOScheduler()

		def register_scheduler(
		self,
		interval: float,
		) -> None:
		self._scheduler.add_job(
		self.update_statuses, IntervalTrigger(seconds=interval), id="instance_monitor"
		)

		def start(self) -> None:
		self._scheduler.start()

		def shutdown(self) -> None:
		self._scheduler.shutdown(wait=True)

		async def update_statuses(self) -> None:
		async with self._session_maker() as session:
		try:
		use_case = CapabilityInstanceMonitorUseCase(
		control_path_registry=self._control_path_registry,
		capability_instances=SqlCapabilityInstanceRepository(session=session),
		service_instances=SqlServiceInstanceRepository(session=session),
		service_orders=SqlServiceOrderRepository(session=session),
		publisher=self._publisher,
		)
		await use_case.execute()
		await session.commit()
		except Exception as exc:
		logger.exception(
		"instances_monitor_tick_failed",
		error=str(exc),
		error_type=type(exc).__name__,
		)
		await session.rollback()

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

+12 −0

Changes for src/srm/adapters/database/repos/catalog.py: 12 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -17,6 +17,7 @@ from srm.adapters.database.sql import (
		ServiceCapabilityRequirementRow,
		ServiceDeploymentUnitRow,
		ServiceInstanceRow,
		ServiceOrderRow,
		ServiceSpecificationRow,
		)
		from srm.adapters.errors import DuplicateServiceSpecificationError, ServiceSpecificationInUseError
		@@ -119,6 +120,17 @@ class SqlServiceSpecificationRepository(ServiceSpecificationRepository):
		.limit(1)
		)
		if any_instance_id is None:
		# An instance a partner hosts leaves no instance row here, only its order, and that
		# order still points at the specification: keep it as retired history instead.
		any_order_id = await self._session.scalar(
		select(ServiceOrderRow.id)
		.where(ServiceOrderRow.service_specification_id == id)
		.limit(1)
		)
		if any_order_id is not None:
		row.state = ServiceSpecificationState.RETIRED
		await self._session.flush()
		return True
		await self._session.execute(
		delete(ServiceCapabilityRequirementRow).where(
		ServiceCapabilityRequirementRow.service_specification_id == id

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

+106 −0

Changes for src/srm/adapters/database/repos/runtime_inventory.py: 106 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -17,9 +17,14 @@ from srm.adapters.database.mappers import (
		from srm.adapters.database.sql import (
		CapabilityInstanceRow,
		CapabilityRow,
		ControlPathBindingRow,
		DomainRow,
		ServiceInstanceRow,
		ServiceOrderRow,
		ZoneRow,
		)
		from srm.adapters.errors import DuplicateServiceInstanceError
		from srm.domain.models.canonical_parameters.result import Result
		from srm.domain.models.runtime_inventory import (
		CapabilityInstance,
		ServiceInstance,
		@@ -33,6 +38,7 @@ from srm.domain.models.runtime_inventory.enums import (
		from srm.domain.models.topology.enums import CapabilityKind
		from srm.domain.ports.database.runtime_inventory import (
		CapabilityInstanceRepository,
		MonitorableCapabilityInstance,
		ServiceInstanceRepository,
		ServiceOrderRepository,
		)
		@@ -177,6 +183,8 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository):
		await self._session.flush()
		except IntegrityError as exc:
		_log_integrity_error(exc, entity="service_instance", operation="create")
		if _pgcode(exc) == "23505":
		raise DuplicateServiceInstanceError() from exc
		raise
		service_instance_id = row.id
		saved = await self.get_by_id(service_instance_id)
		@@ -203,6 +211,34 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository):
		raise RuntimeError("Updated service instance could not be reloaded")
		return saved

		async def reconcile_states(self) -> int:
		stmt = (
		select(ServiceInstanceRow)
		.options(selectinload(ServiceInstanceRow.capability_instances))
		.where(ServiceInstanceRow.state != ServiceInstanceState.TERMINATED)
		)

		rows = (await self._session.scalars(stmt)).all()
		updated = 0
		for row in rows:
		previous_state = row.state
		service_instance = ServiceInstanceMapper.to_domain(row)
		service_instance.update_state()
		if service_instance.state == previous_state:
		continue
		row.state = service_instance.state
		row.terminated_at = service_instance.terminated_at
		updated += 1
		logger.info(
		"service_instance_state_changed",
		service_instance_id=str(row.id),
		from_state=previous_state,
		to_state=service_instance.state,
		)

		await self._session.flush()
		return updated


		class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):
		def __init__(self, session: AsyncSession) -> None:
		@@ -233,6 +269,7 @@ class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):
		state: CapabilityInstanceState,
		*,
		terminated_at: datetime \| None = None,
		result_summary: Result \| None = None,
		) -> CapabilityInstance:
		row = await self._session.get(CapabilityInstanceRow, id)
		if row is None:
		@@ -240,6 +277,8 @@ class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):
		row.state = state
		if terminated_at is not None:
		row.terminated_at = terminated_at
		if result_summary is not None:
		row.result_summary = result_summary.model_dump(mode="json")
		await self._session.flush()
		saved = await self.get_by_id(id)
		if saved is None:
		@@ -306,3 +345,70 @@ class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):
		stmt = stmt.where(CapabilityInstanceRow.external_ref == external_ref)
		rows = (await self._session.execute(stmt)).all()
		return [(CapabilityInstanceMapper.to_domain(row[0]), row[1], row[2]) for row in rows]

		async def list_monitorable(self) -> list[MonitorableCapabilityInstance]:
		stmt = (
		select(
		CapabilityInstanceRow,
		CapabilityRow.kind,
		CapabilityRow.ref,
		DomainRow.ref,
		DomainRow.kind,
		ZoneRow.ref,
		ZoneRow.kind,
		ControlPathBindingRow.ref,
		)
		.join(CapabilityRow, CapabilityInstanceRow.capability_id == CapabilityRow.id)
		.join(DomainRow, CapabilityRow.domain_id == DomainRow.id)
		.join(ZoneRow, DomainRow.zone_id == ZoneRow.id)
		.outerjoin(
		ControlPathBindingRow,
		CapabilityInstanceRow.control_path_binding_id == ControlPathBindingRow.id,
		)
		.where(CapabilityInstanceRow.state != CapabilityInstanceState.TERMINATED)
		.where(CapabilityInstanceRow.external_id.is_not(None))
		.where(CapabilityInstanceRow.control_path_ref_snapshot.is_not(None))
		)
		rows = (await self._session.execute(stmt)).all()
		monitorable: list[MonitorableCapabilityInstance] = []
		for (
		instance_row,
		capability_kind,
		capability_ref,
		domain_ref,
		domain_kind,
		zone_ref,
		zone_kind,
		binding_ref,
		) in rows:
		if instance_row.external_id is None or instance_row.control_path_ref_snapshot is None:
		continue
		monitorable.append(
		MonitorableCapabilityInstance(
		capability_instance_id=instance_row.id,
		service_instance_id=instance_row.service_instance_id,
		originating_service_order_id=instance_row.originating_service_order_id,
		external_id=instance_row.external_id,
		external_ref=instance_row.external_ref,
		control_path_ref=instance_row.control_path_ref_snapshot,
		control_path_binding_ref=binding_ref,
		capability_kind=capability_kind,
		capability_ref=capability_ref,
		domain_ref=domain_ref,
		domain_kind=domain_kind,
		zone_ref=zone_ref,
		zone_kind=zone_kind,
		current_state=instance_row.state,
		created_at=instance_row.created_at,
		duration_seconds=_duration_seconds(instance_row.parameters_snapshot),
		)
		)
		return monitorable


		def _duration_seconds(parameters_snapshot: dict[str, object]) -> int \| None:
		parameters = parameters_snapshot.get("parameters")
		if not isinstance(parameters, dict):
		return None
		value = parameters.get("duration_seconds")
		return value if isinstance(value, int) else None

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

+46 −22

Changes for src/srm/adapters/database/repos/topology.py: 46 added lines, 22 removed lines.
Original line number	Diff line number	Diff line
		from __future__ import annotations

		from typing import Any
		from uuid import UUID

		import structlog
		@@ -41,20 +42,33 @@ def _log_integrity_error(exc: IntegrityError, *, entity: str, operation: str) ->
		)


		def _zone_options() -> Any:
		return (
		selectinload(ZoneRow.domains)
		.selectinload(DomainRow.capabilities)
		.selectinload(CapabilityRow.control_path_bindings)
		)


		def _domain_options() -> Any:
		return selectinload(DomainRow.capabilities).selectinload(CapabilityRow.control_path_bindings)


		def _capability_options() -> Any:
		return selectinload(CapabilityRow.control_path_bindings)


		class SqlZoneRepository(ZoneRepository):
		def __init__(self, session: AsyncSession) -> None:
		self._session = session

		async def get_by_id(self, id: UUID) -> Zone \| None:
		stmt = (
		select(ZoneRow)
		.options(
		selectinload(ZoneRow.domains)
		.selectinload(DomainRow.capabilities)
		.selectinload(CapabilityRow.control_path_bindings)
		)
		.where(ZoneRow.id == id)
		)
		stmt = select(ZoneRow).options(_zone_options()).where(ZoneRow.id == id)
		row = await self._session.scalar(stmt)
		return ZoneMapper.to_domain(row) if row is not None else None

		async def get_by_ref(self, ref: str) -> Zone \| None:
		stmt = select(ZoneRow).options(_zone_options()).where(ZoneRow.ref == ref)
		row = await self._session.scalar(stmt)
		return ZoneMapper.to_domain(row) if row is not None else None

		@@ -69,11 +83,7 @@ class SqlZoneRepository(ZoneRepository):
		) -> list[Zone]:
		stmt = (
		select(ZoneRow)
		.options(
		selectinload(ZoneRow.domains)
		.selectinload(DomainRow.capabilities)
		.selectinload(CapabilityRow.control_path_bindings)
		)
		.options(_zone_options())
		.where(ZoneRow.kind == ZoneKind.RESOURCE)
		.order_by(ZoneRow.ref, ZoneRow.id)
		)
		@@ -106,14 +116,15 @@ class SqlDomainRepository(DomainRepository):
		self._session = session

		async def get_by_id(self, id: UUID) -> Domain \| None:
		stmt = select(DomainRow).options(_domain_options()).where(DomainRow.id == id)
		row = await self._session.scalar(stmt)
		return DomainMapper.to_domain(row) if row is not None else None

		async def get_by_ref(self, zone_id: UUID, ref: str) -> Domain \| None:
		stmt = (
		select(DomainRow)
		.options(
		selectinload(DomainRow.capabilities).selectinload(
		CapabilityRow.control_path_bindings
		)
		)
		.where(DomainRow.id == id)
		.options(_domain_options())
		.where(DomainRow.zone_id == zone_id, DomainRow.ref == ref)
		)
		row = await self._session.scalar(stmt)
		return DomainMapper.to_domain(row) if row is not None else None
		@@ -138,10 +149,15 @@ class SqlCapabilityRepository(CapabilityRepository):
		self._session = session

		async def get_by_id(self, id: UUID) -> Capability \| None:
		stmt = select(CapabilityRow).options(_capability_options()).where(CapabilityRow.id == id)
		row = await self._session.scalar(stmt)
		return CapabilityMapper.to_domain(row) if row is not None else None

		async def get_by_ref(self, domain_id: UUID, ref: str) -> Capability \| None:
		stmt = (
		select(CapabilityRow)
		.options(selectinload(CapabilityRow.control_path_bindings))
		.where(CapabilityRow.id == id)
		.options(_capability_options())
		.where(CapabilityRow.domain_id == domain_id, CapabilityRow.ref == ref)
		)
		row = await self._session.scalar(stmt)
		return CapabilityMapper.to_domain(row) if row is not None else None
		@@ -170,6 +186,14 @@ class SqlControlPathBindingRepository(ControlPathBindingRepository):
		row = await self._session.scalar(stmt)
		return ControlPathBindingMapper.to_domain(row) if row is not None else None

		async def get_by_ref(self, capability_id: UUID, ref: str) -> ControlPathBinding \| None:
		stmt = select(ControlPathBindingRow).where(
		ControlPathBindingRow.capability_id == capability_id,
		ControlPathBindingRow.ref == ref,
		)
		row = await self._session.scalar(stmt)
		return ControlPathBindingMapper.to_domain(row) if row is not None else None

		async def create(self, control_path_binding: ControlPathBinding) -> ControlPathBinding:
		row = ControlPathBindingMapper.to_row(control_path_binding)
		self._session.add(row)

src/srm/adapters/databus/nats_publisher.py

+60 −0

Changes for src/srm/adapters/databus/nats_publisher.py: 60 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import json
		from typing import Any

		import structlog

		from srm.adapters.databus.nats_connection_manager import NatsConnectionManager
		from srm.domain.ports.databus.publisher import DataBusPublisher

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


		def _optional_str(value: Any) -> str \| None:
		if value is None:
		return None
		return str(value)


		def _reason_from_error(error: Any) -> str \| None:
		if not isinstance(error, dict):
		return None
		for key in ("detail", "title"):
		reason = _optional_str(error.get(key))
		if reason:
		return reason
		return None


		def _failure_reason(payload: dict[str, Any]) -> str \| None:
		"""Prefer completed-event error.detail/title; else instances[].error; else metadata.reason."""
		reason = _reason_from_error(payload.get("error"))
		if reason:
		return reason

		instances = payload.get("instances")
		if isinstance(instances, list):
		for instance in instances:
		if not isinstance(instance, dict):
		continue
		reason = _reason_from_error(instance.get("error"))
		if reason:
		return reason

		metadata = payload.get("metadata")
		if isinstance(metadata, dict):
		return _optional_str(metadata.get("reason"))
		return None


		class NatsPublisher(DataBusPublisher):
		def __init__(self, connection_manager: NatsConnectionManager) -> None:
		@@ -18,3 +59,22 @@ class NatsPublisher(DataBusPublisher):

		body = json.dumps(payload).encode("utf-8")
		await self._connection_manager.client.publish(subject, body, headers=headers)

		# operation.status events carry `state`; operation.completed (and FM) carry `status`.
		status_state = _optional_str(payload.get("state"))
		completed_status = _optional_str(payload.get("status"))
		reason = _failure_reason(payload)
		is_failure = completed_status in {"failed", "partially_completed"} or status_state in {
		"failed",
		"failed_before_start",
		}
		log = logger.warning if is_failure else logger.info
		log(
		"databus_message_published",
		subject=subject,
		operation_id=_optional_str(payload.get("operation_id")),
		correlation_id=_optional_str(payload.get("correlation_id")),
		status_state=status_state,
		completed_status=completed_status,
		reason=reason,
		)

SRM Refactor: Wire OpenOp TF-SDK as the control-path adapter factory and use it for service deploy and network capability activation.

Summary

Merge request reports