change usage

Author: sanderegg

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/status_monitor/_monitor.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 from datetime import timedelta
 from functools import cached_property
@@ -8,7 +9,8 @@
 from models_library.projects_nodes_io import NodeID
 from pydantic import NonNegativeFloat, NonNegativeInt
 from servicelib.async_utils import cancel_wait_task
-from servicelib.redis import create_exclusive_periodic_task
+from servicelib.background_task_utils import exclusive_periodic
+from servicelib.redis._client import RedisClientSDK
 from servicelib.utils import limited_gather
 from settings_library.redis import RedisDatabase
 
@@ -60,6 +62,10 @@ def _can_be_removed(model: TrackedServiceModel) -> bool:
     return False
 
 
+def _get_redis_client_from_monitor(monitor: "Monitor") -> RedisClientSDK:
+    return get_redis_client(monitor.app, RedisDatabase.LOCKS)
+
+
 class Monitor:
     def __init__(self, app: FastAPI, status_worker_interval: timedelta) -> None:
         self.app = app
@@ -69,6 +75,11 @@ def __init__(self, app: FastAPI, status_worker_interval: timedelta) -> None:
     def status_worker_interval_seconds(self) -> NonNegativeFloat:
         return self.status_worker_interval.total_seconds()
 
+    @exclusive_periodic(
+        _get_redis_client_from_monitor,
+        task_interval=_INTERVAL_BETWEEN_CHECKS,
+        retry_after=_INTERVAL_BETWEEN_CHECKS,
+    )
     async def _worker_check_services_require_status_update(self) -> None:
         """
         Check if any service requires it's status to be polled.
@@ -133,12 +144,9 @@ async def _worker_check_services_require_status_update(self) -> None:
         )
 
     async def setup(self) -> None:
-        self.app.state.status_monitor_background_task = create_exclusive_periodic_task(
-            get_redis_client(self.app, RedisDatabase.LOCKS),
-            self._worker_check_services_require_status_update,
-            task_period=_INTERVAL_BETWEEN_CHECKS,
-            retry_after=_INTERVAL_BETWEEN_CHECKS,
-            task_name="periodic_service_status_update",
+        self.app.state.status_monitor_background_task = asyncio.create_task(
+            self._worker_check_services_require_status_update(),
+            name="periodic_service_status_update",
         )
 
     async def shutdown(self) -> None:

services/efs-guardian/src/simcore_service_efs_guardian/services/background_tasks.py

@@ -1,9 +1,10 @@
 import logging
-from datetime import datetime, timezone
+from datetime import UTC, datetime, timedelta
 
 from fastapi import FastAPI
 from models_library.projects import ProjectID
 from models_library.projects_state import ProjectStatus
+from servicelib.background_task_utils import exclusive_periodic
 from servicelib.logging_utils import log_context
 from servicelib.project_lock import (
     PROJECT_LOCK_TIMEOUT,
@@ -22,14 +23,19 @@
 _logger = logging.getLogger(__name__)
 
 
+@exclusive_periodic(
+    get_redis_lock_client,
+    task_interval=timedelta(hours=1),
+    retry_after=timedelta(minutes=5),
+)
 async def removal_policy_task(app: FastAPI) -> None:
     _logger.info("Removal policy task started")
 
     app_settings: ApplicationSettings = app.state.settings
     assert app_settings  # nosec
     efs_manager: EfsManager = app.state.efs_manager
 
-    base_start_timestamp = datetime.now(tz=timezone.utc)
+    base_start_timestamp = datetime.now(tz=UTC)
 
     efs_project_ids: list[
         ProjectID

services/efs-guardian/src/simcore_service_efs_guardian/services/background_tasks_setup.py

@@ -1,23 +1,16 @@
 import asyncio
 import logging
 from collections.abc import Awaitable, Callable
-from datetime import timedelta
 from typing import TypedDict
 
 from fastapi import FastAPI
 from servicelib.async_utils import cancel_wait_task
 from servicelib.logging_utils import log_catch, log_context
-from servicelib.redis import create_exclusive_periodic_task
 
 from .background_tasks import removal_policy_task
-from .modules.redis import get_redis_lock_client
 
 _logger = logging.getLogger(__name__)
 
-_SEC = 1  # in s
-_MIN = 60 * _SEC  # in s
-_HOUR = 60 * _MIN  # in s
-
 
 class EfsGuardianBackgroundTask(TypedDict):
     name: str
@@ -41,15 +34,9 @@ async def _startup() -> None:
 
             # Setup periodic tasks
             for task in _EFS_GUARDIAN_BACKGROUND_TASKS:
-                exclusive_task = create_exclusive_periodic_task(
-                    get_redis_lock_client(app),
-                    task["task_func"],
-                    task_period=timedelta(seconds=1 * _HOUR),
-                    retry_after=timedelta(seconds=5 * _MIN),
-                    task_name=task["name"],
-                    app=app,
+                app.state.efs_guardian_background_tasks.append(
+                    asyncio.create_task(task["task_func"](), name=task["name"])
                 )
-                app.state.efs_guardian_background_tasks.append(exclusive_task)
 
     return _startup
 

services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/services/background_task_periodic_heartbeat_check.py

@@ -126,7 +126,7 @@ async def _close_unhealthy_service(
     )
 
 
-async def periodic_check_of_running_services_task(app: FastAPI) -> None:
+async def check_running_services(app: FastAPI) -> None:
     _logger.info("Periodic check started")
 
     # This check runs across all products

services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/services/background_task_periodic_heartbeat_check_setup.py

@@ -1,16 +1,15 @@
+import asyncio
 import logging
 from collections.abc import Awaitable, Callable
 from typing import TypedDict
 
 from fastapi import FastAPI
 from servicelib.async_utils import cancel_wait_task
+from servicelib.background_task_utils import exclusive_periodic
 from servicelib.logging_utils import log_catch, log_context
-from servicelib.redis import create_exclusive_periodic_task
 
 from ..core.settings import ApplicationSettings
-from .background_task_periodic_heartbeat_check import (
-    periodic_check_of_running_services_task,
-)
+from .background_task_periodic_heartbeat_check import check_running_services
 from .modules.redis import get_redis_lock_client
 
 _logger = logging.getLogger(__name__)
@@ -38,17 +37,19 @@ async def _startup() -> None:
 
             app.state.rut_background_task__periodic_check_of_running_services = None
 
-            # Setup periodic task
-            exclusive_task = create_exclusive_periodic_task(
+            @exclusive_periodic(
                 get_redis_lock_client(app),
-                periodic_check_of_running_services_task,
-                task_period=app_settings.RESOURCE_USAGE_TRACKER_MISSED_HEARTBEAT_INTERVAL_SEC,
+                task_interval=app_settings.RESOURCE_USAGE_TRACKER_MISSED_HEARTBEAT_INTERVAL_SEC,
                 retry_after=app_settings.RESOURCE_USAGE_TRACKER_MISSED_HEARTBEAT_INTERVAL_SEC,
-                task_name=_TASK_NAME_PERIODICALY_CHECK_RUNNING_SERVICES,
-                app=app,
             )
+            async def _periodic_check_running_services() -> None:
+                await check_running_services(app)
+
             app.state.rut_background_task__periodic_check_of_running_services = (
-                exclusive_task
+                asyncio.create_task(
+                    _periodic_check_running_services(),
+                    name=_TASK_NAME_PERIODICALY_CHECK_RUNNING_SERVICES,
+                )
             )
 
     return _startup

services/storage/src/simcore_service_storage/dsm_cleaner.py

@@ -25,8 +25,8 @@
 
 from aiohttp import web
 from servicelib.async_utils import cancel_wait_task
+from servicelib.background_task_utils import exclusive_periodic
 from servicelib.logging_utils import log_catch, log_context
-from servicelib.redis import create_exclusive_periodic_task
 
 from .constants import APP_CONFIG_KEY, APP_DSM_KEY
 from .dsm_factory import DataManagerProvider
@@ -45,14 +45,7 @@ async def dsm_cleaner_task(app: web.Application) -> None:
     simcore_s3_dsm: SimcoreS3DataManager = cast(
         SimcoreS3DataManager, dsm.get(SimcoreS3DataManager.get_location_id())
     )
-    try:
-        await simcore_s3_dsm.clean_expired_uploads()
-
-    except asyncio.CancelledError:  # noqa: PERF203
-        _logger.info("cancelled dsm cleaner task")
-        raise
-    except Exception:  # pylint: disable=broad-except
-        _logger.exception("Unhandled error in dsm cleaner task, restarting task...")
+    await simcore_s3_dsm.clean_expired_uploads()
 
 
 def setup_dsm_cleaner(app: web.Application):
@@ -64,15 +57,17 @@ async def _setup(app: web.Application):
             cfg: Settings = app[APP_CONFIG_KEY]
             assert cfg.STORAGE_CLEANER_INTERVAL_S  # nosec
 
-            storage_background_task = create_exclusive_periodic_task(
+            @exclusive_periodic(
                 get_redis_client(app),
-                dsm_cleaner_task,
-                task_period=timedelta(seconds=cfg.STORAGE_CLEANER_INTERVAL_S),
+                task_interval=timedelta(seconds=cfg.STORAGE_CLEANER_INTERVAL_S),
                 retry_after=timedelta(minutes=5),
-                task_name=_TASK_NAME_PERIODICALY_CLEAN_DSM,
-                app=app,
             )
+            async def _periodic_dsm_clean() -> None:
+                await dsm_cleaner_task(app)
 
+            storage_background_task = asyncio.create_task(
+                _periodic_dsm_clean(), name=_TASK_NAME_PERIODICALY_CLEAN_DSM
+            )
             yield
 
             await cancel_wait_task(storage_background_task)