ongoing new scheduler

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/core/settings.py

@@ -4,7 +4,9 @@
 
 import datetime
 from functools import cached_property
+from typing import cast
 
+from fastapi import FastAPI
 from models_library.basic_types import (
     BootModeEnum,
     BuildTargetEnum,
@@ -240,3 +242,7 @@ class AppSettings(BaseCustomSettings, MixinLoggingSettings):
     def _validate_loglevel(cls, value: str) -> str:
         log_level: str = cls.validate_log_level(value)
         return log_level
+
+
+def get_application_settings(app: FastAPI) -> AppSettings:
+    return cast(AppSettings, app.state.settings)

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_distributed_scheduler.py

@@ -0,0 +1,31 @@
+import datetime
+from typing import Final
+
+from fastapi import FastAPI
+from servicelib.redis import RedisClientSDK
+from servicelib.redis_utils import exclusive
+from settings_library.redis import RedisDatabase
+
+from ...core.settings import get_application_settings
+from ...utils.comp_scheduler import SCHEDULED_STATES
+from ..db import get_db_engine
+from ..db.repositories.comp_runs import CompRunsRepository
+from ..redis import get_redis_client_manager
+
+_SCHEDULER_INTERVAL: Final[datetime.timedelta] = datetime.timedelta(seconds=5)
+
+
+def _redis_client_getter(*args, **kwargs) -> RedisClientSDK:
+    assert kwargs  # nosec
+    app = args[0]
+    assert isinstance(app, FastAPI)  # nosec
+    return get_redis_client_manager(app).client(RedisDatabase.LOCKS)
+
+
+@exclusive(_redis_client_getter, lock_key="computational-distributed-scheduler")
+async def schedule_pipelines(app: FastAPI) -> None:
+    app_settings = get_application_settings(app)
+    db_engine = get_db_engine(app)
+    runs_to_schedule = CompRunsRepository.instance(db_engine).list(
+        filter_by_state=SCHEDULED_STATES, scheduled_since=_SCHEDULER_INTERVAL
+    )

services/director-v2/src/simcore_service_director_v2/modules/db/__init__.py

@@ -1,3 +1,6 @@
+from typing import cast
+
+from aiopg.sa import Engine
 from fastapi import FastAPI
 from settings_library.postgres import PostgresSettings
 
@@ -13,3 +16,7 @@ async def on_shutdown() -> None:
 
     app.add_event_handler("startup", on_startup)
     app.add_event_handler("shutdown", on_shutdown)
+
+
+def get_db_engine(app: FastAPI) -> Engine:
+    return cast(Engine, app.state.engine)

services/director-v2/src/simcore_service_director_v2/modules/db/repositories/comp_runs.py

@@ -1,6 +1,5 @@
 import datetime
 import logging
-from collections import deque
 from typing import Any
 
 import arrow
@@ -55,24 +54,32 @@ async def get(
             return CompRunsAtDB.from_orm(row)
 
     async def list(
-        self, filter_by_state: set[RunningState] | None = None
+        self,
+        filter_by_state: set[RunningState] | None = None,
+        scheduled_since: datetime.timedelta | None = None,
     ) -> list[CompRunsAtDB]:
-        if not filter_by_state:
-            filter_by_state = set()
-        runs_in_db: deque[CompRunsAtDB] = deque()
+
+        conditions = []
+        if filter_by_state:
+            conditions.append(
+                or_(
+                    *[
+                        comp_runs.c.result == RUNNING_STATE_TO_DB[s]
+                        for s in filter_by_state
+                    ]
+                )
+            )
+        if scheduled_since is not None:
+            scheduled_cutoff = arrow.utcnow().datetime - scheduled_since
+            conditions.append(comp_runs.c.last_scheduled >= scheduled_cutoff)
+
         async with self.db_engine.acquire() as conn:
-            async for row in conn.execute(
-                sa.select(comp_runs).where(
-                    or_(
-                        *[
-                            comp_runs.c.result == RUNNING_STATE_TO_DB[s]
-                            for s in filter_by_state
-                        ]
-                    )
+            return [
+                CompRunsAtDB.from_orm(row)
+                async for row in conn.execute(
+                    sa.select(comp_runs).where(sa.and_(*conditions))
                 )
-            ):
-                runs_in_db.append(CompRunsAtDB.from_orm(row))
-        return list(runs_in_db)
+            ]
 
     async def create(
         self,