daily work

Author: matusdrobuliak66

packages/models-library/src/models_library/rabbitmq_messages.py

@@ -194,6 +194,23 @@ def routing_key(self) -> str | None:
         return None
 
 
+class DynamicServiceRunningMessage(RabbitMessageBase):
+    channel_name: Literal["io.simcore.service.dynamic-service-running"] = Field(
+        default="io.simcore.service.dynamic-service-running", const=True
+    )
+
+    project_id: ProjectID
+    node_id: NodeID
+    product_name: ProductName | None
+    created_at: datetime.datetime = Field(
+        default_factory=lambda: arrow.utcnow().datetime,
+        description="message creation datetime",
+    )
+
+    def routing_key(self) -> str | None:
+        return None
+
+
 class RabbitResourceTrackingStartedMessage(RabbitResourceTrackingBaseMessage):
     message_type: RabbitResourceTrackingMessageType = Field(
         default=RabbitResourceTrackingMessageType.TRACKING_STARTED, const=True

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

@@ -151,6 +151,7 @@ async def subscribe(
         message_ttl: NonNegativeInt = RABBIT_QUEUE_MESSAGE_DEFAULT_TTL_MS,
         unexpected_error_retry_delay_s: float = _DEFAULT_UNEXPECTED_ERROR_RETRY_DELAY_S,
         unexpected_error_max_attempts: int = _DEFAULT_UNEXPECTED_ERROR_MAX_ATTEMPTS,
+        queue_name: str | None = None,
     ) -> str:
         """subscribe to exchange_name calling ``message_handler`` for every incoming message
         - exclusive_queue: True means that every instance of this application will
@@ -216,6 +217,7 @@ async def subscribe(
                 exclusive_queue=exclusive_queue,
                 message_ttl=message_ttl,
                 arguments={"x-dead-letter-exchange": delayed_exchange_name},
+                queue_name=queue_name,
             )
             if topics is None:
                 await queue.bind(exchange, routing_key="")
@@ -235,6 +237,7 @@ async def subscribe(
                 exclusive_queue=exclusive_queue,
                 message_ttl=int(unexpected_error_retry_delay_s * 1000),
                 arguments={"x-dead-letter-exchange": exchange.name},
+                queue_name=queue_name,
             )
             await delayed_queue.bind(delayed_exchange)
 

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

@@ -70,6 +70,7 @@ async def declare_queue(
     exclusive_queue: bool,
     arguments: dict[str, Any] | None = None,
     message_ttl: NonNegativeInt = RABBIT_QUEUE_MESSAGE_DEFAULT_TTL_MS,
+    queue_name: str | None = None,
 ) -> aio_pika.abc.AbstractRobustQueue:
     default_arguments = {"x-message-ttl": message_ttl}
     if arguments is not None:
@@ -82,7 +83,7 @@ async def declare_queue(
     }
     if not exclusive_queue:
         # NOTE: setting a name will ensure multiple instance will take their data here
-        queue_parameters |= {"name": exchange_name}
+        queue_parameters |= {"name": queue_name if queue_name else exchange_name}
 
     # NOTE: if below line raises something similar to ``ChannelPreconditionFailed: PRECONDITION_FAILED``
     # most likely someone changed the signature of the queues (parameters etc...)

services/docker-compose.yml

@@ -407,6 +407,11 @@ services:
       RABBIT_PORT: ${RABBIT_PORT}
       RABBIT_SECURE: ${RABBIT_SECURE}
       RABBIT_USER: ${RABBIT_USER}
+      REDIS_HOST: ${REDIS_HOST}
+      REDIS_PASSWORD: ${REDIS_PASSWORD}
+      REDIS_PORT: ${REDIS_PORT}
+      REDIS_SECURE: ${REDIS_SECURE}
+      REDIS_USER: ${REDIS_USER}
       SC_USER_ID: ${SC_USER_ID}
       SC_USER_NAME: ${SC_USER_NAME}
       EFS_USER_ID: ${EFS_USER_ID}

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/core/rabbitmq.py

@@ -5,6 +5,7 @@
 from fastapi import FastAPI
 from models_library.progress_bar import ProgressReport
 from models_library.rabbitmq_messages import (
+    DynamicServiceRunningMessage,
     EventRabbitMessage,
     LoggerRabbitMessage,
     ProgressRabbitMessageNode,
@@ -34,6 +35,12 @@ async def post_resource_tracking_message(
     await _post_rabbit_message(app, message)
 
 
+async def post_dynamic_service_running_message(
+    app: FastAPI, message: DynamicServiceRunningMessage
+):
+    await _post_rabbit_message(app, message)
+
+
 async def post_log_message(
     app: FastAPI, log: LogMessageStr, *, log_level: LogLevelInt
 ) -> None:

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/resource_tracking/_core.py

@@ -1,9 +1,11 @@
+import asyncio
 import logging
 from typing import Final
 
 from fastapi import FastAPI
 from models_library.generated_models.docker_rest_api import ContainerState
 from models_library.rabbitmq_messages import (
+    DynamicServiceRunningMessage,
     RabbitResourceTrackingHeartbeatMessage,
     RabbitResourceTrackingStartedMessage,
     RabbitResourceTrackingStoppedMessage,
@@ -19,7 +21,10 @@
     are_all_containers_in_expected_states,
     get_container_states,
 )
-from ...core.rabbitmq import post_resource_tracking_message
+from ...core.rabbitmq import (
+    post_dynamic_service_running_message,
+    post_resource_tracking_message,
+)
 from ...core.settings import ApplicationSettings, ResourceTrackingSettings
 from ...models.shared_store import SharedStore
 from ._models import ResourceTrackingState
@@ -70,10 +75,20 @@ async def _heart_beat_task(app: FastAPI):
     )
 
     if are_all_containers_in_expected_states(container_states.values()):
-        message = RabbitResourceTrackingHeartbeatMessage(
+        rut_message = RabbitResourceTrackingHeartbeatMessage(
             service_run_id=settings.DY_SIDECAR_RUN_ID
         )
-        await post_resource_tracking_message(app, message)
+        dyn_message = DynamicServiceRunningMessage(
+            project_id=settings.DY_SIDECAR_PROJECT_ID,
+            node_id=settings.DY_SIDECAR_NODE_ID,
+            product_name=settings.DY_SIDECAR_PRODUCT_NAME,
+        )
+        await asyncio.gather(
+            *[
+                post_resource_tracking_message(app, rut_message),
+                post_dynamic_service_running_message(app, dyn_message),
+            ]
+        )
     else:
         _logger.info(
             "heart beat message skipped: container_states=%s", container_states

services/efs-guardian/src/simcore_service_efs_guardian/core/application.py

@@ -13,8 +13,11 @@
 )
 from ..api.rest.routes import setup_api_routes
 from ..api.rpc.routes import setup_rpc_routes
+from ..services.background_tasks_setup import setup as setup_background_tasks
 from ..services.efs_manager_setup import setup as setup_efs_manager
 from ..services.modules.rabbitmq import setup as setup_rabbitmq
+from ..services.modules.redis import setup as setup_redis
+from ..services.process_messages_setup import setup as setup_process_messages
 from .settings import ApplicationSettings
 
 logger = logging.getLogger(__name__)
@@ -40,11 +43,14 @@ def create_app(settings: ApplicationSettings) -> FastAPI:
 
     # PLUGINS SETUP
     setup_rabbitmq(app)
+    setup_redis(app)
 
     setup_api_routes(app)
     setup_rpc_routes(app)
 
     setup_efs_manager(app)
+    setup_background_tasks(app)
+    setup_process_messages(app)
 
     # EVENTS
     async def _on_startup() -> None:

services/efs-guardian/src/simcore_service_efs_guardian/core/settings.py

@@ -12,6 +12,7 @@
 from settings_library.base import BaseCustomSettings
 from settings_library.efs import AwsEfsSettings
 from settings_library.rabbit import RabbitSettings
+from settings_library.redis import RedisSettings
 from settings_library.tracing import TracingSettings
 from settings_library.utils_logging import MixinLoggingSettings
 
@@ -76,6 +77,7 @@ class ApplicationSettings(BaseCustomSettings, MixinLoggingSettings):
 
     EFS_GUARDIAN_AWS_EFS_SETTINGS: AwsEfsSettings = Field(auto_default_from_env=True)
     EFS_GUARDIAN_RABBITMQ: RabbitSettings = Field(auto_default_from_env=True)
+    EFS_GUARDIAN_REDIS: RedisSettings = Field(auto_default_from_env=True)
     EFS_GUARDIAN_TRACING: TracingSettings | None = Field(
         auto_default_from_env=True, description="settings for opentelemetry tracing"
     )

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

@@ -0,0 +1,18 @@
+import logging
+
+from fastapi import FastAPI
+
+from ..core.settings import ApplicationSettings
+
+_logger = logging.getLogger(__name__)
+
+
+async def removal_policy_task(app: FastAPI) -> None:
+    _logger.info("Removal policy task started")
+
+    # After X days of inactivity remove data from EFS
+    # Probably use `last_modified_data` in the project DB table
+    # Maybe lock project during this time lock_project()
+
+    app_settings: ApplicationSettings = app.state.settings
+    assert app_settings  # nosec

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

@@ -0,0 +1,73 @@
+import asyncio
+import logging
+from collections.abc import Awaitable, Callable
+from datetime import timedelta
+from typing import TypedDict
+
+from fastapi import FastAPI
+from servicelib.background_task import stop_periodic_task
+from servicelib.logging_utils import log_catch, log_context
+from servicelib.redis_utils import start_exclusive_periodic_task
+
+from .background_tasks import removal_policy_task
+from .modules.redis import get_redis_client
+
+_logger = logging.getLogger(__name__)
+
+
+class EfsGuardianBackgroundTask(TypedDict):
+    name: str
+    task_func: Callable
+
+
+_EFS_GUARDIAN_BACKGROUND_TASKS = [
+    EfsGuardianBackgroundTask(
+        name="efs_removal_policy_task", task_func=removal_policy_task
+    )
+]
+
+
+def on_app_startup(app: FastAPI) -> Callable[[], Awaitable[None]]:
+    async def _startup() -> None:
+        with log_context(
+            _logger, logging.INFO, msg="Efs Guardian startup.."
+        ), log_catch(_logger, reraise=False):
+            app.state.efs_guardian_background_tasks = []
+
+            # Setup periodic tasks
+            for task in _EFS_GUARDIAN_BACKGROUND_TASKS:
+                exclusive_task = start_exclusive_periodic_task(
+                    get_redis_client(app),
+                    task["task_func"],
+                    task_period=timedelta(seconds=60),  # 1 minute
+                    retry_after=timedelta(seconds=60),  # 5 minutes
+                    task_name=task["name"],
+                    app=app,
+                )
+                app.state.efs_guardian_background_tasks.append(exclusive_task)
+
+    return _startup
+
+
+def on_app_shutdown(
+    _app: FastAPI,
+) -> Callable[[], Awaitable[None]]:
+    async def _stop() -> None:
+        with log_context(
+            _logger, logging.INFO, msg="Efs Guardian shutdown.."
+        ), log_catch(_logger, reraise=False):
+            assert _app  # nosec
+            if _app.state.efs_guardian_background_tasks:
+                await asyncio.gather(
+                    *[
+                        stop_periodic_task(task)
+                        for task in _app.state.efs_guardian_background_tasks
+                    ]
+                )
+
+    return _stop
+
+
+def setup(app: FastAPI) -> None:
+    app.add_event_handler("startup", on_app_startup(app))
+    app.add_event_handler("shutdown", on_app_shutdown(app))