🔨♻️Profiling tools for catalog

packages/service-library/src/servicelib/async_utils.py

@@ -6,7 +6,7 @@
 from functools import wraps
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Deque
 
-from .utils_profiling_middleware import dont_profile, is_profiling, profile
+from .utils_profiling_middleware import dont_profile, is_profiling, profile_context
 
 logger = logging.getLogger(__name__)
 
@@ -167,7 +167,7 @@ async def worker(in_q: Queue[QueueElement], out_q: Queue) -> None:
                             awaitable = element.input
                             if awaitable is None:
                                 break
-                            with profile(do_profile):
+                            with profile_context(do_profile):
                                 result = await awaitable
                         except Exception as e:  # pylint: disable=broad-except
                             result = e

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/catalog/services.py

@@ -17,6 +17,7 @@
 from models_library.users import UserID
 from pydantic import NonNegativeInt, parse_obj_as, validate_arguments
 from servicelib.logging_utils import log_decorator
+from servicelib.rabbitmq._constants import RPC_REQUEST_DEFAULT_TIMEOUT_S
 
 from ..._client_rpc import RabbitMQRPCClient
 
@@ -52,6 +53,7 @@ async def _call(
             user_id=user_id,
             limit=limit,
             offset=offset,
+            timeout_s=2 * RPC_REQUEST_DEFAULT_TIMEOUT_S,
         )
 
     result = await _call(
@@ -91,6 +93,7 @@ async def _call(
             user_id=user_id,
             service_key=service_key,
             service_version=service_version,
+            timeout_s=2 * RPC_REQUEST_DEFAULT_TIMEOUT_S,
         )
 
     result = await _call(

packages/service-library/src/servicelib/utils_profiling_middleware.py

@@ -1,13 +1,15 @@
 import contextvars
 import json
+from collections.abc import Iterator
 from contextlib import contextmanager
-from typing import Iterator
+from typing import Final
 
+from models_library.utils.json_serialization import json_dumps, json_loads
 from pyinstrument import Profiler
-from servicelib.mimetype_constants import (
-    MIMETYPE_APPLICATION_JSON,
-    MIMETYPE_APPLICATION_ND_JSON,
-)
+
+from .mimetype_constants import MIMETYPE_APPLICATION_JSON, MIMETYPE_APPLICATION_ND_JSON
+
+_UNSET: Final = None
 
 _profiler = Profiler(async_mode="enabled")
 _is_profiling = contextvars.ContextVar("_is_profiling", default=False)
@@ -18,11 +20,11 @@ def is_profiling() -> bool:
 
 
 @contextmanager
-def profile(do_profile: bool | None = None) -> Iterator[None]:
+def profile_context(enable: bool | None = _UNSET) -> Iterator[None]:
     """Context manager which temporarily removes request profiler from context"""
-    if do_profile is None:
-        do_profile = _is_profiling.get()
-    if do_profile:
+    if enable is _UNSET:
+        enable = _is_profiling.get()
+    if enable:
         try:
             _profiler.start()
             yield
@@ -46,22 +48,22 @@ def dont_profile() -> Iterator[None]:
 
 def append_profile(body: str, profile_text: str) -> str:
     try:
-        json.loads(body)
+        json_loads(body)
         body += "\n" if not body.endswith("\n") else ""
     except json.decoder.JSONDecodeError:
         pass
-    body += json.dumps({"profile": profile_text})
+    body += json_dumps({"profile": profile_text})
     return body
 
 
 def check_response_headers(
     response_headers: dict[bytes, bytes]
 ) -> list[tuple[bytes, bytes]]:
     original_content_type: str = response_headers[b"content-type"].decode()
-    assert original_content_type in {
+    assert original_content_type in {  # nosec
         MIMETYPE_APPLICATION_ND_JSON,
         MIMETYPE_APPLICATION_JSON,
-    }  # nosec
+    }
     headers: dict = {}
     headers[b"content-type"] = MIMETYPE_APPLICATION_ND_JSON.encode()
     return list(headers.items())

services/catalog/src/simcore_service_catalog/api/rpc/_services.py

@@ -1,3 +1,4 @@
+import functools
 import logging
 
 from fastapi import FastAPI
@@ -11,6 +12,7 @@
 from models_library.services_types import ServiceKey, ServiceVersion
 from models_library.users import UserID
 from pydantic import NonNegativeInt
+from pyinstrument import Profiler
 from servicelib.logging_utils import log_decorator
 from servicelib.rabbitmq import RPCRouter
 from servicelib.rabbitmq.rpc_interfaces.catalog.errors import (
@@ -27,8 +29,30 @@
 router = RPCRouter()
 
 
+def _profile_rpc_call(coro):
+    @functools.wraps(coro)
+    async def _wrapper(app: FastAPI, **kwargs):
+        profile_enabled = (
+            (settings := getattr(app.state, "settings", None))
+            and settings.CATALOG_PROFILING
+            and _logger.isEnabledFor(logging.INFO)
+        )
+        if profile_enabled:
+            with Profiler() as profiler:
+                result = await coro(app, **kwargs)
+            profiler_output = profiler.output_text(unicode=True, color=False)
+            _logger.info("[PROFILING]: %s", profiler_output)
+            return result
+
+        # bypasses w/o profiling
+        return await coro(app, **kwargs)
+
+    return _wrapper
+
+
 @router.expose(reraise_if_error_type=(CatalogForbiddenError,))
 @log_decorator(_logger, level=logging.DEBUG)
+@_profile_rpc_call
 async def list_services_paginated(
     app: FastAPI,
     *,
@@ -61,6 +85,7 @@ async def list_services_paginated(
 
 @router.expose(reraise_if_error_type=(CatalogItemNotFoundError, CatalogForbiddenError))
 @log_decorator(_logger, level=logging.DEBUG)
+@_profile_rpc_call
 async def get_service(
     app: FastAPI,
     *,

services/catalog/tests/unit/test_db_repositories_services_sql.py

@@ -24,13 +24,16 @@ def _check(func_smt, **kwargs):
 
     # some data
     product_name = "osparc"
-    user_id = 4  # 425 (san)  # 4 (odei)
+    user_id = 425  # 425 (guidon)  # 4 (odei)
     service_key = "simcore/services/comp/isolve"
     service_version = "2.0.85"
 
     service_key = "simcore/services/dynamic/raw-graphs"
     service_version = "2.11.2"
 
+    service_key = "simcore/services/dynamic/s4l-core-8-0-0-dy"
+    service_version = "3.2.39"
+
     _check(
         get_service_history_stmt,
         product_name=product_name,
@@ -62,8 +65,8 @@ def _check(func_smt, **kwargs):
         product_name=product_name,
         user_id=user_id,
         access_rights=AccessRightsClauses.can_read,
-        limit=100,
-        offset=None,
+        limit=15,
+        offset=80,
     )
 
     _check(

services/docker-compose.yml

@@ -136,7 +136,7 @@ services:
   catalog:
     image: ${DOCKER_REGISTRY:-itisfoundation}/catalog:${DOCKER_IMAGE_TAG:-latest}
     init: true
-    hostname: "{{.Node.Hostname}}-{{.Task.Slot}}"
+    hostname: "cat-{{.Node.Hostname}}-{{.Task.Slot}}"
     environment:
       CATALOG_BACKGROUND_TASK_REST_TIME: ${CATALOG_BACKGROUND_TASK_REST_TIME}
       CATALOG_DEV_FEATURES_ENABLED: ${CATALOG_DEV_FEATURES_ENABLED}
@@ -390,7 +390,7 @@ services:
   invitations:
     image: ${DOCKER_REGISTRY:-itisfoundation}/invitations:${DOCKER_IMAGE_TAG:-latest}
     init: true
-    hostname: "{{.Node.Hostname}}-{{.Task.Slot}}"
+    hostname: "inv-{{.Node.Hostname}}-{{.Task.Slot}}"
     networks:
       - default
     environment:
@@ -406,7 +406,7 @@ services:
   payments:
     image: ${DOCKER_REGISTRY:-itisfoundation}/payments:${DOCKER_IMAGE_TAG:-latest}
     init: true
-    hostname: "{{.Node.Hostname}}-{{.Task.Slot}}"
+    hostname: "pay-{{.Node.Hostname}}-{{.Task.Slot}}"
     networks:
       - default
     environment:
@@ -764,7 +764,6 @@ services:
 
     networks: *webserver_networks
 
-
   wb-db-event-listener:
     image: ${DOCKER_REGISTRY:-itisfoundation}/webserver:${DOCKER_IMAGE_TAG:-latest}
     init: true
@@ -1007,7 +1006,7 @@ services:
   storage:
     image: ${DOCKER_REGISTRY:-itisfoundation}/storage:${DOCKER_IMAGE_TAG:-latest}
     init: true
-    hostname: "{{.Node.Hostname}}-{{.Task.Slot}}"
+    hostname: "sto-{{.Node.Hostname}}-{{.Task.Slot}}"
     environment:
       BF_API_KEY: ${BF_API_KEY}
       BF_API_SECRET: ${BF_API_SECRET}
@@ -1216,12 +1215,6 @@ services:
     networks:
       - default
       - interactive_services_subnet # for legacy dynamic services
-    #healthcheck:
-    #  test: wget --quiet --tries=1 --spider http://localhost:9082/ping || exit 1
-    #  interval: 3s
-    #  timeout: 1s
-    #  retries: 3
-    #  start_period: 20s
 
 
 volumes: