✨Adds metrics regarding new style dynamic services timings

packages/service-library/src/servicelib/fastapi/docker_utils.py

@@ -108,7 +108,7 @@ async def pull_images(
     registry_settings: RegistrySettings,
     progress_cb: AsyncReportCB,
     log_cb: LogCB,
-) -> None:
+) -> int:
     images_layer_information = await asyncio.gather(
         *[
             retrieve_image_layer_information(image, registry_settings)
@@ -147,3 +147,4 @@ async def pull_images(
                 )
             ]
         )
+    return images_total_size

services/director-v2/src/simcore_service_director_v2/api/routes/dynamic_services.py

@@ -136,7 +136,6 @@ async def create_dynamic_service(
         logger.debug("Redirecting %s", redirect_url_with_query)
         return RedirectResponse(str(redirect_url_with_query))
 
-    #
     if not await is_sidecar_running(
         service.node_uuid, dynamic_services_settings.DYNAMIC_SCHEDULER.SWARM_STACK_NAME
     ):

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

@@ -7,9 +7,6 @@
     override_fastapi_openapi_method,
 )
 from servicelib.fastapi.profiler_middleware import ProfilerMiddleware
-from servicelib.fastapi.prometheus_instrumentation import (
-    setup_prometheus_instrumentation,
-)
 from servicelib.logging_utils import config_all_loggers
 
 from ..api.entrypoints import api_router
@@ -27,6 +24,7 @@
     director_v0,
     dynamic_services,
     dynamic_sidecar,
+    instrumentation,
     notifier,
     rabbitmq,
     redis,
@@ -190,8 +188,7 @@ def init_app(settings: AppSettings | None = None) -> FastAPI:
 
     resource_usage_tracker_client.setup(app)
 
-    if settings.DIRECTOR_V2_PROMETHEUS_INSTRUMENTATION_ENABLED:
-        setup_prometheus_instrumentation(app)
+    instrumentation.setup(app)
 
     if settings.DIRECTOR_V2_PROFILING:
         app.add_middleware(ProfilerMiddleware)

services/director-v2/src/simcore_service_director_v2/models/dynamic_services_scheduler.py

@@ -2,12 +2,14 @@
 import logging
 import re
 from collections.abc import Mapping
+from datetime import datetime
 from enum import Enum
 from functools import cached_property
 from pathlib import Path
 from typing import Any, TypeAlias
 from uuid import UUID
 
+import arrow
 from models_library.api_schemas_directorv2.dynamic_services import DynamicServiceCreate
 from models_library.api_schemas_directorv2.dynamic_services_service import (
     CommonServiceDetails,
@@ -170,6 +172,32 @@ def mark_removed(self) -> None:
         self.was_removed = True
 
 
+class MetricsTimers(BaseModel):
+    start_duration_of_service_starting: datetime | None = Field(
+        None,
+        description="moment in which the process of starting the service was requested",
+    )
+    start_duration_of_service_closing: datetime | None = Field(
+        None,
+        description="moment in which the process of stopping the service was requested",
+    )
+
+    def get_start_duration(self) -> float | None:
+        if self.start_duration_of_service_starting is None:
+            return None
+
+        return (
+            arrow.utcnow().datetime - self.start_duration_of_service_starting
+        ).total_seconds()
+
+    def get_stop_duration(self) -> float | None:
+        if self.start_duration_of_service_closing is None:
+            return None
+        return (
+            arrow.utcnow().datetime - self.start_duration_of_service_closing
+        ).total_seconds()
+
+
 class DynamicSidecar(BaseModel):
     status: Status = Field(
         Status.create_as_initially_ok(),
@@ -254,6 +282,11 @@ def compose_spec_submitted(self) -> bool:
         description="set True if the dy-sidecar saves the state and uploads the outputs",
     )
 
+    metrics_timers: MetricsTimers = Field(
+        default_factory=lambda: MetricsTimers.parse_obj({}),
+        description="keeps track of start times for various operations",
+    )
+
     # below had already been validated and
     # used only to start the proxy
     dynamic_sidecar_id: ServiceId | None = Field(

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/api_client/_public.py

@@ -2,7 +2,7 @@
 from collections import deque
 from collections.abc import Coroutine
 from functools import cached_property
-from typing import Any, Final
+from typing import Any, Final, cast
 
 from fastapi import FastAPI, status
 from httpx import AsyncClient
@@ -349,33 +349,35 @@ async def stop_service(
             progress_callback,
         )
 
-    async def restore_service_state(self, dynamic_sidecar_endpoint: AnyHttpUrl) -> None:
+    async def restore_service_state(self, dynamic_sidecar_endpoint: AnyHttpUrl) -> int:
         response = await self._thin_client.post_containers_tasks_state_restore(
             dynamic_sidecar_endpoint
         )
         task_id: TaskId = response.json()
 
-        await self._await_for_result(
+        result: Any | None = await self._await_for_result(
             task_id,
             dynamic_sidecar_endpoint,
             self._dynamic_services_scheduler_settings.DYNAMIC_SIDECAR_API_SAVE_RESTORE_STATE_TIMEOUT,
             _debug_progress_callback,
         )
+        return cast(int, result)
 
     async def pull_user_services_images(
         self, dynamic_sidecar_endpoint: AnyHttpUrl
-    ) -> None:
+    ) -> int:
         response = await self._thin_client.post_containers_images_pull(
             dynamic_sidecar_endpoint
         )
         task_id: TaskId = response.json()
 
-        await self._await_for_result(
+        result: Any | None = await self._await_for_result(
             task_id,
             dynamic_sidecar_endpoint,
             self._dynamic_services_scheduler_settings.DYNAMIC_SIDECAR_API_USER_SERVICES_PULLING_TIMEOUT,
             _debug_progress_callback,
         )
+        return cast(int, result)
 
     async def save_service_state(
         self,
@@ -416,18 +418,19 @@ async def pull_service_output_ports(
         self,
         dynamic_sidecar_endpoint: AnyHttpUrl,
         port_keys: list[str] | None = None,
-    ) -> None:
+    ) -> int:
         response = await self._thin_client.post_containers_tasks_ports_outputs_pull(
             dynamic_sidecar_endpoint, port_keys
         )
         task_id: TaskId = response.json()
 
-        await self._await_for_result(
+        result: Any | None = await self._await_for_result(
             task_id,
             dynamic_sidecar_endpoint,
             self._dynamic_services_scheduler_settings.DYNAMIC_SIDECAR_API_SAVE_RESTORE_STATE_TIMEOUT,
             _debug_progress_callback,
         )
+        return cast(int, result)
 
     async def push_service_output_ports(
         self,

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

@@ -19,6 +19,8 @@
     DynamicServicesSchedulerSettings,
 )
 from .....models.dynamic_services_scheduler import SchedulerData
+from .....modules.instrumentation._setup import get_instrumentation
+from .....modules.instrumentation._utils import get_start_stop_labels
 from .....utils.db import get_repository
 from ....db.repositories.groups_extra_properties import GroupsExtraPropertiesRepository
 from ....db.repositories.projects import ProjectsRepository
@@ -222,4 +224,10 @@ async def progress_create_containers(
 
     scheduler_data.dynamic_sidecar.were_containers_created = True
 
+    start_duration = scheduler_data.dynamic_sidecar.metrics_timers.get_start_duration()
+    if start_duration:
+        get_instrumentation(app).dynamic_sidecar_metrics.start_time_seconds.labels(
+            **get_start_stop_labels(scheduler_data)
+        ).observe(start_duration)
+
     _logger.info("Internal state after creating user services %s", scheduler_data)

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

@@ -44,6 +44,12 @@
     DockerStatus,
     SchedulerData,
 )
+from .....modules.instrumentation import (
+    get_instrumentation,
+    get_label_from_size,
+    get_start_stop_labels,
+    track_duration,
+)
 from .....utils.db import get_repository
 from ....db.repositories.projects import ProjectsRepository
 from ....db.repositories.projects_networks import ProjectsNetworksRepository
@@ -375,6 +381,14 @@ async def attempt_pod_removal_and_data_saving(
     rabbitmq_client: RabbitMQClient = app.state.rabbitmq_client
     await rabbitmq_client.publish(message.channel_name, message)
 
+    # metrics
+
+    stop_duration = scheduler_data.dynamic_sidecar.metrics_timers.get_stop_duration()
+    if stop_duration:
+        get_instrumentation(app).dynamic_sidecar_metrics.stop_time_seconds.labels(
+            **get_start_stop_labels(scheduler_data)
+        ).observe(stop_duration)
+
 
 async def attach_project_networks(app: FastAPI, scheduler_data: SchedulerData) -> None:
     _logger.debug("Attaching project networks for %s", scheduler_data.service_name)
@@ -460,14 +474,54 @@ async def prepare_services_environment(
         )
     )
 
+    async def _pull_output_ports_with_metrics() -> None:
+        with track_duration() as duration:
+            size: int = await sidecars_client.pull_service_output_ports(
+                dynamic_sidecar_endpoint
+            )
+
+        get_instrumentation(
+            app
+        ).dynamic_sidecar_metrics.output_ports_pull_seconds.labels(
+            **get_label_from_size(size)
+        ).observe(
+            duration.to_flaot()
+        )
+
+    async def _pull_user_services_images_with_metrics() -> None:
+        with track_duration() as duration:
+            size = await sidecars_client.pull_user_services_images(
+                dynamic_sidecar_endpoint
+            )
+
+        get_instrumentation(
+            app
+        ).dynamic_sidecar_metrics.pull_user_services_images_seconds.labels(
+            **get_label_from_size(size)
+        ).observe(
+            duration.to_flaot()
+        )
+
+    async def _restore_service_state_with_metrics() -> None:
+        with track_duration() as duration:
+            size = await sidecars_client.restore_service_state(dynamic_sidecar_endpoint)
+
+        get_instrumentation(
+            app
+        ).dynamic_sidecar_metrics.restore_service_state_seconds.labels(
+            **get_label_from_size(size)
+        ).observe(
+            duration.to_flaot()
+        )
+
     tasks = [
-        sidecars_client.pull_user_services_images(dynamic_sidecar_endpoint),
-        sidecars_client.pull_service_output_ports(dynamic_sidecar_endpoint),
+        _pull_user_services_images_with_metrics(),
+        _pull_output_ports_with_metrics(),
     ]
     # When enabled no longer downloads state via nodeports
     # S3 is used to store state paths
     if not app_settings.DIRECTOR_V2_DEV_FEATURE_R_CLONE_MOUNTS_ENABLED:
-        tasks.append(sidecars_client.restore_service_state(dynamic_sidecar_endpoint))
+        tasks.append(_restore_service_state_with_metrics())
 
     await limited_gather(*tasks, limit=3)
 

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

@@ -17,10 +17,12 @@
 import contextlib
 import functools
 import logging
+import time
 from asyncio import Lock, Queue, Task
 from dataclasses import dataclass, field
 from typing import Final
 
+import arrow
 from fastapi import FastAPI
 from models_library.api_schemas_directorv2.dynamic_services import (
     DynamicServiceCreate,
@@ -54,6 +56,7 @@
     DynamicServicesSchedulerSettings,
 )
 from .....models.dynamic_services_scheduler import SchedulerData, ServiceName
+from .....modules.instrumentation import get_instrumentation, get_label_from_size
 from ...api_client import SidecarsClient, get_sidecars_client
 from ...docker_api import update_scheduler_data_label
 from ...errors import DynamicSidecarError, DynamicSidecarNotFoundError
@@ -255,6 +258,9 @@ async def add_service(
             request_simcore_user_agent=request_simcore_user_agent,
             can_save=can_save,
         )
+        scheduler_data.dynamic_sidecar.metrics_timers.start_duration_of_service_starting = (
+            arrow.utcnow().datetime
+        )
         await self.add_service_from_scheduler_data(scheduler_data)
 
     async def add_service_from_scheduler_data(
@@ -353,6 +359,10 @@ async def mark_service_for_removal(
                 )
                 return
 
+            current.dynamic_sidecar.metrics_timers.start_duration_of_service_closing = (
+                arrow.utcnow().datetime
+            )
+
             # PC-> ANE: could you please review what to do when can_save=None
             assert can_save is not None  # nosec
             current.dynamic_sidecar.service_removal_state.mark_to_remove(
@@ -455,9 +465,19 @@ async def retrieve_service_inputs(
         dynamic_sidecar_endpoint: AnyHttpUrl = scheduler_data.endpoint
         sidecars_client: SidecarsClient = await get_sidecars_client(self.app, node_uuid)
 
+        started = time.time()
         transferred_bytes = await sidecars_client.pull_service_input_ports(
             dynamic_sidecar_endpoint, port_keys
         )
+        duration = time.time() - started
+
+        get_instrumentation(
+            self.app
+        ).dynamic_sidecar_metrics.input_ports_pull_seconds.labels(
+            **get_label_from_size(transferred_bytes)
+        ).observe(
+            duration
+        )
 
         if scheduler_data.restart_policy == RestartPolicy.ON_INPUTS_DOWNLOADED:
             logger.info("Will restart containers")

services/director-v2/src/simcore_service_director_v2/modules/instrumentation/__init__.py

@@ -0,0 +1,10 @@
+from ._setup import get_instrumentation, setup
+from ._utils import get_label_from_size, get_start_stop_labels, track_duration
+
+__all__: tuple[str, ...] = (
+    "get_instrumentation",
+    "get_label_from_size",
+    "get_start_stop_labels",
+    "setup",
+    "track_duration",
+)