feat: deliver cloudevent webhook callbacks on completion (cccc232c) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/http/callback_client.py

0 → 100644

+24 −0

Original line number	Diff line number	Diff line
		import httpx
		import structlog

		from open_exposure_gateway.core.config import get_settings
		from open_exposure_gateway.domain.edge_application_management import (
		AppInstanceStatusChangeCloudEvent,
		)

		logger = structlog.get_logger(__name__)


		class HttpCallbackClient:
		def __init__(self) -> None:
		settings = get_settings()
		self.timeout = settings.callback_settings.timeout

		async def deliver(self, sink: str, event: AppInstanceStatusChangeCloudEvent) -> 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()

src/open_exposure_gateway/application/mappers/edge_application_mapper.py

+27 −1

Original line number	Diff line number	Diff line
		import re
		from collections import defaultdict
		from typing import Any, Optional
		from uuid import UUID
		from uuid import UUID, uuid4

		from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas import (
		AccessEndpoint,
		@@ -30,6 +30,8 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas i
		)
		from open_exposure_gateway.domain.edge_application_management import (
		AppDeploymentTranslation,
		AppInstanceStatusChangeCloudEvent,
		AppInstanceStatusChangeData,
		ApplicationResources,
		AppRegistrationTranslation,
		AppRepo,
		@@ -60,6 +62,7 @@ from open_exposure_gateway.domain.edge_application_management import (
		SRMTerminatePayload,
		SRMTopologyConstraints,
		)
		from open_exposure_gateway.domain.models import AppInstance

		_PACKAGE_TYPE_TO_RUNTIME_KIND: dict[str, str] = {
		"HELM": "helm",
		@@ -78,6 +81,11 @@ _RUNTIME_KIND_TO_PACKAGE_TYPE: dict[str, str] = {
		v: k for k, v in _PACKAGE_TYPE_TO_RUNTIME_KIND.items()
		}

		# Placeholder pending a configured public NBI base URL (no such setting exists
		# yet, and handle_completed runs from a NATS message, not an HTTP request, so
		# there's no live request to derive it from).
		_CALLBACK_EVENT_SOURCE = "https://api.example.com/edge-application-management/v1"

		_VISIBILITY_REVERSE_MAP: dict[str, str] = {v: k for k, v in _VISIBILITY_MAP.items()}

		_SRM_STATE_TO_APP_INSTANCE_STATUS: dict[str, AppInstanceStatus] = {
		@@ -613,3 +621,21 @@ def build_terminate_instance_command(
		service_instance_id=str(app_instance_id),
		terminate=SRMTerminatePayload(),
		)


		def build_app_instance_status_change_event(
		app_instance: AppInstance,
		app_id: UUID,
		occurred_at: str,
		) -> AppInstanceStatusChangeCloudEvent:
		return AppInstanceStatusChangeCloudEvent(
		id=uuid4(),
		source=_CALLBACK_EVENT_SOURCE,
		time=occurred_at,
		data=AppInstanceStatusChangeData(
		appInstanceId=app_instance.app_instance_id,
		appId=app_id,
		edgeCloudZoneId=app_instance.edge_cloud_zone_id,
		status=app_instance.state.value,
		),
		)

src/open_exposure_gateway/application/services/edge_application_management_service.py

+70 −3

Original line number	Diff line number	Diff line
		@@ -19,6 +19,7 @@ from open_exposure_gateway.api.camara.edge_application_management.vwip.schemas i
		from open_exposure_gateway.application.mappers.edge_application_mapper import (
		build_app_deployment_translation,
		build_app_instance_info,
		build_app_instance_status_change_event,
		build_app_manifest,
		build_app_registration_translation,
		build_catalog_payload,
		@@ -44,13 +45,18 @@ from open_exposure_gateway.domain.models import (
		AppInstanceState,
		AppRegistration,
		AppRegistrationStatus,
		CallbackDelivery,
		CallbackRegistration,
		Operation,
		OperationStatus,
		OperationType,
		PackageType,
		)
		from open_exposure_gateway.ports.database.callbacks import CallbackRegistrationRepository
		from open_exposure_gateway.ports.callback_delivery_port import CallbackDeliveryPort
		from open_exposure_gateway.ports.database.callbacks import (
		CallbackDeliveryRepository,
		CallbackRegistrationRepository,
		)
		from open_exposure_gateway.ports.database.instances import AppInstanceRepository
		from open_exposure_gateway.ports.database.operations import OperationRepository
		from open_exposure_gateway.ports.database.registration import AppRegistrationRepository
		@@ -86,6 +92,8 @@ class EdgeApplicationManagementService:
		operation_repo: OperationRepository \| None = None,
		app_instance_repo: AppInstanceRepository \| None = None,
		callback_registration_repo: CallbackRegistrationRepository \| None = None,
		callback_delivery_port: CallbackDeliveryPort \| None = None,
		callback_delivery_repo: CallbackDeliveryRepository \| None = None,
		) -> None:
		self.srm_client = srm_client
		self._publisher = publisher
		@@ -93,6 +101,8 @@ class EdgeApplicationManagementService:
		self._operation_repo = operation_repo
		self._app_instance_repo = app_instance_repo
		self._callback_registration_repo = callback_registration_repo
		self._callback_delivery_port = callback_delivery_port
		self._callback_delivery_repo = callback_delivery_repo

		async def get_edge_cloud_zones(
		self,
		@@ -399,6 +409,8 @@ class EdgeApplicationManagementService:
		)
		await self._operation_repo.save(updated)

		updated_instances: list[AppInstance] = []

		if event.instances:
		for instance in event.instances:
		app_instance_id = UUID(instance.service_instance_id)
		@@ -409,11 +421,12 @@ class EdgeApplicationManagementService:
		app_instance_id=instance.service_instance_id,
		)
		continue
		await self._app_instance_repo.save(
		saved = await self._app_instance_repo.save(
		app_instance.model_copy(
		update={"state": _APP_INSTANCE_COMPLETION_STATE_MAP[instance.status]}
		)
		)
		updated_instances.append(saved)
		elif status == OperationStatus.FAILED:
		# Total failure carries no instances[] entries to match against.
		# POST /appinstances always creates exactly one app_instances row
		@@ -421,6 +434,60 @@ class EdgeApplicationManagementService:
		# leaving the pre-created row stuck at instantiating forever.
		app_instance = await self._app_instance_repo.get_by_operation_id(operation_id)
		if app_instance is not None:
		await self._app_instance_repo.save(
		saved = await self._app_instance_repo.save(
		app_instance.model_copy(update={"state": AppInstanceState.FAILED})
		)
		updated_instances.append(saved)

		if updated_instances:
		await self._deliver_callbacks(operation_id, event.completed_at, updated_instances)

		async def _deliver_callbacks(
		self,
		operation_id: UUID,
		occurred_at: str,
		app_instances: list[AppInstance],
		) -> None:
		if self._callback_registration_repo is None:
		return
		registration = await self._callback_registration_repo.get_by_operation_id(operation_id)
		if registration is None or not registration.is_active:
		return
		if self._callback_delivery_port is None or self._callback_delivery_repo is None:
		raise RuntimeError("CallbackDeliveryPort/CallbackDeliveryRepository is not available")

		for app_instance in app_instances:
		app_id = await self._resolve_app_id(app_instance.app_registration_id)
		if app_id is None:
		logger.warning(
		"callback_skipped_unresolvable_app_id",
		app_instance_id=str(app_instance.app_instance_id),
		)
		continue

		cloud_event = build_app_instance_status_change_event(app_instance, app_id, occurred_at)
		last_error: Optional[str] = None
		try:
		await self._callback_delivery_port.deliver(registration.sink, cloud_event)
		state = "delivered"
		except Exception as exc:
		state = "failed"
		last_error = str(exc)
		logger.warning("callback_delivery_failed", sink=registration.sink, error=str(exc))

		await self._callback_delivery_repo.save(
		CallbackDelivery(
		id=uuid4(),
		callback_registration_id=registration.id,
		operation_id=operation_id,
		attempt=1,
		state=state,
		last_error=last_error,
		)
		)

		async def _resolve_app_id(self, app_registration_id: UUID) -> Optional[UUID]:
		if self._app_registration_repo is None:
		return None
		app_registration = await self._app_registration_repo.get_by_id(app_registration_id)
		return app_registration.app_id if app_registration else None

src/open_exposure_gateway/core/config.py

+5 −0

Original line number	Diff line number	Diff line
		@@ -21,6 +21,10 @@ class NatsSettings(BaseModel):
		max_reconnect_attempts: int = 3


		class CallbackSettings(BaseModel):
		timeout: float = 10.0


		class ObservabilitySettings(BaseModel):
		log_level: str = "INFO"

		@@ -43,6 +47,7 @@ class Settings(BaseSettings):
		postgresql_settings: PostgreSQLSettings = PostgreSQLSettings()
		nats_settings: NatsSettings = NatsSettings()
		observability_settings: ObservabilitySettings = ObservabilitySettings()
		callback_settings: CallbackSettings = CallbackSettings()


		@lru_cache

src/open_exposure_gateway/domain/edge_application_management.py

+22 −0

Original line number	Diff line number	Diff line
		@@ -318,3 +318,25 @@ class SRMServiceInstance(BaseModel):
		resource_zone_id: str \| None = None
		name: str \| None = None
		capability_instances: list[SRMCapabilityInstanceSummary] = []


		class AppInstanceStatusChangeData(BaseModel):
		appInstanceId: UUID
		appId: UUID
		edgeCloudZoneId: UUID
		status: str


		class AppInstanceStatusChangeCloudEvent(BaseModel):
		# CloudEvents v1.0 attributes, per the vendored spec's onAppInstanceStatusChange
		# callback (ADR-0008). The referenced ../common/CAMARA_event_common.yaml
		# defining the shared CloudEvent schema is not itself vendored into this repo;
		# this shape is transcribed from the worked example in
		# architecture/oeg/app-instance-flow.md instead.
		id: UUID
		source: str
		specversion: str = "1.0"
		type: str = "org.camaraproject.edge-application-management.v0.app-instance-status-change"
		time: str
		datacontenttype: str = "application/json"
		data: AppInstanceStatusChangeData

Admin message