🎨Maintenance: make redis client use the client name to ease debugging (#6700)

Author: sanderegg

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

@@ -60,6 +60,7 @@ async def _cancel_or_warn(task: Task) -> None:
 @dataclass
 class RedisClientSDK:
     redis_dsn: str
+    client_name: str
     decode_responses: bool = _DEFAULT_DECODE_RESPONSES
     health_check_interval: datetime.timedelta = _DEFAULT_HEALTH_CHECK_INTERVAL
 
@@ -86,7 +87,7 @@ def __post_init__(self):
             socket_connect_timeout=_DEFAULT_SOCKET_TIMEOUT.total_seconds(),
             encoding="utf-8",
             decode_responses=self.decode_responses,
-            auto_close_connection_pool=True,
+            client_name=self.client_name,
         )
 
     @retry(**RedisRetryPolicyUponInitialization(_logger).kwargs)
@@ -238,6 +239,7 @@ class RedisClientsManager:
 
     databases_configs: set[RedisManagerDBConfig]
     settings: RedisSettings
+    client_name: str
 
     _client_sdks: dict[RedisDatabase, RedisClientSDK] = field(default_factory=dict)
 
@@ -247,6 +249,7 @@ async def setup(self) -> None:
                 redis_dsn=self.settings.build_redis_dsn(config.database),
                 decode_responses=config.decode_responses,
                 health_check_interval=config.health_check_interval,
+                client_name=f"{self.client_name}",
             )
 
         for client in self._client_sdks.values():

packages/service-library/tests/conftest.py

@@ -80,9 +80,12 @@ async def _(
         database: RedisDatabase, decode_response: bool = True  # noqa: FBT002
     ) -> AsyncIterator[RedisClientSDK]:
         redis_resources_dns = redis_service.build_redis_dsn(database)
-        client = RedisClientSDK(redis_resources_dns, decode_responses=decode_response)
+        client = RedisClientSDK(
+            redis_resources_dns, decode_responses=decode_response, client_name="pytest"
+        )
         assert client
         assert client.redis_dsn == redis_resources_dns
+        assert client.client_name == "pytest"
         await client.setup()
 
         yield client
@@ -94,7 +97,9 @@ async def _cleanup_redis_data(clients_manager: RedisClientsManager) -> None:
             await clients_manager.client(db).redis.flushall()
 
     async with RedisClientsManager(
-        {RedisManagerDBConfig(db) for db in RedisDatabase}, redis_service
+        {RedisManagerDBConfig(db) for db in RedisDatabase},
+        redis_service,
+        client_name="pytest",
     ) as clients_manager:
         await _cleanup_redis_data(clients_manager)
         yield _

packages/service-library/tests/deferred_tasks/example_app.py

@@ -60,6 +60,7 @@ def __init__(self, redis_settings: RedisSettings, port: int) -> None:
         self.redis: Redis = RedisClientSDK(
             redis_settings.build_redis_dsn(RedisDatabase.DEFERRED_TASKS),
             decode_responses=True,
+            client_name="example_app",
         ).redis
         self.port = port
 
@@ -84,6 +85,7 @@ def __init__(
         self._redis_client = RedisClientSDK(
             redis_settings.build_redis_dsn(RedisDatabase.DEFERRED_TASKS),
             decode_responses=False,
+            client_name="example_app",
         )
         self._manager = DeferredManager(
             rabbit_settings,

packages/service-library/tests/deferred_tasks/test__base_deferred_handler.py

@@ -55,6 +55,7 @@ async def redis_client_sdk(
     sdk = RedisClientSDK(
         redis_service.build_redis_dsn(RedisDatabase.DEFERRED_TASKS),
         decode_responses=False,
+        client_name="pytest",
     )
     await sdk.setup()
     yield sdk

packages/service-library/tests/test_pools.py

@@ -1,4 +1,4 @@
-from asyncio import BaseEventLoop
+import asyncio
 from concurrent.futures import ProcessPoolExecutor
 
 from servicelib.pools import (
@@ -11,17 +11,25 @@ def return_int_one() -> int:
     return 1
 
 
-async def test_default_thread_pool_executor(event_loop: BaseEventLoop) -> None:
-    assert await event_loop.run_in_executor(None, return_int_one) == 1
+async def test_default_thread_pool_executor() -> None:
+    assert await asyncio.get_running_loop().run_in_executor(None, return_int_one) == 1
 
 
-async def test_blocking_process_pool_executor(event_loop: BaseEventLoop) -> None:
-    assert await event_loop.run_in_executor(ProcessPoolExecutor(), return_int_one) == 1
+async def test_blocking_process_pool_executor() -> None:
+    assert (
+        await asyncio.get_running_loop().run_in_executor(
+            ProcessPoolExecutor(), return_int_one
+        )
+        == 1
+    )
 
 
-async def test_non_blocking_process_pool_executor(event_loop: BaseEventLoop) -> None:
+async def test_non_blocking_process_pool_executor() -> None:
     with non_blocking_process_pool_executor() as executor:
-        assert await event_loop.run_in_executor(executor, return_int_one) == 1
+        assert (
+            await asyncio.get_running_loop().run_in_executor(executor, return_int_one)
+            == 1
+        )
 
 
 async def test_same_pool_instances() -> None:
@@ -36,9 +44,12 @@ async def test_different_pool_instances() -> None:
         assert first != second
 
 
-async def test_non_blocking_thread_pool_executor(event_loop: BaseEventLoop) -> None:
+async def test_non_blocking_thread_pool_executor() -> None:
     with non_blocking_thread_pool_executor() as executor:
-        assert await event_loop.run_in_executor(executor, return_int_one) == 1
+        assert (
+            await asyncio.get_running_loop().run_in_executor(executor, return_int_one)
+            == 1
+        )
 
 
 async def test_same_thread_pool_instances() -> None:

packages/service-library/tests/test_redis.py

@@ -277,7 +277,9 @@ async def test_redis_client_sdks_manager(
         RedisManagerDBConfig(db) for db in RedisDatabase
     }
     manager = RedisClientsManager(
-        databases_configs=all_redis_configs, settings=redis_service
+        databases_configs=all_redis_configs,
+        settings=redis_service,
+        client_name="pytest",
     )
 
     async with manager:
@@ -290,7 +292,7 @@ async def test_redis_client_sdk_setup_shutdown(
 ):
     # setup
     redis_resources_dns = redis_service.build_redis_dsn(RedisDatabase.RESOURCES)
-    client = RedisClientSDK(redis_resources_dns)
+    client = RedisClientSDK(redis_resources_dns, client_name="pytest")
     assert client
     assert client.redis_dsn == redis_resources_dns
 

packages/service-library/tests/test_redis__recoonection.py renamed to packages/service-library/tests/test_redis__reconection.py

@@ -21,9 +21,9 @@ async def test_redis_client_sdk_lost_connection(
     docker_client: docker.client.DockerClient,
 ):
     redis_client_sdk = RedisClientSDK(
-        redis_service.build_redis_dsn(RedisDatabase.RESOURCES)
+        redis_service.build_redis_dsn(RedisDatabase.RESOURCES), client_name="pytest"
     )
-
+    assert redis_client_sdk.client_name == "pytest"
     await redis_client_sdk.setup()
 
     assert await redis_client_sdk.ping() is True
@@ -41,3 +41,5 @@ async def test_redis_client_sdk_lost_connection(
     ):
         with attempt:
             assert await redis_client_sdk.ping() is False
+
+    await redis_client_sdk.shutdown()

packages/service-library/tests/test_utils.py

@@ -5,7 +5,6 @@
 import asyncio
 from collections.abc import AsyncIterator, Awaitable, Coroutine, Iterator
 from copy import copy, deepcopy
-from random import randint
 from typing import NoReturn
 from unittest import mock
 
@@ -66,7 +65,6 @@ def mock_logger(mocker: MockerFixture) -> Iterator[mock.Mock]:
 
 
 async def test_logged_gather(
-    event_loop: asyncio.AbstractEventLoop,
     coros: list[Coroutine],
     mock_logger: mock.Mock,
 ):
@@ -79,7 +77,7 @@ async def test_logged_gather(
     # NOTE: only first error in the list is raised, since it is not RuntimeError, that task
     assert isinstance(excinfo.value, ValueError)
 
-    for task in asyncio.all_tasks(event_loop):
+    for task in asyncio.all_tasks(asyncio.get_running_loop()):
         if task is not asyncio.current_task():
             # info
             task.print_stack()
@@ -148,7 +146,7 @@ async def test_fire_and_forget_1000s_tasks(faker: Faker):
     tasks_collection = set()
 
     async def _some_task(n: int) -> str:
-        await asyncio.sleep(randint(1, 3))
+        await asyncio.sleep(faker.random_int(1, 3))
         return f"I'm great since I slept a bit, and by the way I'm task {n}"
 
     for n in range(1000):
@@ -251,7 +249,6 @@ async def test_limited_gather_limits(
 
 
 async def test_limited_gather(
-    event_loop: asyncio.AbstractEventLoop,
     coros: list[Coroutine],
     mock_logger: mock.Mock,
 ):
@@ -266,7 +263,7 @@ async def test_limited_gather(
 
     unfinished_tasks = [
         task
-        for task in asyncio.all_tasks(event_loop)
+        for task in asyncio.all_tasks(asyncio.get_running_loop())
         if task is not asyncio.current_task()
     ]
     final_results = await asyncio.gather(*unfinished_tasks, return_exceptions=True)
@@ -288,9 +285,7 @@ async def test_limited_gather_wo_raising(
     assert results[5] == 5
 
 
-async def test_limited_gather_cancellation(
-    event_loop: asyncio.AbstractEventLoop, slow_successful_coros_list: list[Coroutine]
-):
+async def test_limited_gather_cancellation(slow_successful_coros_list: list[Coroutine]):
     task = asyncio.create_task(limited_gather(*slow_successful_coros_list, limit=0))
     await asyncio.sleep(3)
     task.cancel()
@@ -300,7 +295,7 @@ async def test_limited_gather_cancellation(
     # check all coros are cancelled
     unfinished_tasks = [
         task
-        for task in asyncio.all_tasks(event_loop)
+        for task in asyncio.all_tasks(asyncio.get_running_loop())
         if task is not asyncio.current_task()
     ]
     assert not unfinished_tasks

services/autoscaling/src/simcore_service_autoscaling/modules/redis.py

@@ -5,6 +5,8 @@
 from servicelib.redis import RedisClientSDK
 from settings_library.redis import RedisDatabase, RedisSettings
 
+from .._meta import APP_NAME
+
 logger = logging.getLogger(__name__)
 
 
@@ -13,7 +15,9 @@ async def on_startup() -> None:
         app.state.redis_client_sdk = None
         settings: RedisSettings = app.state.settings.AUTOSCALING_REDIS
         redis_locks_dsn = settings.build_redis_dsn(RedisDatabase.LOCKS)
-        app.state.redis_client_sdk = client = RedisClientSDK(redis_locks_dsn)
+        app.state.redis_client_sdk = client = RedisClientSDK(
+            redis_locks_dsn, client_name=APP_NAME
+        )
         await client.setup()
 
     async def on_shutdown() -> None:

services/clusters-keeper/src/simcore_service_clusters_keeper/modules/clusters_management_task.py

@@ -6,6 +6,7 @@
 from servicelib.background_task import start_periodic_task, stop_periodic_task
 from servicelib.redis_utils import exclusive
 
+from .._meta import APP_NAME
 from ..core.settings import ApplicationSettings
 from ..modules.redis import get_redis_client
 from .clusters_management_core import check_clusters
@@ -19,7 +20,7 @@ def on_app_startup(app: FastAPI) -> Callable[[], Awaitable[None]]:
     async def _startup() -> None:
         app_settings: ApplicationSettings = app.state.settings
 
-        lock_key = f"{app.title}:clusters-management_lock"
+        lock_key = f"{APP_NAME}:clusters-management_lock"
         lock_value = json.dumps({})
         app.state.clusters_cleaning_task = start_periodic_task(
             exclusive(get_redis_client(app), lock_key=lock_key, lock_value=lock_value)(