Commit db8db3cb authored by Paris Stentoumis's avatar Paris Stentoumis
Browse files

feat: wip fm client

parent 0c248d79

.env.example

+3 −0
Original line number Diff line number Diff line
@@ -10,3 +10,6 @@ NATS_SETTINGS__URL = "nats://localhost:4222" 
NATS_SETTINGS__CONNECT_TIMEOUT = 10

NATS_SETTINGS__MAX_RECONNECT_ATTEMPTS = 3

NATS_SETTINGS__DRAIN_TIMEOUT = 30



FEDERATION_MANAGER_SETTINGS__BASE_URL = "http://localhost:8082"

FEDERATION_MANAGER_SETTINGS__TIMEOUT = 0.5

src/srm/adapters/errors.py

+12 −1
Original line number Diff line number Diff line 
from typing import ClassVar

from typing import Any, ClassVar





class DuplicateEntryError(Exception):
@@ -15,3 +15,14 @@ class DuplicateServiceSpecificationError(DuplicateEntryError): 
class ServiceSpecificationInUseError(Exception):

    def __init__(self) -> None:

        super().__init__("Service specification has live service instances.")





class NotFoundException(Exception):

    def __init__(self) -> None:

        super().__init__("Resource Not Found")





class DownstreamServiceException(Exception):

    def __init__(self, message: str = "Service Unavailable", details: Any | None = None):

        super().__init__(message)

        self._details = details

src/srm/adapters/federation_manager/fm_client.py

0 → 100644
+139 −0
Original line number Diff line number Diff line 
"""HTTP/Nats Federation Manager adapter."""



from typing import Any

from uuid import UUID



import httpx

import structlog



from srm.adapters.errors import DownstreamServiceException, NotFoundException

from srm.config import FederationManagerSettings

from srm.domain.models.runtime_inventory.models import ServiceInstance

from srm.domain.models.topology.models import Zone

from srm.domain.ports.databus.publisher import DataBusPublisher

from srm.domain.ports.federation_manager.ports import FederationManagerPort



_GET_ZONE_ENDPOINT = "/internal/federated-zones/{zone_id}"

_LIST_ZONES_ENDPOINT = "/internal/federated-zones"

_GET_SERVICE_INSTANCE_ENDPOINT = "/internal/federated-service-instances/{service_instance_id}"

_LIST_SERVICE_INSTANCES_ENDPOINT = "/internal/federated-service-instances"





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





class FederationManager(FederationManagerPort):

    def __init__(self, settings: FederationManagerSettings, publisher: DataBusPublisher):

        self._settings = settings

        self._publisher = publisher

        logger.info("fm_client_init", settings=self._settings)



    async def _request(

        self,

        method: str,

        path: str,

        json: dict[str, Any] | None = None,

        params: dict[str, Any] | None = None,

        headers: dict[str, str] | None = None,

    ) -> Any:

        url = f"{self._settings.base_url}{path}"

        log = logger.bind(

            method=method,

            url=url,

        )



        try:

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

                response = await client.request(

                    method=method,

                    url=url,

                    json=json,

                    params=params,

                    headers=headers,

                )



            if response.status_code == 404:

                log.warning("FM resource not found", url=url)

                raise NotFoundException()



            if response.status_code >= 400:

                log.error("FM returned error", status=response.status_code, body=response.text)

                raise DownstreamServiceException(

                    message="FM request failed",

                    details={

                        "status_code": response.status_code,

                        "response": response.text,

                    },

                )



            if response.status_code == 204:

                return None



            if not response.content:

                log.error(

                    "FM returned empty body for non-204 response", status=response.status_code

                )

                raise DownstreamServiceException(

                    message="FM returned empty response",

                    details={"status_code": response.status_code},

                )



            return response.json()



        except httpx.TimeoutException as exc:

            log.exception("FM request timed out", error=str(exc))

            raise DownstreamServiceException(

                message="FM request timed out",

                details=str(exc),

            )

        except httpx.ConnectError as exc:

            log.exception("FM connection failed", error=str(exc))

            raise DownstreamServiceException(

                message="Could not connect to FM",

                details=str(exc),

            )

        except httpx.RequestError as exc:

            log.exception("FM request error", error=str(exc))

            raise DownstreamServiceException(

                message="FM request failed",

                details=str(exc),

            )



    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:

        try:

            res = await self._request("GET", str.format(_GET_ZONE_ENDPOINT, zone_id=zone_id))

            return Zone.model_validate_json(res)

        except NotFoundException:

            return None



    async def list_zones(self) -> list[Zone]:

        res = await self._request("GET", _LIST_ZONES_ENDPOINT)

        zones = []

        for obj in res:

            zones.append(Zone.model_validate_json(obj))

        return zones



    async def list_full_service_instances(self) -> list[ServiceInstance]:

        res = await self._request("GET", _LIST_SERVICE_INSTANCES_ENDPOINT)

        instances = []

        for obj in res:

            instances.append(ServiceInstance.model_validate_json(obj))

        return instances



    async def get_service_instance(self, service_instance_id: UUID) -> ServiceInstance:

        res = await self._request(

            "GET",

            str.format(_GET_SERVICE_INSTANCE_ENDPOINT, service_instance_id=service_instance_id),

        )



        return ServiceInstance.model_validate_json(res)



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

        logger.info(

            "deploy_service_on_partner",

            command=command,

        )

        await self._publisher.publish("command.fm.service.deploy", command)

src/srm/api/databus/dependencies.py

+13 −5
Original line number Diff line number Diff line
@@ -13,6 +13,7 @@ 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.federation_manager.fm_client import FederationManager

from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator

from srm.application.command_handlers.network_capability_activate import (

    NetworkCapabilityActivateCommandCoordinator,
@@ -25,15 +26,19 @@ from srm.domain.ports.databus.publisher import DataBusPublisher 




def get_deploy_service_use_case(

    session: AsyncSession,

    connection_manager: NatsConnectionManager,

    session: AsyncSession, connection_manager: NatsConnectionManager, fm_client: FederationManager

) -> DeployServiceUseCase:

    return build_deploy_service_use_case(session, NatsPublisher(connection_manager))

    return build_deploy_service_use_case(

        session,

        NatsPublisher(connection_manager),

        fm_client=fm_client,

    )





def build_deploy_service_use_case(

    session: AsyncSession,

    publisher: DataBusPublisher,

    fm_client: FederationManager,

) -> DeployServiceUseCase:

    return DeployServiceUseCase(

        service_specifications=SqlServiceSpecificationRepository(session),
@@ -44,17 +49,20 @@ def build_deploy_service_use_case( 
        capability_instances=SqlCapabilityInstanceRepository(session),

        zones=SqlZoneRepository(session),

        publisher=publisher,

        fm_client=fm_client,

    )





def get_deploy_service_command_coordinator(

    session_maker: async_sessionmaker[AsyncSession],

    connection_manager: NatsConnectionManager,

    pub: DataBusPublisher,

    fm_client: FederationManager,

) -> DeployServiceCommandCoordinator:

    return DeployServiceCommandCoordinator(

        session_maker=session_maker,

        use_case_factory=build_deploy_service_use_case,

        publisher=NatsPublisher(connection_manager),

        publisher=pub,

        fm_client=fm_client,

    )

src/srm/api/databus/nats_subscriber.py

+9 −1
Original line number Diff line number Diff line
@@ -7,11 +7,14 @@ from pydantic import ValidationError 
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker



from srm.adapters.databus.nats_connection_manager import NatsConnectionManager

from srm.adapters.databus.nats_publisher import NatsPublisher

from srm.adapters.federation_manager.fm_client import FederationManager

from srm.api.databus.dependencies import (

    get_deploy_service_command_coordinator,

    get_network_capability_activate_command_coordinator,

)

from srm.api.databus.schemas import AnswerableCommandV1, InboundMessage

from srm.config import FederationManagerSettings



logger: structlog.BoundLogger = structlog.get_logger(__name__)
@@ -45,10 +48,15 @@ COMMAND_SUBJECTS = ( 
async def subscribe_to_subjects(

    connection_manager: NatsConnectionManager,

    session_maker: async_sessionmaker[AsyncSession],

    fm_settings: FederationManagerSettings,

) -> list["NatsSubscriber"]:

    publisher = NatsPublisher(connection_manager=connection_manager)

    fm_client = FederationManager(settings=fm_settings, publisher=publisher)



    deploy_coordinator = get_deploy_service_command_coordinator(

        session_maker,

        connection_manager,

        pub=publisher,

        fm_client=fm_client,

    )

    network_activate_coordinator = get_network_capability_activate_command_coordinator(

        session_maker,