🎨♻️Computational backend: stability improvements step 4

packages/common-library/src/common_library/async_tools.py

@@ -2,12 +2,15 @@
 import datetime
 import functools
 import logging
+import sys
 from collections.abc import Awaitable, Callable, Coroutine
 from concurrent.futures import Executor
 from functools import wraps
 from inspect import isawaitable
 from typing import Any, ParamSpec, TypeVar, overload
 
+from .logging.logging_errors import create_troubleshooting_log_kwargs
+
 _logger = logging.getLogger(__name__)
 
 R = TypeVar("R")
@@ -102,11 +105,13 @@ async def cancel_wait_task(
             # from observing the cancellation/finalization of task.
             asyncio.wait_for(task, timeout=max_delay)
         )
-    except TimeoutError:
+    except TimeoutError as exc:
         _logger.exception(
-            "Timeout while cancelling task %s after %s seconds",
-            task.get_name(),
-            max_delay,
+            **create_troubleshooting_log_kwargs(
+                f"Timeout while cancelling task {task.get_name()} after {max_delay} seconds",
+                error=exc,
+                error_context={"task_name": task.get_name(), "max_delay": max_delay},
+            )
         )
         raise
     except asyncio.CancelledError:
@@ -117,7 +122,20 @@ async def cancel_wait_task(
             raise
     finally:
         if not task.done():
-            _logger.error("Failed to cancel %s", task.get_name())
+            current_exception = sys.exception()
+            _logger.error(
+                **create_troubleshooting_log_kwargs(
+                    f"Failed to cancel ask {task.get_name()}",
+                    error=(
+                        current_exception if current_exception else Exception("Unknown")
+                    ),
+                    error_context={
+                        "task_name": task.get_name(),
+                        "max_delay": max_delay,
+                    },
+                    tip="Consider increasing max_delay or fixing the task to handle cancellations properly",
+                )
+            )
         else:
             _logger.debug("Task %s cancelled", task.get_name())
 

packages/service-library/src/servicelib/logging_errors.py → packages/common-library/src/common_library/logging/logging_errors.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, TypedDict
+from typing import Any, Final, TypedDict
 
 from common_library.error_codes import ErrorCodeStr
 from common_library.errors_classes import OsparcErrorMixin
@@ -9,6 +9,8 @@
 
 _logger = logging.getLogger(__name__)
 
+_MAX_LOGGED_CAUSES: Final[int] = 10
+
 
 def create_troubleshooting_log_message(
     user_error_msg: str,
@@ -36,7 +38,7 @@ def _collect_causes(exc: BaseException) -> str:
             seen.add(id(current))
             causes.append(f"[{type(current).__name__}]'{current}'")
             current = getattr(current, "__cause__", None)
-            if len(causes) > 10:  # Prevent excessive chains
+            if len(causes) > _MAX_LOGGED_CAUSES:  # Prevent excessive chains
                 causes.append("[... truncated]")
                 break
         return " <- ".join(causes)

packages/service-library/tests/test_logging_errors.py → packages/common-library/tests/test_logging_errors.py

@@ -5,7 +5,7 @@
 import pytest
 from common_library.error_codes import create_error_code, parse_error_code_parts
 from common_library.errors_classes import OsparcErrorMixin
-from servicelib.logging_errors import (
+from common_library.logging.logging_errors import (
     create_troubleshooting_log_kwargs,
     create_troubleshooting_log_message,
 )

packages/service-library/tests/test_logging_utils_filtering.py → packages/common-library/tests/test_logging_utils_filtering.py

@@ -1,10 +1,10 @@
 # pylint: disable=redefined-outer-name
 
 import logging
-from typing import Generator
+from collections.abc import Generator
 
 import pytest
-from servicelib.logging_utils_filtering import GeneralLogFilter
+from common_library.logging.logging_utils_filtering import GeneralLogFilter
 
 
 @pytest.fixture

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

@@ -13,11 +13,11 @@
 from aiohttp.web_response import StreamResponse
 from common_library.error_codes import ErrorCodeStr, create_error_code
 from common_library.json_serialization import json_dumps, json_loads
+from common_library.logging.logging_errors import create_troubleshooting_log_kwargs
 from common_library.user_messages import user_message
 from models_library.basic_types import IDStr
 from models_library.rest_error import ErrorGet, ErrorItemType, LogMessageType
 
-from ..logging_errors import create_troubleshooting_log_kwargs
 from ..mimetype_constants import MIMETYPE_APPLICATION_JSON
 from ..rest_constants import RESPONSE_MODEL_POLICY
 from ..rest_responses import is_enveloped_from_text
@@ -202,7 +202,6 @@ def _handle_exception_as_http_error(
 
 
 def error_middleware_factory(api_version: str) -> Middleware:
-
     @web.middleware
     async def _middleware_handler(request: web.Request, handler: Handler):
         """

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

@@ -5,10 +5,9 @@
 from functools import wraps
 from typing import Any, Final, ParamSpec, TypeVar
 
+from common_library.logging.logging_errors import create_troubleshooting_log_kwargs
 from pydantic import BaseModel, Field, NonNegativeFloat, PrivateAttr
 
-from .logging_errors import create_troubleshooting_log_kwargs
-
 _logger = logging.getLogger(__name__)
 
 _SKIPS_MESSAGE: Final[str] = "skip(s) of exception"

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

@@ -2,6 +2,7 @@
 from collections.abc import Awaitable, Callable
 from typing import TypeVar
 
+from common_library.logging.logging_errors import create_troubleshooting_log_kwargs
 from fastapi import FastAPI, HTTPException, status
 from fastapi.encoders import jsonable_encoder
 from fastapi.exceptions import RequestValidationError
@@ -11,7 +12,6 @@
 from fastapi.responses import JSONResponse
 from pydantic import ValidationError
 
-from ..logging_errors import create_troubleshooting_log_kwargs
 from ..status_codes_utils import is_5xx_server_error
 
 validation_error_response_definition["properties"] = {

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

@@ -2,6 +2,7 @@
 from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import AsyncExitStack
 
+from common_library.logging.logging_utils_filtering import LoggerName, MessageSubstring
 from fastapi import FastAPI
 from settings_library.tracing import TracingSettings
 
@@ -10,7 +11,6 @@
     async_loggers,
     log_context,
 )
-from ..logging_utils_filtering import LoggerName, MessageSubstring
 from .lifespan_utils import Lifespan
 
 _logger = logging.getLogger(__name__)

packages/service-library/src/servicelib/fastapi/long_running_tasks/_context_manager.py

@@ -5,9 +5,9 @@
 from contextlib import asynccontextmanager
 from typing import Any, Final
 
+from common_library.logging.logging_errors import create_troubleshooting_log_message
 from pydantic import PositiveFloat
 
-from ...logging_errors import create_troubleshooting_log_message
 from ...long_running_tasks.errors import TaskClientTimeoutError, TaskExceptionError
 from ...long_running_tasks.models import (
     ProgressCallback,

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

@@ -20,11 +20,15 @@
 from typing import Any, Final, TypeAlias, TypedDict, TypeVar
 
 from common_library.json_serialization import json_dumps
+from common_library.logging.logging_base import LogExtra
+from common_library.logging.logging_errors import create_troubleshooting_log_kwargs
+from common_library.logging.logging_utils_filtering import (
+    GeneralLogFilter,
+    LoggerName,
+    MessageSubstring,
+)
 from settings_library.tracing import TracingSettings
 
-from .logging_base import LogExtra
-from .logging_errors import create_troubleshooting_log_kwargs
-from .logging_utils_filtering import GeneralLogFilter, LoggerName, MessageSubstring
 from .tracing import setup_log_tracing
 from .utils_secrets import mask_sensitive_data
 
@@ -74,9 +78,13 @@ def __init__(self, fmt: str, *, log_format_local_dev_enabled: bool) -> None:
 
     def format(self, record) -> str:
         if hasattr(record, "func_name_override"):
-            record.funcName = record.func_name_override
+            record.funcName = (
+                record.func_name_override
+            )  # pyright: ignore[reportAttributeAccessIssue]
         if hasattr(record, "file_name_override"):
-            record.filename = record.file_name_override
+            record.filename = (
+                record.file_name_override
+            )  # pyright: ignore[reportAttributeAccessIssue]
 
         # pylint: disable=no-member
         optional_keys = LogExtra.__optional_keys__ | frozenset(

packages/service-library/src/servicelib/long_running_tasks/task.py

@@ -9,6 +9,7 @@
 from uuid import uuid4
 
 from common_library.async_tools import cancel_wait_task
+from common_library.logging.logging_errors import create_troubleshooting_log_kwargs
 from models_library.api_schemas_long_running_tasks.base import TaskProgress
 from pydantic import NonNegativeFloat, PositiveFloat
 from settings_library.redis import RedisDatabase, RedisSettings
@@ -20,7 +21,6 @@
 )
 
 from ..background_task import create_periodic_task
-from ..logging_errors import create_troubleshooting_log_kwargs
 from ..logging_utils import log_catch, log_context
 from ..redis import RedisClientSDK, exclusive
 from ..utils import limited_gather

packages/service-library/src/servicelib/rabbitmq/_rpc_router.py

@@ -7,8 +7,8 @@
 from typing import Any, TypeVar
 
 from common_library.error_codes import create_error_code
+from common_library.logging.logging_errors import create_troubleshooting_log_kwargs
 from models_library.rabbitmq_basic_types import RPCMethodName
-from servicelib.logging_errors import create_troubleshooting_log_kwargs
 
 from ..logging_utils import log_context
 from ._errors import RPCServerError
@@ -45,9 +45,7 @@ def expose(
         *,
         reraise_if_error_type: tuple[type[Exception], ...] | None = None,
     ) -> Callable[[DecoratedCallable], DecoratedCallable]:
-
         def _decorator(func: DecoratedCallable) -> DecoratedCallable:
-
             @functools.wraps(func)
             async def _wrapper(*args, **kwargs):
                 with log_context(

packages/service-library/src/servicelib/redis/_decorators.py

@@ -9,10 +9,10 @@
 import arrow
 import redis.exceptions
 from common_library.async_tools import cancel_wait_task
+from common_library.logging.logging_errors import create_troubleshooting_log_kwargs
 from redis.asyncio.lock import Lock
 
 from ..background_task import periodic
-from ..logging_errors import create_troubleshooting_log_kwargs
 from ._client import RedisClientSDK
 from ._constants import DEFAULT_LOCK_TTL
 from ._errors import CouldNotAcquireLockError, LockLostError

packages/service-library/src/servicelib/redis/_semaphore_decorator.py

@@ -8,9 +8,9 @@
 from typing import Any, ParamSpec, TypeVar
 
 from common_library.async_tools import cancel_wait_task
+from common_library.logging.logging_errors import create_troubleshooting_log_kwargs
 
 from ..background_task import periodic
-from ..logging_errors import create_troubleshooting_log_kwargs
 from ._client import RedisClientSDK
 from ._constants import (
     DEFAULT_SEMAPHORE_TTL,

packages/service-library/tests/test_logging_utils.py

@@ -10,8 +10,8 @@
 from typing import Any
 
 import pytest
+from common_library.logging.logging_base import get_log_record_extra
 from faker import Faker
-from servicelib.logging_base import get_log_record_extra
 from servicelib.logging_utils import (
     _DEFAULT_FORMATTING,
     CustomFormatter,

services/agent/src/simcore_service_agent/core/settings.py

@@ -2,10 +2,10 @@
 from typing import Annotated
 
 from common_library.basic_types import DEFAULT_FACTORY
+from common_library.logging.logging_utils_filtering import LoggerName, MessageSubstring
 from models_library.basic_types import BootModeEnum, LogLevel
 from models_library.docker import DockerNodeID
 from pydantic import AliasChoices, AnyHttpUrl, Field, field_validator
-from servicelib.logging_utils_filtering import LoggerName, MessageSubstring
 from settings_library.base import BaseCustomSettings
 from settings_library.r_clone import S3Provider
 from settings_library.rabbit import RabbitSettings

services/api-server/src/simcore_service_api_server/_service_function_jobs_task_client.py

@@ -5,6 +5,7 @@
 
 from celery_library.errors import TaskNotFoundError
 from common_library.exclude import as_dict_exclude_none
+from common_library.logging.logging_errors import create_troubleshooting_log_kwargs
 from models_library.functions import (
     FunctionClass,
     FunctionID,
@@ -33,7 +34,6 @@
 from models_library.users import UserID
 from servicelib.celery.models import TaskMetadata, TasksQueue, TaskUUID
 from servicelib.celery.task_manager import TaskManager
-from servicelib.logging_errors import create_troubleshooting_log_kwargs
 from simcore_service_api_server.models.schemas.functions import (
     FunctionJobCreationTaskStatus,
 )
@@ -137,15 +137,16 @@ async def list_function_jobs_with_status(
             pagination_offset=pagination_offset, pagination_limit=pagination_limit
         )
 
-        function_jobs_list_ws, meta = (
-            await self._web_rpc_client.list_function_jobs_with_status(
-                user_id=self.user_id,
-                product_name=self.product_name,
-                filter_by_function_id=filter_by_function_id,
-                filter_by_function_job_ids=filter_by_function_job_ids,
-                filter_by_function_job_collection_id=filter_by_function_job_collection_id,
-                **pagination_kwargs,
-            )
+        (
+            function_jobs_list_ws,
+            meta,
+        ) = await self._web_rpc_client.list_function_jobs_with_status(
+            user_id=self.user_id,
+            product_name=self.product_name,
+            filter_by_function_id=filter_by_function_id,
+            filter_by_function_job_ids=filter_by_function_job_ids,
+            filter_by_function_job_collection_id=filter_by_function_job_collection_id,
+            **pagination_kwargs,
         )
 
         for function_job_wso in function_jobs_list_ws:
@@ -297,7 +298,6 @@ async def function_job_outputs(
         function_job: RegisteredFunctionJob,
         stored_job_outputs: FunctionOutputs | None,
     ) -> FunctionOutputs:
-
         if stored_job_outputs is not None:
             return stored_job_outputs
 
@@ -360,7 +360,6 @@ async def create_function_job_creation_task(
         parent_project_uuid: ProjectID | None = None,
         parent_node_id: NodeID | None = None,
     ) -> RegisteredFunctionJob:
-
         job_inputs = await self._function_job_service.create_function_job_inputs(
             function=function, function_inputs=function_inputs
         )

services/api-server/src/simcore_service_api_server/api/routes/tasks.py

@@ -2,6 +2,7 @@
 from typing import Annotated, Any
 
 from common_library.error_codes import create_error_code
+from common_library.logging.logging_errors import create_troubleshooting_log_kwargs
 from fastapi import APIRouter, Depends, FastAPI, HTTPException, status
 from models_library.api_schemas_long_running_tasks.base import TaskProgress
 from models_library.api_schemas_long_running_tasks.tasks import (
@@ -16,7 +17,6 @@
 from models_library.users import UserID
 from servicelib.celery.models import TaskState, TaskUUID
 from servicelib.fastapi.dependencies import get_app
-from servicelib.logging_errors import create_troubleshooting_log_kwargs
 from simcore_service_api_server.models.domain.celery_models import ApiWorkerTaskFilter
 
 from ...models.schemas.base import ApiServerEnvelope
@@ -57,7 +57,6 @@ async def list_tasks(
     user_id: Annotated[UserID, Depends(get_current_user_id)],
     product_name: Annotated[ProductName, Depends(get_product_name)],
 ):
-
     task_manager = get_task_manager(app)
     task_filter = ApiWorkerTaskFilter(
         user_id=user_id,