Merge branch 'refactor/tf-sdk-srm-bug-fixes' into 'refactor/tf-sdk-wiring' (078cfe03) · Commits · OOP / code / service-resource-manager

.env.example

+2 −0

Changes for .env.example: 2 added lines, 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
		@@ -13,3 +14,4 @@ NATS_SETTINGS__DRAIN_TIMEOUT = 30

		FEDERATION_MANAGER_SETTINGS__BASE_URL = "http://localhost:8082"
		FEDERATION_MANAGER_SETTINGS__TIMEOUT = 0.5
		FEDERATION_MANAGER_SETTINGS__ENABLED = true

pyproject.toml

+2 −2

Changes for pyproject.toml: 2 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -18,11 +18,11 @@ dependencies = [
		"pydantic-settings>=2.13.1",
		"python-dotenv>=1.2.2",
		"structlog>=25.5.0",
		"sunrise6g-opensdk==2.0.0",
		"sqlalchemy>=2.0.48",
		"asyncpg>=0.31.0",
		"nats-py>=2.10.0",
		"openop-tfsdk==1.1.1",
		"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/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/databus/nats_publisher.py

+45 −1

Changes for src/srm/adapters/databus/nats_publisher.py: 45 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -15,6 +15,37 @@ def _optional_str(value: Any) -> str \| 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:
		self._connection_manager = connection_manager
		@@ -28,9 +59,22 @@ class NatsPublisher(DataBusPublisher):

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

		# 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,
		)