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

SRM Refactor: Wire OpenOp TF-SDK as the control-path adapter factory and use... 

SRM Refactor: Wire OpenOp TF-SDK as the control-path adapter factory and use it for service deploy and network capability activation and several bug fixes.

1. Network Capability Deactivate and Service Terminate Commands
2. SRM Bug fixes found after the wiring of TF-SDK
3. Capability Instance monitor with operation-status publish
parent 5883b516

.env.example

+2 −0
Changes for .env.example: 2 added lines, 0 removed lines.
Original line number Diff line number Diff line 
APP_NAME="Service Resource Manager"

APP_DESCRIPTION="Service Resource Manager is the brains of the ETSI SDG Open Operator Platform - OpenOP"

APP_VERSION="1.5.0"

INSTANCES_MONITOR_INTERVAL=10.0



POSTGRES_SETTINGS__URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/srm"

POSTGRES_SETTINGS__ECHO = true
@@ -13,3 +14,4 @@ NATS_SETTINGS__DRAIN_TIMEOUT = 30 


FEDERATION_MANAGER_SETTINGS__BASE_URL = "http://localhost:8082"

FEDERATION_MANAGER_SETTINGS__TIMEOUT = 0.5

FEDERATION_MANAGER_SETTINGS__ENABLED = true

Dockerfile

+2 −0
Changes for Dockerfile: 2 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -17,6 +17,8 @@ RUN uv sync --locked 


FROM python:3.12-slim AS runtime



COPY --from=alpine/helm:3.18.6 /usr/bin/helm /usr/local/bin/helm



WORKDIR /app



COPY --from=builder /app/.venv /app/.venv

pyproject.toml

+10 −1
Changes for pyproject.toml: 10 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -18,10 +18,11 @@ dependencies = [ 
    "pydantic-settings>=2.13.1",

    "python-dotenv>=1.2.2",

    "structlog>=25.5.0",

    "sunrise6g-opensdk==2.0.0",

    "sqlalchemy>=2.0.48",

    "asyncpg>=0.31.0",

    "nats-py>=2.10.0",

    "openop-tfsdk==2.0.0rc2",

    "apscheduler<4.0.0",

]



[project.optional-dependencies]
@@ -37,6 +38,14 @@ dev = [ 
    "ruff>=0.15.6",

]



[tool.uv.sources]

openop-tfsdk = { index = "etsi-tf-sdk" }



[[tool.uv.index]]

name = "etsi-tf-sdk"

url = "https://labs.etsi.org/rep/api/v4/projects/396/packages/pypi/simple"

explicit = true



[tool.ruff]

line-length = 100

target-version = "py312"

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/catalog.py

+12 −0
Changes for src/srm/adapters/database/repos/catalog.py: 12 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -17,6 +17,7 @@ from srm.adapters.database.sql import ( 
    ServiceCapabilityRequirementRow,

    ServiceDeploymentUnitRow,

    ServiceInstanceRow,

    ServiceOrderRow,

    ServiceSpecificationRow,

)

from srm.adapters.errors import DuplicateServiceSpecificationError, ServiceSpecificationInUseError
@@ -119,6 +120,17 @@ class SqlServiceSpecificationRepository(ServiceSpecificationRepository): 
            .limit(1)

        )

        if any_instance_id is None:

            # An instance a partner hosts leaves no instance row here, only its order, and that

            # order still points at the specification: keep it as retired history instead.

            any_order_id = await self._session.scalar(

                select(ServiceOrderRow.id)

                .where(ServiceOrderRow.service_specification_id == id)

                .limit(1)

            )

            if any_order_id is not None:

                row.state = ServiceSpecificationState.RETIRED

                await self._session.flush()

                return True

            await self._session.execute(

                delete(ServiceCapabilityRequirementRow).where(

                    ServiceCapabilityRequirementRow.service_specification_id == id