feat: add HTTP tf-sdk mock seam and terminate-service command handler (060da092) · Commits · OOP / code / service-resource-manager

src/srm/adapters/transformation_functions/init.py

0 → 100644

+9 −0

Original line number	Diff line number	Diff line
		from srm.adapters.transformation_functions.http_mock import (
		HttpMockTransformationFunctionExecutionAdapter,
		build_transformation_function_execution_adapter,
		)

		__all__ = [
		"HttpMockTransformationFunctionExecutionAdapter",
		"build_transformation_function_execution_adapter",
		]

src/srm/adapters/transformation_functions/http_mock.py

0 → 100644

+274 −0

Original line number	Diff line number	Diff line
		"""HTTP-backed mock implementation of the transformation-function execution port.

		The real tf-sdk is not vendored on this branch. This adapter satisfies the
		ADR-0027 ``TransformationFunctionExecutionPort`` contract by forwarding the two
		lifecycle operations SRM's service pipeline uses -- ``activate_capability`` and
		``deactivate_capability`` -- to the ``tf-sdk-mock`` HTTP service, which turns
		them into ``kubectl apply`` / ``kubectl delete`` against a kind cluster.

		Contract of the remote service (see the OOP e2e harness ``tf-sdk-mock/app.py``):

		POST {base_url}/v1/deploy
		{operation_id, namespace, name, image, ports:[int], replicas,
		control_path_ref, labels:{}}
		-> 200 {status, external_id, external_ref, endpoints:[{interface_id,fqdn,port}]}

		POST {base_url}/v1/terminate
		{operation_id, namespace}
		-> 200 {status}

		Discovery, update and query operations are not implemented -- SRM's deploy path
		reads topology from its own database and never calls them on this adapter.
		"""

		from __future__ import annotations

		import re
		from typing import Any, NoReturn
		from uuid import UUID

		import httpx
		import structlog

		from srm.config import get_tf_sdk_settings
		from srm.domain.models.canonical_parameters.result import Result, ResultStatus, ServiceEndpoint
		from srm.domain.models.runtime_inventory import CapabilityInstanceState
		from srm.domain.models.topology import CapabilityKind, DomainKind, ZoneKind
		from srm.domain.ports.transformation_functions import (
		CapabilityLifecycleRequest,
		CapabilityLifecycleResult,
		CapabilityQueryRequest,
		CapabilityQueryResult,
		CapabilityRuntimeState,
		CapabilityRuntimeStateRequest,
		DomainExecutionFailed,
		TransformationFunctionCapability,
		TransformationFunctionControlPathBinding,
		TransformationFunctionDomain,
		TransformationFunctionExecutionPort,
		TransformationFunctionZone,
		UnsupportedAction,
		)

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


		def _k8s_name(raw: str) -> str:
		name = re.sub(r"[^a-z0-9-]", "-", raw.lower()).strip("-")[:40].strip("-")
		return name or "app"


		def _namespace_for(service_instance_id: str) -> str:
		return "oop-" + UUID(service_instance_id).hex[:20]


		class HttpMockTransformationFunctionExecutionAdapter(TransformationFunctionExecutionPort):
		def __init__(self, base_url: str, timeout_seconds: float = 120.0) -> None:
		self._base_url = base_url.rstrip("/")
		self._timeout = timeout_seconds

		# --- execution -------------------------------------------------------------

		def activate_capability(
		self,
		request: CapabilityLifecycleRequest,
		) -> CapabilityLifecycleResult:
		unit = request.deployment_unit
		if unit is None or not unit.artifact_ref:
		raise DomainExecutionFailed(
		"tf_sdk_missing_artifact",
		"activate_capability needs a deployment_unit with an artifact_ref",
		)
		service_instance_id = str(request.metadata.get("service_instance_id", request.operation_id))
		namespace = _namespace_for(service_instance_id)
		name = _k8s_name(str(request.metadata.get("name") or unit.deployment_unit_ref))
		ports = _ports_from_request(request) or [80]

		body = self._post(
		"/v1/deploy",
		{
		"operation_id": request.operation_id,
		"namespace": namespace,
		"name": name,
		"image": unit.artifact_ref,
		"ports": ports,
		"replicas": 1,
		"control_path_ref": request.target.control_path_ref,
		"labels": {"srm.service-instance-id": service_instance_id},
		},
		)

		external_id = str(body.get("external_id") or f"{namespace}/{name}")
		endpoints = [
		ServiceEndpoint(
		interface_id=str(item.get("interface_id", "")),
		fqdn=str(item.get("fqdn", "")),
		port=int(item.get("port", 0)),
		)
		for item in body.get("endpoints", [])
		if isinstance(item, dict)
		]
		return CapabilityLifecycleResult(
		operation_id=request.operation_id,
		capability_kind=request.capability_kind,
		control_path_ref_snapshot=request.target.control_path_ref,
		zone_ref=request.target.zone_ref,
		domain_ref=request.target.domain_ref,
		capability_ref=request.target.capability_ref,
		control_path_binding_ref=request.target.control_path_binding_ref,
		state=CapabilityInstanceState.ACTIVE,
		external_id=external_id,
		external_ref=str(body.get("external_ref") or external_id),
		resource_type="k8s.deployment",
		endpoints=endpoints,
		result_summary=Result(status=ResultStatus.ACTIVE, endpoints=endpoints),
		)

		def deactivate_capability(
		self,
		request: CapabilityLifecycleRequest,
		) -> CapabilityLifecycleResult:
		namespace = self._namespace_for_teardown(request)
		self._post(
		"/v1/terminate",
		{"operation_id": request.operation_id, "namespace": namespace},
		)
		return CapabilityLifecycleResult(
		operation_id=request.operation_id,
		capability_kind=request.capability_kind,
		control_path_ref_snapshot=request.target.control_path_ref,
		zone_ref=request.target.zone_ref,
		domain_ref=request.target.domain_ref,
		capability_ref=request.target.capability_ref,
		control_path_binding_ref=request.target.control_path_binding_ref,
		state=CapabilityInstanceState.TERMINATED,
		external_id=request.external_id,
		external_ref=request.external_id,
		result_summary=Result(status=ResultStatus.ACTIVE),
		)

		def update_capability(
		self,
		request: CapabilityLifecycleRequest,
		) -> CapabilityLifecycleResult:
		self._unsupported("update_capability")

		def query_capability(
		self,
		request: CapabilityQueryRequest,
		) -> CapabilityQueryResult:
		self._unsupported("query_capability")

		def get_capability_runtime_state(
		self,
		request: CapabilityRuntimeStateRequest,
		) -> CapabilityRuntimeState:
		self._unsupported("get_capability_runtime_state")

		# --- discovery (unused on this path) -------------------------------------

		def get_zones(self, kind: ZoneKind \| None = None) -> list[TransformationFunctionZone]:
		self._unsupported("get_zones")

		def get_domains(
		self,
		zone_ref: str \| None = None,
		kind: DomainKind \| None = None,
		) -> list[TransformationFunctionDomain]:
		self._unsupported("get_domains")

		def get_capabilities(
		self,
		domain_ref: str \| None = None,
		kind: CapabilityKind \| None = None,
		) -> list[TransformationFunctionCapability]:
		self._unsupported("get_capabilities")

		def get_control_path_bindings(
		self,
		capability_ref: str \| None = None,
		) -> list[TransformationFunctionControlPathBinding]:
		self._unsupported("get_control_path_bindings")

		# --- internals ---------------------------------------------------------

		@staticmethod
		def _namespace_for_teardown(request: CapabilityLifecycleRequest) -> str:
		if request.external_id and "/" in request.external_id:
		return request.external_id.split("/", 1)[0]
		service_instance_id = request.metadata.get("service_instance_id")
		if service_instance_id:
		return _namespace_for(str(service_instance_id))
		return _namespace_for(request.operation_id)

		def _post(self, path: str, payload: dict[str, Any]) -> dict[str, Any]:
		url = f"{self._base_url}{path}"
		try:
		with httpx.Client(timeout=self._timeout) as client:
		response = client.post(url, json=payload)
		except httpx.HTTPError as exc:
		logger.error("tf_sdk_request_failed", url=url, error=str(exc))
		raise DomainExecutionFailed(
		"tf_sdk_transport", f"tf-sdk request to {path} failed: {exc}"
		) from exc

		if response.status_code >= 400:
		detail = _safe_detail(response)
		logger.error(
		"tf_sdk_returned_error", url=url, status_code=response.status_code, detail=detail
		)
		raise DomainExecutionFailed(
		"tf_sdk_http",
		f"tf-sdk {path} returned {response.status_code}: {detail}",
		details={"status_code": response.status_code, "detail": detail},
		)

		try:
		data = response.json()
		except ValueError as exc:
		raise DomainExecutionFailed(
		"tf_sdk_bad_body", f"tf-sdk {path} returned a non-JSON body"
		) from exc
		if not isinstance(data, dict):
		raise DomainExecutionFailed(
		"tf_sdk_bad_body", f"tf-sdk {path} returned {type(data).__name__}, expected object"
		)
		return data

		@staticmethod
		def _unsupported(operation: str) -> NoReturn:
		raise UnsupportedAction(
		"tf_sdk_unsupported",
		f"{operation} is not implemented by the tf-sdk HTTP mock",
		)


		def _ports_from_request(request: CapabilityLifecycleRequest) -> list[int]:
		unit = request.deployment_unit
		if unit is None:
		return []
		ports = {interface.port for interface in unit.exposed_interfaces}
		if not ports and unit.resource_requirements is not None:
		ports = {interface.port for interface in unit.resource_requirements.interfaces}
		return sorted(ports)


		def _safe_detail(response: httpx.Response) -> str:
		try:
		body = response.json()
		except ValueError:
		return response.text[:500]
		if isinstance(body, dict) and "detail" in body:
		return str(body["detail"])[:500]
		return str(body)[:500]


		def build_transformation_function_execution_adapter() -> TransformationFunctionExecutionPort \| None:
		"""Construct the adapter from settings, or ``None`` when the seam is disabled."""
		settings = get_tf_sdk_settings()
		if not settings.enabled or not settings.base_url:
		return None
		return HttpMockTransformationFunctionExecutionAdapter(
		base_url=settings.base_url,
		timeout_seconds=settings.timeout_seconds,
		)

src/srm/api/databus/dependencies.py

+27 −0

Original line number	Diff line number	Diff line
		@@ -13,8 +13,14 @@ from srm.adapters.database.repos.runtime_inventory import (
		from srm.adapters.database.repos.topology import SqlZoneRepository
		from srm.adapters.databus.nats_connection_manager import NatsConnectionManager
		from srm.adapters.databus.nats_publisher import NatsPublisher
		from srm.adapters.transformation_functions import (
		build_transformation_function_execution_adapter,
		)
		from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator
		from srm.application.command_handlers.terminate_service import TerminateServiceCommandCoordinator
		from srm.application.use_cases.deploy_service import DeployServiceUseCase
		from srm.application.use_cases.terminate_service import TerminateServiceUseCase
		from srm.config import get_tf_sdk_settings
		from srm.domain.ports.databus.publisher import DataBusPublisher


		@@ -38,6 +44,7 @@ def build_deploy_service_use_case(
		capability_instances=SqlCapabilityInstanceRepository(session),
		zones=SqlZoneRepository(session),
		publisher=publisher,
		transformation_functions=build_transformation_function_execution_adapter(),
		)


		@@ -50,3 +57,23 @@ def get_deploy_service_command_coordinator(
		use_case_factory=build_deploy_service_use_case,
		publisher=NatsPublisher(connection_manager),
		)


		def build_terminate_service_use_case(session: AsyncSession) -> TerminateServiceUseCase:
		return TerminateServiceUseCase(
		service_instances=SqlServiceInstanceRepository(session),
		zones=SqlZoneRepository(session),
		transformation_functions=build_transformation_function_execution_adapter(),
		)


		def get_terminate_service_command_coordinator(
		session_maker: async_sessionmaker[AsyncSession],
		connection_manager: NatsConnectionManager,
		) -> TerminateServiceCommandCoordinator:
		return TerminateServiceCommandCoordinator(
		session_maker=session_maker,
		use_case_factory=build_terminate_service_use_case,
		publisher=NatsPublisher(connection_manager),
		debounce_seconds=get_tf_sdk_settings().terminate_debounce_seconds,
		)

src/srm/api/databus/nats_subscriber.py

+14 −3

Original line number	Diff line number	Diff line
		@@ -7,7 +7,10 @@ from pydantic import ValidationError
		from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

		from srm.adapters.databus.nats_connection_manager import NatsConnectionManager
		from srm.api.databus.dependencies import get_deploy_service_command_coordinator
		from srm.api.databus.dependencies import (
		get_deploy_service_command_coordinator,
		get_terminate_service_command_coordinator,
		)
		from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage

		logger: structlog.BoundLogger = structlog.get_logger(__name__)
		@@ -27,13 +30,16 @@ async def _noop_callback(envelope: AnswerableCommandV1) -> None:

		COMMAND_HANDLERS: dict[str, MessageHandler] = {
		"command.srm.service.scale": _noop_callback,
		"command.srm.service.terminate": _noop_callback,
		"command.srm.network.capability.activate": _noop_callback,
		"command.srm.network.capability.update": _noop_callback,
		"command.srm.network.capability.deactivate": _noop_callback,
		}

		COMMAND_SUBJECTS = ("command.srm.service.deploy", *COMMAND_HANDLERS.keys())
		COMMAND_SUBJECTS = (
		"command.srm.service.deploy",
		"command.srm.service.terminate",
		*COMMAND_HANDLERS.keys(),
		)


		async def subscribe_to_subjects(
		@@ -44,9 +50,14 @@ async def subscribe_to_subjects(
		session_maker,
		connection_manager,
		)
		terminate_coordinator = get_terminate_service_command_coordinator(
		session_maker,
		connection_manager,
		)

		command_handlers = {
		"command.srm.service.deploy": deploy_coordinator.handle,
		"command.srm.service.terminate": terminate_coordinator.handle,
		**COMMAND_HANDLERS,
		}
		subscribers = [

src/srm/application/command_handlers/terminate_service.py

0 → 100644

+190 −0

Original line number	Diff line number	Diff line
		from __future__ import annotations

		import asyncio
		from collections.abc import Callable
		from dataclasses import dataclass, field
		from datetime import UTC, datetime
		from uuid import UUID

		import structlog
		from pydantic import ValidationError
		from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

		from srm.api.databus.schemas import AnswerableCommandV1, SrmServiceTerminateV1
		from srm.application.use_cases.terminate_service import (
		TerminatedInstance,
		TerminateServiceCommand,
		TerminateServiceUseCase,
		)
		from srm.domain.ports.databus.events import OperationCompletedStatus, SrmOperationCompletedV1
		from srm.domain.ports.databus.publisher import DataBusPublisher

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

		TerminateUseCaseFactory = Callable[[AsyncSession], TerminateServiceUseCase]


		@dataclass
		class _Pending:
		correlation_id: str
		instances: list[TerminatedInstance] = field(default_factory=list)
		timer: asyncio.Task[None] \| None = None


		class TerminateServiceCommandCoordinator:
		"""Consumes ``command.srm.service.terminate`` and answers with a completion.

		OEG fans out one terminate command per instance for ``DELETE /deployments``,
		all sharing one ``operation_id``, and waits for a single
		``event.srm.operation.completed`` listing every instance. Commands are
		processed as they arrive, their results buffered per ``operation_id``, and a
		short debounce window collapses them into one event.
		"""

		def __init__(
		self,
		session_maker: async_sessionmaker[AsyncSession],
		use_case_factory: TerminateUseCaseFactory,
		publisher: DataBusPublisher,
		debounce_seconds: float = 2.5,
		) -> None:
		self._session_maker = session_maker
		self._use_case_factory = use_case_factory
		self._publisher = publisher
		self._debounce_seconds = debounce_seconds
		self._pending: dict[UUID, _Pending] = {}
		self._lock = asyncio.Lock()

		async def handle(self, envelope: AnswerableCommandV1) -> None:
		command = self._parse_command(envelope)
		if command is None:
		await self._publish_failed_before_start(envelope)
		return

		try:
		result = await self._run(command)
		except Exception:
		logger.exception(
		"terminate_service_failed",
		operation_id=str(command.operation_id),
		service_instance_id=str(command.service_instance_id),
		)
		await self._publish_failed(command)
		return

		async with self._lock:
		pending = self._pending.get(command.operation_id)
		if pending is None:
		pending = _Pending(correlation_id=command.correlation_id)
		self._pending[command.operation_id] = pending
		pending.instances.append(result)
		if pending.timer is not None:
		pending.timer.cancel()
		pending.timer = asyncio.create_task(self._flush_after_delay(command.operation_id))

		async def _run(self, command: TerminateServiceCommand) -> TerminatedInstance:
		async with self._session_maker() as session:
		try:
		use_case = self._use_case_factory(session)
		result = await use_case.terminate_one(command)
		await session.commit()
		return result
		except Exception:
		await session.rollback()
		raise

		async def _flush_after_delay(self, operation_id: UUID) -> None:
		try:
		await asyncio.sleep(self._debounce_seconds)
		except asyncio.CancelledError:
		return
		async with self._lock:
		pending = self._pending.pop(operation_id, None)
		if pending is None:
		return

		any_failed = any(item.status != "completed" for item in pending.instances)
		status = (
		OperationCompletedStatus.PARTIALLY_COMPLETED
		if any_failed
		else OperationCompletedStatus.COMPLETED
		)
		event = SrmOperationCompletedV1(
		schema_version="1.0",
		operation_id=operation_id,
		status=status,
		instances=[
		{
		"service_instance_id": str(item.service_instance_id),
		"zone_id": str(item.zone_id)
		if item.zone_id is not None
		else "00000000-0000-0000-0000-000000000000",
		"status": item.status,
		**({"error": item.error} if item.error is not None else {}),
		}
		for item in pending.instances
		],
		correlation_id=pending.correlation_id,
		completed_at=datetime.now(UTC),
		)
		await self._publisher.publish(
		"event.srm.operation.completed",
		event.model_dump(mode="json"),
		)
		logger.info(
		"terminate_service_completed",
		operation_id=str(operation_id),
		instance_count=len(pending.instances),
		status=status.value,
		)

		def _parse_command(
		self,
		envelope: AnswerableCommandV1,
		) -> TerminateServiceCommand \| None:
		try:
		message = SrmServiceTerminateV1.model_validate(envelope.model_dump(mode="python"))
		except ValidationError:
		logger.exception("terminate_command_invalid", operation_id=str(envelope.operation_id))
		return None
		return TerminateServiceCommand(
		operation_id=message.operation_id,
		correlation_id=message.correlation_id,
		service_instance_id=message.service_instance_id,
		app_provider_id=message.app_provider_id,
		)

		async def _publish_failed(self, command: TerminateServiceCommand) -> None:
		await self._publish_completed_failed(
		operation_id=command.operation_id,
		correlation_id=command.correlation_id,
		)

		async def _publish_failed_before_start(self, envelope: AnswerableCommandV1) -> None:
		await self._publish_completed_failed(
		operation_id=envelope.operation_id,
		correlation_id=envelope.correlation_id,
		)

		async def _publish_completed_failed(
		self,
		*,
		operation_id: UUID,
		correlation_id: str,
		) -> None:
		event = SrmOperationCompletedV1(
		schema_version="1.0",
		operation_id=operation_id,
		status=OperationCompletedStatus.FAILED,
		error={
		"type": "about:blank",
		"title": "Terminate service failed.",
		"status": 500,
		},
		correlation_id=correlation_id,
		completed_at=datetime.now(UTC),
		)
		await self._publisher.publish(
		"event.srm.operation.completed",
		event.model_dump(mode="json"),
		)