Merge current SRM wiring target into SDK consumer qualification (954c0329) · Commits · OOP / code / service-resource-manager

.env.example

+1 −0

Changes for .env.example: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		APP_NAME="Service Resource Manager"
		APP_DESCRIPTION="Service Resource Manager is the brains of the ETSI SDG Open Operator Platform - OpenOP"
		APP_VERSION="1.5.0"
		INSTANCES_MONITOR_INTERVAL=10.0

		POSTGRES_SETTINGS__URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/srm"
		POSTGRES_SETTINGS__ECHO = true

pyproject.toml

+1 −0

Changes for pyproject.toml: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -22,6 +22,7 @@ dependencies = [
		"asyncpg>=0.31.0",
		"nats-py>=2.10.0",
		"openop-tfsdk==2.0.0rc2",
		"apscheduler<4.0.0",
		]

		[project.optional-dependencies]

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