SRM Bug fixes found after the wiring of TF-SDK (!45) · 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/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,
		)

src/srm/adapters/federation_manager/fm_client.py

+53 −8

Changes for src/srm/adapters/federation_manager/fm_client.py: 53 added lines, 8 removed lines.
Original line number	Diff line number	Diff line
		@@ -7,7 +7,11 @@ import httpx
		import structlog
		from pydantic import BaseModel

		from srm.adapters.errors import DownstreamServiceException, NotFoundException
		from srm.adapters.errors import (
		DownstreamServiceException,
		FederationUnavailableError,
		NotFoundException,
		)
		from srm.config import FederationManagerSettings
		from srm.domain.models.runtime_inventory.models import ServiceInstance
		from srm.domain.models.topology.enums import ZoneKind, ZoneState
		@@ -85,6 +89,16 @@ class FederationManager(FederationManagerPort):
		log.warning("FM resource not found", url=url)
		raise NotFoundException()

		if response.status_code >= 500:
		log.error("FM returned error", status=response.status_code, body=response.text)
		raise FederationUnavailableError(
		message="FM request failed",
		details={
		"status_code": response.status_code,
		"response": response.text,
		},
		)

		if response.status_code >= 400:
		log.error("FM returned error", status=response.status_code, body=response.text)
		raise DownstreamServiceException(
		@@ -111,38 +125,61 @@ class FederationManager(FederationManagerPort):

		except httpx.TimeoutException as exc:
		log.exception("FM request timed out", error=str(exc))
		raise DownstreamServiceException(
		raise FederationUnavailableError(
		message="FM request timed out",
		details=str(exc),
		)
		) from exc
		except httpx.ConnectError as exc:
		log.exception("FM connection failed", error=str(exc))
		raise DownstreamServiceException(
		raise FederationUnavailableError(
		message="Could not connect to FM",
		details=str(exc),
		)
		) from exc
		except httpx.RequestError as exc:
		log.exception("FM request error", error=str(exc))
		raise DownstreamServiceException(
		raise FederationUnavailableError(
		message="FM request failed",
		details=str(exc),
		)
		) from exc

		def _disabled(self, op: str) -> bool:
		if self._settings.enabled:
		return False
		logger.warning("fm_unavailable", op=op, reason="disabled")
		return True

		def _log_unavailable(self, op: str, exc: FederationUnavailableError) -> None:
		logger.warning("fm_unavailable", op=op, error=str(exc), details=exc._details)

		async def has_zone(self, zone_id: UUID) -> bool:
		return await self.get_zone(zone_id=zone_id) is not None

		async def get_zone(self, zone_id: UUID) -> Zone \| None:
		if self._disabled("get_zone"):
		return None
		try:
		res = await self._request("GET", str.format(_GET_ZONE_ENDPOINT, zone_id=zone_id))
		return FederatedZone.model_validate(res).to_zone()
		except NotFoundException:
		return None
		except FederationUnavailableError as exc:
		self._log_unavailable("get_zone", exc)
		return None

		async def list_zones(self) -> list[Zone]:
		if self._disabled("list_zones"):
		return []
		try:
		res = await self._request("GET", _LIST_ZONES_ENDPOINT)
		return [FederatedZone.model_validate(obj).to_zone() for obj in res]
		except FederationUnavailableError as exc:
		self._log_unavailable("list_zones", exc)
		return []

		async def list_full_service_instances(self, app_provider_id: str) -> list[ServiceInstance]:
		if self._disabled("list_full_service_instances"):
		return []
		try:
		res = await self._request(
		"GET",
		_LIST_SERVICE_INSTANCES_ENDPOINT,
		@@ -152,14 +189,22 @@ class FederationManager(FederationManagerPort):
		for obj in res:
		instances.append(ServiceInstance.model_validate(obj))
		return instances
		except FederationUnavailableError as exc:
		self._log_unavailable("list_full_service_instances", exc)
		return []

		async def get_service_instance(self, service_instance_id: UUID) -> ServiceInstance:
		if self._disabled("get_service_instance"):
		raise NotFoundException()
		try:
		res = await self._request(
		"GET",
		str.format(_GET_SERVICE_INSTANCE_ENDPOINT, service_instance_id=service_instance_id),
		)

		return ServiceInstance.model_validate(res)
		except FederationUnavailableError as exc:
		self._log_unavailable("get_service_instance", exc)
		raise NotFoundException() from exc

		async def request_deploy_service(self, command: dict[str, Any]) -> None:
		logger.info(

src/srm/adapters/transformation_functions/registry.py

+5 −0

Changes for src/srm/adapters/transformation_functions/registry.py: 5 added lines, 0 removed lines.
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
		@@ -31,6 +35,7 @@ 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:

SRM Bug fixes found after the wiring of TF-SDK

Summary

Merge request reports