✨activate tracing in all fastapi apps (ITISFoundation#2591)

Author: sanderegg

services/api-server/src/simcore_service_api_server/core/application.py

@@ -4,6 +4,7 @@
 from fastapi import FastAPI
 from fastapi.exceptions import RequestValidationError
 from httpx import HTTPStatusError
+from servicelib.fastapi.tracing import setup_tracing
 from servicelib.logging_utils import config_all_loggers
 from starlette import status
 from starlette.exceptions import HTTPException
@@ -65,6 +66,9 @@ def init_app(settings: Optional[AppSettings] = None) -> FastAPI:
     if settings.API_SERVER_DIRECTOR_V2:
         director_v2.setup(app, settings.API_SERVER_DIRECTOR_V2)
 
+    if settings.API_SERVER_TRACING:
+        setup_tracing(app, settings.API_SERVER_TRACING)
+
     # setup app
     app.add_event_handler("startup", create_start_app_handler(app))
     app.add_event_handler("shutdown", create_stop_app_handler(app))

services/api-server/src/simcore_service_api_server/core/settings.py

@@ -8,6 +8,7 @@
 from settings_library.base import BaseCustomSettings
 from settings_library.logging_utils import MixinLoggingSettings
 from settings_library.postgres import PostgresSettings
+from settings_library.tracing import TracingSettings
 
 # SERVICES CLIENTS --------------------------------------------
 
@@ -91,6 +92,7 @@ class AppSettings(BaseCustomSettings, MixinLoggingSettings):
     API_SERVER_CATALOG: Optional[CatalogSettings]
     API_SERVER_STORAGE: Optional[StorageSettings]
     API_SERVER_DIRECTOR_V2: Optional[DirectorV2Settings]
+    API_SERVER_TRACING: Optional[TracingSettings]
 
     API_SERVER_DEV_FEATURES_ENABLED: bool = Field(
         False, env=["API_SERVER_DEV_FEATURES_ENABLED", "FAKE_API_SERVER_ENABLED"]

services/datcore-adapter/src/simcore_service_datcore_adapter/core/application.py

@@ -3,6 +3,7 @@
 
 from fastapi import FastAPI
 from models_library.basic_types import BootModeEnum
+from servicelib.fastapi.tracing import setup_tracing
 
 from ..api.module_setup import setup_api
 from ..meta import api_version, api_vtag
@@ -53,4 +54,7 @@ def create_app(settings: Optional[Settings] = None) -> FastAPI:
     if settings.PENNSIEVE.PENNSIEVE_ENABLED:
         pennsieve.setup(app, settings.PENNSIEVE)
 
+    if settings.DATCORE_ADAPTER_TRACING:
+        setup_tracing(app, settings.DATCORE_ADAPTER_TRACING)
+
     return app

services/datcore-adapter/src/simcore_service_datcore_adapter/utils/requests_decorators.py

@@ -15,13 +15,21 @@
 async def _cancel_task_if_client_disconnected(
     request: Request, task: asyncio.Task, interval: float = _DEFAULT_CHECK_INTERVAL_S
 ) -> None:
-    with suppress(CancelledError):
+    try:
         while True:
+            if task.done():
+                logger.debug("task %s is done", task)
+                break
             if await request.is_disconnected():
                 logger.warning("client %s disconnected!", request.client)
                 task.cancel()
                 break
             await asyncio.sleep(interval)
+    except CancelledError:
+        logger.debug("task was cancelled")
+        raise
+    finally:
+        logger.debug("task completed")
 
 
 def cancellable_request(handler: Callable[..., Coroutine[Any, Any, Optional[Any]]]):
@@ -56,5 +64,7 @@ async def decorator(*args, **kwargs) -> Optional[Any]:
             return Response("Oh No!", status_code=499)
         finally:
             auto_cancel_task.cancel()
+            with suppress(CancelledError):
+                await auto_cancel_task
 
     return decorator

services/datcore-adapter/tests/unit/conftest.py

@@ -13,6 +13,7 @@
 import pytest
 import respx
 import simcore_service_datcore_adapter
+from _pytest.monkeypatch import MonkeyPatch
 from asgi_lifespan import LifespanManager
 from fastapi.applications import FastAPI
 from simcore_service_datcore_adapter.modules.pennsieve import _create_pennsieve_client
@@ -68,8 +69,14 @@ def client(minimal_app: FastAPI) -> TestClient:
         return cli
 
 
+@pytest.fixture
+def app_envs(monkeypatch: MonkeyPatch):
+    # disable tracing as together with LifespanManager, it does not remove itself nicely
+    monkeypatch.setenv("DATCORE_ADAPTER_TRACING", "null")
+
+
 @pytest.fixture()
-async def initialized_app(minimal_app: FastAPI) -> Iterator[FastAPI]:
+async def initialized_app(app_envs: None, minimal_app: FastAPI) -> Iterator[FastAPI]:
     async with LifespanManager(minimal_app):
         yield minimal_app
 

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

@@ -4,6 +4,7 @@
 from fastapi import FastAPI
 from fastapi.exceptions import RequestValidationError
 from servicelib.fastapi.openapi import override_fastapi_openapi_method
+from servicelib.fastapi.tracing import setup_tracing
 from simcore_service_director_v2.modules import dask_client
 from starlette import status
 from starlette.exceptions import HTTPException
@@ -89,6 +90,9 @@ def init_app(settings: Optional[AppSettings] = None) -> FastAPI:
     ):
         comp_scheduler.setup(app)
 
+    if settings.DIRECTOR_V2_TRACING:
+        setup_tracing(app, settings.DIRECTOR_V2_TRACING)
+
     # setup app --
     app.add_event_handler("startup", on_startup)
     app.add_event_handler("shutdown", on_shutdown)

services/director-v2/src/simcore_service_director_v2/modules/dask_client.py

@@ -3,6 +3,8 @@
 import contextlib
 import functools
 import logging
+import os
+import socket
 from dataclasses import dataclass, field
 from pprint import pformat
 from typing import Any, Awaitable, Callable, Dict, Iterable, List, Tuple, Union
@@ -99,7 +101,7 @@ async def create(
             client=await dask.distributed.Client(
                 f"tcp://{settings.DASK_SCHEDULER_HOST}:{settings.DASK_SCHEDULER_PORT}",
                 asynchronous=True,
-                name="director-v2-client",
+                name=f"director-v2-client_{socket.gethostname()}_{os.getpid()}",
             ),  # type: ignore
             settings=settings,
         )
@@ -131,7 +133,7 @@ async def delete(self) -> None:
     async def reconnect_client(self):
         if self.client:
             await self.client.close()  # type: ignore
-        self.client = await dask.distributed.Client(
+        self.client = await distributed.Client(
             f"tcp://{self.settings.DASK_SCHEDULER_HOST}:{self.settings.DASK_SCHEDULER_PORT}",
             asynchronous=True,
             name="director-v2-client",
@@ -149,10 +151,9 @@ def register_handlers(
         async def _dask_sub_handler(
             dask_sub_topic_name: str, handler: Callable[[str], Awaitable[None]]
         ):
-            dask_sub = distributed.Sub(dask_sub_topic_name)
-
             while True:
                 try:
+                    dask_sub = distributed.Sub(dask_sub_topic_name)
                     async for event in dask_sub:
                         logger.debug("received event %s", event)
                         await handler(event)

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

@@ -343,9 +343,10 @@ async def _run_scheduler_task(self) -> None:
 
                 await sleep(settings.DIRECTOR_V2_DYNAMIC_SCHEDULER_INTERVAL_SECONDS)
             except asyncio.CancelledError:  # pragma: no cover
-                break  # pragma: no cover
-
-        logger.warning("Scheduler was shut down")
+                logger.info("Stopped dynamic scheduler")
+                raise
+            except Exception:  # pylint: disable=broad-except
+                logger.error("Unexpected error in dynamic scheduler", exc_info=True)
 
     async def _discover_running_services(self) -> None:
         """discover all services which were started before and add them to the scheduler"""
@@ -395,6 +396,7 @@ async def shutdown(self):
 
         if self._trigger_observation_queue_task is not None:
             await self._trigger_observation_queue.put(None)
+
             self._trigger_observation_queue_task.cancel()
             with contextlib.suppress(asyncio.CancelledError):
                 await self._trigger_observation_queue_task
@@ -405,7 +407,6 @@ async def shutdown(self):
 async def setup_scheduler(app: FastAPI):
     dynamic_sidecars_scheduler = DynamicSidecarsScheduler(app)
     app.state.dynamic_sidecar_scheduler = dynamic_sidecars_scheduler
-
     settings: DynamicServicesSchedulerSettings = (
         app.state.settings.DYNAMIC_SERVICES.DYNAMIC_SCHEDULER
     )

services/director-v2/tests/conftest.py

@@ -131,6 +131,9 @@ def mock_env(monkeypatch: MonkeyPatch) -> None:
 
     monkeypatch.setenv("SC_BOOT_MODE", "production")
 
+    # disable tracing as together with LifespanManager, it does not remove itself nicely
+    monkeypatch.setenv("DIRECTOR_V2_TRACING", "null")
+
 
 @pytest.fixture(scope="function")
 def client(loop: asyncio.AbstractEventLoop, mock_env: None) -> Iterable[TestClient]:

services/director-v2/tests/integration/01/test_computation_api.py

@@ -69,6 +69,7 @@ def mock_env(monkeypatch: MonkeyPatch, request) -> None:
         "DIRECTOR_V2_CELERY_SCHEDULER_ENABLED",
         "1" if request.param == "celery" else "0",
     )
+    monkeypatch.setenv("DIRECTOR_V2_TRACING", "null")
     monkeypatch.setenv("SIMCORE_SERVICES_NETWORK_NAME", "test_swarm_network_name")
     monkeypatch.setenv("TRAEFIK_SIMCORE_ZONE", "test_mocked_simcore_zone")
     monkeypatch.setenv("SWARM_STACK_NAME", "test_mocked_stack_name")