feat: capability instances monitor + fix: changed 422 -> 400 for validation error in test_catalog (9909e9f8) · Commits · OOP / code / service-resource-manager

.env.example

+1 −0

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

Original line number	Diff line number	Diff line
		@@ -22,6 +22,7 @@ dependencies = [
		"sqlalchemy>=2.0.48",
		"asyncpg>=0.31.0",
		"nats-py>=2.10.0",
		"apscheduler<4.0.0",
		]

		[project.optional-dependencies]

src/srm/adapters/background_tasks/instances_monitor.py

0 → 100644

+66 −0

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,
		)
		from srm.adapters.transformation_functions.registry import ConfiguredControlPathRegistry
		from srm.application.use_cases.instance_monitor import CapabilityInstanceMonitorUseCase
		from srm.domain.ports.background_tasks.instances_monitor import InstancesMonitorPort

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


		class APSchedulerInstancesMonitor(InstancesMonitorPort):
		def __init__(
		self,
		session_maker: async_sessionmaker[AsyncSession],
		cp_registry: ConfiguredControlPathRegistry,
		) -> None:
		self._session_maker = session_maker
		self._control_path_registry = cp_registry
		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()

		async def update_statuses(self) -> None:
		logger.info("update_statuses")
		async with self._session_maker() as session:
		try:
		capability_instances_repo: SqlCapabilityInstanceRepository = (
		SqlCapabilityInstanceRepository(session=session)
		)
		service_instances_repo: SqlServiceInstanceRepository = SqlServiceInstanceRepository(
		session=session
		)
		use_case = CapabilityInstanceMonitorUseCase(
		cp_registry=self._control_path_registry,
		capability_instances_repo=capability_instances_repo,
		service_instances_repo=service_instances_repo,
		)

		await use_case._update_capability_instances_state()
		await use_case._update_service_instances_state()
		await session.commit()
		except Exception as e:
		logger.error("update_statuses_failed", exc=e)
		await session.rollback()

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

+91 −2

Original line number	Diff line number	Diff line
		@@ -4,7 +4,7 @@ from datetime import datetime
		from uuid import UUID

		import structlog
		from sqlalchemy import select
		from sqlalchemy import and_, or_, select, update
		from sqlalchemy.exc import IntegrityError
		from sqlalchemy.ext.asyncio import AsyncSession
		from sqlalchemy.orm import selectinload
		@@ -20,7 +20,11 @@ from srm.domain.models.runtime_inventory import (
		ServiceInstance,
		ServiceOrder,
		)
		from srm.domain.models.runtime_inventory.enums import ServiceInstanceState, ServiceOrderState
		from srm.domain.models.runtime_inventory.enums import (
		CapabilityInstanceState,
		ServiceInstanceState,
		ServiceOrderState,
		)
		from srm.domain.ports.database.runtime_inventory import (
		CapabilityInstanceRepository,
		ServiceInstanceRepository,
		@@ -151,6 +155,31 @@ class SqlServiceInstanceRepository(ServiceInstanceRepository):
		raise RuntimeError("Updated service instance could not be reloaded")
		return saved

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

		logger.info("service_instance_reconcile_states", stmt=str(stmt))
		rows = (await self._session.scalars(stmt)).all()
		for row in rows:
		service_instance = ServiceInstanceMapper.to_domain(row)
		logger.info(
		"before_update_state",
		instance=service_instance.terminated_at,
		row=row.terminated_at,
		)
		service_instance.update_state()
		row.state = service_instance.state
		row.terminated_at = service_instance.terminated_at
		logger.info(
		"after_update_state", instance=service_instance.terminated_at, row=row.terminated_at
		)

		await self._session.flush()


		class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):
		def __init__(self, session: AsyncSession) -> None:
		@@ -174,3 +203,63 @@ class SqlCapabilityInstanceRepository(CapabilityInstanceRepository):
		if saved is None:
		raise RuntimeError("Created capability instance could not be reloaded")
		return saved

		async def reconcile_states(
		self,
		states: list[tuple[CapabilityInstanceState, str \| None, str \| None]],
		) -> None:
		logger.info("reconcile_states_start")

		if len(states) > 0:
		state_update_conditions = [
		and_(
		CapabilityInstanceRow.state != capability_state[0],
		CapabilityInstanceRow.external_id == capability_state[1],
		CapabilityInstanceRow.external_ref == capability_state[2],
		)
		for capability_state in states
		]

		update_query = select(CapabilityInstanceRow).where(or_(*state_update_conditions))

		result = await self._session.scalars(update_query)
		rows = result.all()
		logger.info("update_non_running", query=str(update_query))
		logger.info("reconcile_states", runnin_instances=len(rows))
		rows_by_key = {(row.external_id, row.external_ref): row for row in rows}

		for capability_state in states:
		key = (capability_state[1], capability_state[2])

		row = rows_by_key.get(key)

		if row is None:
		continue

		row.state = capability_state[0]

		non_running_conditions = [
		and_(
		CapabilityInstanceRow.external_id == capability_state[1],
		CapabilityInstanceRow.external_ref == capability_state[2],
		)
		for capability_state in states
		]

		update_non_running = (
		update(CapabilityInstanceRow)
		.where(
		CapabilityInstanceRow.state != CapabilityInstanceState.TERMINATED,
		~or_(*non_running_conditions) if len(states) > 0 else and_(True),
		)
		.values(state=CapabilityInstanceState.TERMINATED)
		.values(terminated_at=datetime.now())
		)

		logger.info("update_non_running", query=str(update_non_running))

		await self._session.execute(update_non_running)

		logger.info("execute_non_running")

		await self._session.flush()

src/srm/adapters/transformation_functions/registry.py

+8 −0

Original line number	Diff line number	Diff line
		from collections.abc import Mapping

		import structlog

		from srm.config import ControlPathSettings
		from srm.domain.ports.transformation_functions import (
		ControlPathRegistry,
		@@ -8,6 +10,8 @@ from srm.domain.ports.transformation_functions import (
		UnknownControlPathRef,
		)

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


		class NoAdapterFactoryInstalled(RuntimeError):
		pass
		@@ -26,11 +30,15 @@ class ConfiguredControlPathRegistry(ControlPathRegistry):
		def control_path_refs(self) -> list[str]:
		return sorted(self._adapters)

		def adapters(self) -> list[TransformationFunctionExecutionPort]:
		return list(self._adapters.values())


		def build_control_path_registry(
		settings: ControlPathSettings,
		factory: TransformationFunctionAdapterFactory \| None,
		) -> ConfiguredControlPathRegistry:
		logger.info("building_cpr", entries=settings.entries)
		if not settings.entries:
		return ConfiguredControlPathRegistry({})
		if factory is None: