Refactor/qod

George Papathanailrequested to merge
refactor/qod into develop

Async QoD sessions: POST/DELETE /sessions, completion handling, qos-profiles stubs Summary Brings QoD POST/DELETE /qod/v0/sessions in line with the same async pattern used for POST /appinstances: OEG mints its own ids, persists bookkeeping rows, publishes a command.srm.* message, and reacts to event.srm.operation.completed — instead of the previous synchronous SRMClient calls against SRM endpoints that don't actually exist. Also fixes the NBI schemas to match the vendored qod-api.yaml, and adds stub routes for the two QoD operations that had none at all.

What's included:

  • CAMARA schema conformance (schemas.py/router.py): renamed to the spec's own names (CreateSession/SessionInfo), fixed the sink→webhook field, added the validators the spec requires (device/applicationServer presence, IPv4 anyOf, qosProfile pattern, ports), declared the documented error responses.
  • POST /sessions async: SRMNetworkCapabilityActivateCommand (domain/quality_on_demand.py), build_activate_command mapper, and the service rewrite — mints operation_id/session_id/service_specification_id, persists Operation (PENDING) + a new qod_sessions row (REQUESTED), publishes the activate command, returns 201 built from the request (per the spec's own REQUESTED-status design — no 202 deviation needed).
  • qod_sessions persistence: new table/domain model/repo, scoped to async bookkeeping only — not a read cache for GET.
  • DELETE /sessions/{id} async: new OperationType.NETWORK_CAPABILITY_DEACTIVATE, SRMNetworkCapabilityDeactivateCommand/build_deactivate_command, service rewrite — resolves external_ref from qod_sessions, 404s if the session is unknown or not yet confirmed (still REQUESTED), publishes the deactivate command keyed on external_ref, returns 204.
  • GET /qos-profiles, GET /qos-profiles/{name}: routes added, both return 501 — the catalog itself isn't implemented, but callers now get a proper response instead of a route-not-found 404.

Open points:

  1. GET /sessions/{id} no confirmed SRM contract to build against. srm/interface-contract.md §E.1 only documents reads by service_instance_id or operation_id; nothing keyed by external_ref/caller-facing sessionId. Needs SRM-side input: is there an endpoint (existing or planned) to read a capability_instance by external_ref? If we resolve service_instance_id ourselves instead, does GET /internal/service-instances/{id} actually return result_summary/qos_status, or only service_instance-level fields as documented?

  2. event.srm.operation.status semantics for backend-driven drops are undefined. SrmOperationStatusV1.state (§C.1) is accepted | failed_before_start | completed | failed | in_progress — none obviously means "a previously-AVAILABLE session just went UNAVAILABLE." What value/field actually signals this, and does the event reuse the original activate's operation_id?

  3. operations.operation_type enum doc is stale. oeg/persistence-model.md line 176 lists deploy, scale, terminate, network_capability — doesn't include network_capability_deactivate, which this branch added to distinguish activate-completions from deactivate-completions in handle_completed.

  4. service_specification_id has no catalog backing. Minted fresh (uuid4()) per request with nothing onboarded behind it. srm/interface-contract.md §B.5 marks it "required for catalog/profile validation" — if SRM's real activate handler enforces that, this will reject. Needs confirmation: does SRM validate this today, and if so what's the onboarding path for QoD profiles?

Edited by George Papathanail

Merge request reports

src/open_exposure_gateway/adapters/database/repos/qod_sessions.py

0 → 100644
+32 −0
Changes for src/open_exposure_gateway/adapters/database/repos/qod_sessions.py: 32 added lines, 0 removed lines.
Original line number Diff line number Diff line 
from uuid import UUID



from sqlalchemy import select

from sqlalchemy.ext.asyncio import AsyncSession



from open_exposure_gateway.adapters.database.mappers import QodSessionMapper

from open_exposure_gateway.adapters.database.sql import QodSessionRow

from open_exposure_gateway.domain.models import QodSession

from open_exposure_gateway.ports.database.qod_sessions import QodSessionRepository





class SqlQodSessionRepository(QodSessionRepository):

    def __init__(self, session: AsyncSession) -> None:

        self._session = session



    async def get_by_id(self, session_id: UUID) -> QodSession | None:

        stmt = select(QodSessionRow).where(QodSessionRow.session_id == session_id)

        row = await self._session.scalar(stmt)

        return QodSessionMapper.to_domain(row) if row is not None else None



    async def get_by_operation_id(self, operation_id: UUID) -> QodSession | None:

        stmt = select(QodSessionRow).where(QodSessionRow.operation_id == operation_id)

        row = await self._session.scalar(stmt)

        return QodSessionMapper.to_domain(row) if row is not None else None



    async def save(self, qod_session: QodSession) -> QodSession:

        merged = await self._session.merge(QodSessionMapper.to_row(qod_session))

        await self._session.flush()

        saved = await self.get_by_id(merged.session_id)

        if saved is None:

            raise RuntimeError("Saved QoD session could not be reloaded")

        return saved

src/open_exposure_gateway/adapters/database/mappers.py

+34 −0
Changes for src/open_exposure_gateway/adapters/database/mappers.py: 34 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -4,6 +4,7 @@ from open_exposure_gateway.adapters.database.sql import ( 
    CallbackDeliveryRow,

    CallbackRegistrationRow,

    OperationRow,

    QodSessionRow,

)

from open_exposure_gateway.domain.models import (

    AppInstance,
@@ -11,6 +12,7 @@ from open_exposure_gateway.domain.models import ( 
    CallbackDelivery,

    CallbackRegistration,

    Operation,

    QodSession,

)
@@ -108,6 +110,38 @@ class AppInstanceMapper: 
        )





class QodSessionMapper:

    @staticmethod

    def to_domain(row: QodSessionRow) -> QodSession:

        return QodSession(

            session_id=row.session_id,

            operation_id=row.operation_id,

            service_specification_id=row.service_specification_id,

            qos_profile=row.qos_profile,

            duration_seconds=row.duration_seconds,

            state=row.state,

            external_ref=row.external_ref,

            device_ports=row.device_ports,

            application_server_ports=row.application_server_ports,

            created_at=row.created_at,

            updated_at=row.updated_at,

        )



    @staticmethod

    def to_row(domain: QodSession) -> QodSessionRow:

        return QodSessionRow(

            session_id=domain.session_id,

            operation_id=domain.operation_id,

            service_specification_id=domain.service_specification_id,

            qos_profile=domain.qos_profile,

            duration_seconds=domain.duration_seconds,

            state=domain.state,

            external_ref=domain.external_ref,

            device_ports=domain.device_ports,

            application_server_ports=domain.application_server_ports,

        )





class CallbackRegistrationMapper:

    @staticmethod

    def to_domain(row: CallbackRegistrationRow) -> CallbackRegistration:

src/open_exposure_gateway/adapters/database/sql.py

+18 −0
Changes for src/open_exposure_gateway/adapters/database/sql.py: 18 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -26,6 +26,7 @@ from open_exposure_gateway.domain.models.operations.enums import ( 
    OperationStatus,

    OperationType,

)

from open_exposure_gateway.domain.models.qod_sessions.enums import QodSessionState

from open_exposure_gateway.domain.models.registration.enums import (

    AppRegistrationStatus,

    PackageType,
@@ -184,3 +185,20 @@ class AppInstanceRow(AuditedMixin, Base): 
    )

    edge_cloud_zone_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), nullable=False)

    state: Mapped[AppInstanceState] = mapped_column(_enum_type(AppInstanceState), nullable=False)





class QodSessionRow(AuditedMixin, Base):

    __tablename__ = "qod_sessions"

    __table_args__ = (Index("idx_qod_sessions_operation", "operation_id"),)



    session_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)

    operation_id: Mapped[UUID] = mapped_column(

        ForeignKey("operations.operation_id"), nullable=False

    )

    service_specification_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), nullable=False)

    qos_profile: Mapped[str] = mapped_column(String(256), nullable=False)

    duration_seconds: Mapped[int] = mapped_column(Integer, nullable=False)

    state: Mapped[QodSessionState] = mapped_column(_enum_type(QodSessionState), nullable=False)

    external_ref: Mapped[str | None] = mapped_column(String(255))

    device_ports: Mapped[dict[str, object] | None] = mapped_column(JSONB)

    application_server_ports: Mapped[dict[str, object] | None] = mapped_column(JSONB)

src/open_exposure_gateway/adapters/databus/nats_adapter.py

+36 −0
Changes for src/open_exposure_gateway/adapters/databus/nats_adapter.py: 36 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -10,6 +10,7 @@ from pydantic import ValidationError 


from open_exposure_gateway.core.config import NatsSettings

from open_exposure_gateway.domain.edge_application_management import SRMOperationCompleted

from open_exposure_gateway.domain.quality_on_demand import SRMOperationStatus

from open_exposure_gateway.ports.databus_port import DataBusPort



logger: structlog.BoundLogger = structlog.get_logger(__name__)
@@ -109,3 +110,38 @@ class NatsOperationConsumer: 
            await self._handler(event)

        except Exception:

            logger.exception("operation_completed_handler_failed", operation_id=event.operation_id)





class NatsOperationStatusConsumer:

    def __init__(

        self,

        client: Client,

        subject: str,

        handler: Callable[[SRMOperationStatus], Awaitable[None]],

    ) -> None:

        self._client = client

        self._subject = subject

        self._handler = handler

        self._subscription: Subscription | None = None



    async def start(self) -> None:

        # TODO: same at-most-once caveat as NatsOperationConsumer.start() — see there.

        self._subscription = await self._client.subscribe(self._subject, cb=self._handle_message)



    async def _handle_message(self, msg: _Msg) -> None:

        try:

            raw: Any = json.loads(msg.data.decode())

        except (json.JSONDecodeError, UnicodeDecodeError):

            logger.warning("invalid_json", subject=msg.subject)

            return



        try:

            event = SRMOperationStatus.model_validate(raw)

        except ValidationError as exc:

            logger.warning("invalid_operation_status_event", subject=msg.subject, error=str(exc))

            return



        try:

            await self._handler(event)

        except Exception:

            logger.exception("operation_status_handler_failed", operation_id=event.operation_id)

src/open_exposure_gateway/adapters/http/qod_callback_client.py

0 → 100644
+19 −0
Changes for src/open_exposure_gateway/adapters/http/qod_callback_client.py: 19 added lines, 0 removed lines.
Original line number Diff line number Diff line 
import httpx



from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.domain.quality_on_demand import QosStatusChangedCloudEvent





class HttpQodCallbackClient:

    def __init__(self) -> None:

        settings = get_settings()

        self.timeout = settings.callback_settings.timeout



    async def deliver(self, sink: str, event: QosStatusChangedCloudEvent) -> None:

        async with httpx.AsyncClient(timeout=self.timeout) as client:

            response = await client.post(

                sink,

                content=event.model_dump_json(),

                headers={"Content-Type": "application/cloudevents+json"},

            )

        response.raise_for_status()