🐛 Adding metrics to dynamic-sidecar created services (ITISFoundation#3666)

Author: GitHK

packages/service-library/src/servicelib/aiohttp/monitor_services.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import List, Union
+from typing import Union
 
 from aiohttp import web
 from prometheus_client import Counter
@@ -26,12 +26,12 @@
 kSERVICE_STARTED = f"{__name__}.services_started"
 kSERVICE_STOPPED = f"{__name__}.services_stopped"
 
-SERVICE_STARTED_LABELS: List[str] = [
+SERVICE_STARTED_LABELS: list[str] = [
     "service_key",
     "service_tag",
 ]
 
-SERVICE_STOPPED_LABELS: List[str] = [
+SERVICE_STOPPED_LABELS: list[str] = [
     "service_key",
     "service_tag",
     "result",
@@ -66,11 +66,6 @@ class ServiceResult(Enum):
     FAILURE = "FAILURE"
 
 
-class ServiceType(Enum):
-    COMPUTATIONAL = 0
-    DYNAMIC = 1
-
-
 def service_started(
     # pylint: disable=too-many-arguments
     app: web.Application,

services/director-v2/src/simcore_service_director_v2/core/application.py

@@ -134,19 +134,26 @@ def init_app(settings: Optional[AppSettings] = None) -> FastAPI:
     if settings.DYNAMIC_SERVICES.DIRECTOR_V2_DYNAMIC_SERVICES_ENABLED:
         dynamic_services.setup(app, settings.DYNAMIC_SERVICES)
 
-    if settings.DYNAMIC_SERVICES.DYNAMIC_SIDECAR and (
+    dynamic_scheduler_enabled = settings.DYNAMIC_SERVICES.DYNAMIC_SIDECAR and (
         settings.DYNAMIC_SERVICES.DYNAMIC_SCHEDULER
         and settings.DYNAMIC_SERVICES.DYNAMIC_SCHEDULER.DIRECTOR_V2_DYNAMIC_SCHEDULER_ENABLED
-    ):
+    )
+
+    computational_backend_enabled = (
+        settings.DIRECTOR_V2_COMPUTATIONAL_BACKEND.COMPUTATIONAL_BACKEND_ENABLED
+    )
+    if dynamic_scheduler_enabled or computational_backend_enabled:
+        rabbitmq.setup(app)
+
+    if dynamic_scheduler_enabled:
         dynamic_sidecar.setup(app)
 
     if (
         settings.DIRECTOR_V2_COMPUTATIONAL_BACKEND.COMPUTATIONAL_BACKEND_DASK_CLIENT_ENABLED
     ):
         dask_clients_pool.setup(app, settings.DIRECTOR_V2_COMPUTATIONAL_BACKEND)
 
-    if settings.DIRECTOR_V2_COMPUTATIONAL_BACKEND.COMPUTATIONAL_BACKEND_ENABLED:
-        rabbitmq.setup(app)
+    if computational_backend_enabled:
         comp_scheduler.setup(app)
 
     node_rights.setup(app)

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/scheduler/_utils.py

@@ -3,13 +3,15 @@
 from typing import Any, Deque, Final, Optional
 
 from fastapi import FastAPI
+from models_library.rabbitmq_messages import InstrumentationRabbitMessage
 from pydantic import AnyHttpUrl
 from servicelib.fastapi.long_running_tasks.client import (
     ProgressCallback,
     TaskClientResultError,
 )
 from servicelib.fastapi.long_running_tasks.server import TaskProgress
 from servicelib.utils import logged_gather
+from simcore_postgres_database.models.comp_tasks import NodeClass
 
 from ....api.dependencies.database import get_base_repository
 from ....core.errors import NodeRightsAcquireError
@@ -19,6 +21,7 @@
     DockerStatus,
     SchedulerData,
 )
+from ....modules.rabbitmq import RabbitMQClient
 from ...db.repositories import BaseRepository
 from ...director_v0 import DirectorV0Client
 from ...node_rights import NodeRightsManager, ResourceName
@@ -275,3 +278,17 @@ async def _remove_containers_save_state_and_outputs() -> None:
     await service_remove_sidecar_proxy_docker_networks_and_volumes(
         TaskProgress.create(), app, scheduler_data, dynamic_sidecar_settings
     )
+
+    # instrumentation
+    message = InstrumentationRabbitMessage(
+        metrics="service_stopped",
+        user_id=scheduler_data.user_id,
+        project_id=scheduler_data.project_id,
+        node_id=scheduler_data.node_uuid,
+        service_uuid=scheduler_data.node_uuid,
+        service_type=NodeClass.INTERACTIVE.value,
+        service_key=scheduler_data.key,
+        service_tag=scheduler_data.version,
+    )
+    rabbitmq_client: RabbitMQClient = app.state.rabbitmq_client
+    await rabbitmq_client.publish(message.channel_name, message.json())

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/scheduler/events.py

@@ -10,6 +10,7 @@
 from models_library.projects_networks import ProjectsNetworks
 from models_library.projects_nodes import Node
 from models_library.projects_nodes_io import NodeIDStr
+from models_library.rabbitmq_messages import InstrumentationRabbitMessage
 from models_library.service_settings_labels import (
     SimcoreServiceLabels,
     SimcoreServiceSettingsLabel,
@@ -19,6 +20,7 @@
 from servicelib.fastapi.long_running_tasks.client import TaskId
 from servicelib.json_serialization import json_dumps
 from servicelib.utils import logged_gather
+from simcore_postgres_database.models.comp_tasks import NodeClass
 from simcore_service_director_v2.utils.dict_utils import nested_update
 from tenacity import TryAgain
 from tenacity._asyncio import AsyncRetrying
@@ -34,6 +36,7 @@
     DockerStatus,
 )
 from ....modules.director_v0 import DirectorV0Client
+from ....modules.rabbitmq import RabbitMQClient
 from ...catalog import CatalogClient
 from ...db.repositories.projects import ProjectsRepository
 from ...db.repositories.projects_networks import ProjectsNetworksRepository
@@ -101,6 +104,21 @@ async def will_trigger(cls, app: FastAPI, scheduler_data: SchedulerData) -> bool
 
     @classmethod
     async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
+
+        # instrumentation
+        message = InstrumentationRabbitMessage(
+            metrics="service_started",
+            user_id=scheduler_data.user_id,
+            project_id=scheduler_data.project_id,
+            node_id=scheduler_data.node_uuid,
+            service_uuid=scheduler_data.node_uuid,
+            service_type=NodeClass.INTERACTIVE.value,
+            service_key=scheduler_data.key,
+            service_tag=scheduler_data.version,
+        )
+        rabbitmq_client: RabbitMQClient = app.state.rabbitmq_client
+        await rabbitmq_client.publish(message.channel_name, message.json())
+
         dynamic_sidecar_settings: DynamicSidecarSettings = (
             app.state.settings.DYNAMIC_SERVICES.DYNAMIC_SIDECAR
         )

services/director-v2/tests/conftest.py

@@ -9,6 +9,7 @@
 from copy import deepcopy
 from pathlib import Path
 from typing import Any, AsyncIterable, Iterable
+from unittest.mock import AsyncMock
 
 import httpx
 import pytest
@@ -238,3 +239,13 @@ def fake_workbench_complete_adjacency(
     fake_workbench_complete_adjacency_file: Path,
 ) -> dict[str, Any]:
     return json.loads(fake_workbench_complete_adjacency_file.read_text())
+
+
+@pytest.fixture
+def disable_rabbitmq(mocker) -> None:
+    def mock_setup(app: FastAPI) -> None:
+        app.state.rabbitmq_client = AsyncMock()
+
+    mocker.patch(
+        "simcore_service_director_v2.modules.rabbitmq.setup", side_effect=mock_setup
+    )

services/director-v2/tests/unit/test_api_route_dynamic_scheduler.py

@@ -26,7 +26,10 @@
 
 @pytest.fixture
 def mock_env(
-    mock_env: EnvVarsDict, monkeypatch: MonkeyPatch, docker_swarm: None
+    disable_rabbitmq: None,
+    mock_env: EnvVarsDict,
+    monkeypatch: MonkeyPatch,
+    docker_swarm: None,
 ) -> None:
     monkeypatch.setenv("SC_BOOT_MODE", "default")
     monkeypatch.setenv("DIRECTOR_ENABLED", "false")

services/director-v2/tests/unit/test_modules_dynamic_sidecar_scheduler.py

@@ -108,6 +108,7 @@ async def _assert_get_dynamic_services_mocked(
 
 @pytest.fixture
 def mock_env(
+    disable_rabbitmq: None,
     mock_env: EnvVarsDict,
     monkeypatch: MonkeyPatch,
     simcore_services_network_name: str,

services/director-v2/tests/unit/test_modules_dynamic_sidecar_scheduler_task.py

@@ -36,6 +36,7 @@
 
 @pytest.fixture
 def mock_env(
+    disable_rabbitmq: None,
     mock_env: EnvVarsDict,
     monkeypatch: MonkeyPatch,
     simcore_services_network_name: str,