Celery scheduler (ITISFoundation#2325)

Author: sanderegg

ci/github/unit-testing/director_v2.bash

@@ -14,7 +14,11 @@ install() {
 test() {
     pytest --numprocesses=auto --cov=simcore_service_director_v2 --durations=10 --cov-append \
           --color=yes --cov-report=term-missing --cov-report=xml --cov-config=.coveragerc \
-          -v -m "not travis" services/director-v2/tests/unit
+          -v -m "not travis" services/director-v2/tests/unit --ignore=services/director-v2/tests/unit/with_dbs;
+    # these tests cannot be run in parallel
+    pytest --cov=simcore_service_director_v2 --durations=10 --cov-append \
+          --color=yes --cov-report=term-missing --cov-report=xml --cov-config=.coveragerc \
+          -v -m "not travis" services/director-v2/tests/unit/with_dbs;
 }
 
 # Check if the function exists (bash specific)

packages/postgres-database/src/simcore_postgres_database/migration/versions/da8abd0d8e42_add_comp_runs_table.py

@@ -0,0 +1,78 @@
+"""add comp runs table
+
+Revision ID: da8abd0d8e42
+Revises: ec3ef74cddae
+Create Date: 2021-05-12 13:46:36.676255+00:00
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "da8abd0d8e42"
+down_revision = "ec3ef74cddae"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "comp_runs",
+        sa.Column(
+            "run_id",
+            sa.BigInteger(),
+            autoincrement=True,
+            nullable=False,
+            primary_key=True,
+        ),
+        sa.Column("project_uuid", sa.String(), nullable=False),
+        sa.Column("user_id", sa.BigInteger(), nullable=False),
+        sa.Column("iteration", sa.BigInteger(), autoincrement=False, nullable=False),
+        sa.Column(
+            "result",
+            sa.dialects.postgresql.ENUM(
+                "NOT_STARTED",
+                "PUBLISHED",
+                "PENDING",
+                "RUNNING",
+                "SUCCESS",
+                "FAILED",
+                "ABORTED",
+                name="statetype",
+                create_type=False,  # necessary to avoid alembic statetype already exists error
+            ),
+            server_default="NOT_STARTED",
+            nullable=False,
+        ),
+        sa.Column(
+            "created", sa.DateTime(), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.Column(
+            "modified", sa.DateTime(), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.Column("started", sa.DateTime(), nullable=True),
+        sa.Column("ended", sa.DateTime(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["project_uuid"],
+            ["projects.uuid"],
+            name="fk_comp_runs_project_uuid_projects",
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["user_id"],
+            ["users.id"],
+            name="fk_comp_runs_user_id_users",
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        sa.UniqueConstraint("project_uuid", "user_id", "iteration"),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("comp_runs")
+    # ### end Alembic commands ###

packages/postgres-database/src/simcore_postgres_database/models/comp_runs.py

@@ -0,0 +1,90 @@
+""" Computational Runs Table
+
+"""
+import sqlalchemy as sa
+from sqlalchemy.sql import func
+
+from .base import metadata
+from .comp_pipeline import StateType
+
+comp_runs = sa.Table(
+    "comp_runs",
+    metadata,
+    # this task db id
+    sa.Column(
+        "run_id",
+        sa.BigInteger,
+        nullable=False,
+        autoincrement=True,
+        primary_key=True,
+        doc="Primary key, identifies the run",
+    ),
+    sa.Column(
+        "project_uuid",
+        sa.String,
+        sa.ForeignKey(
+            "projects.uuid",
+            name="fk_comp_runs_project_uuid_projects",
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        nullable=False,
+        doc="The project uuid with which the run entry is associated",
+    ),
+    sa.Column(
+        "user_id",
+        sa.BigInteger(),
+        sa.ForeignKey(
+            "users.id",
+            name="fk_comp_runs_user_id_users",
+            onupdate="CASCADE",
+            ondelete="CASCADE",
+        ),
+        nullable=False,
+        doc="The user id with which the run entry is associated",
+    ),
+    sa.Column(
+        "iteration",
+        sa.BigInteger,
+        nullable=False,
+        autoincrement=False,
+        doc="A computational run is always associated to a user, a project and a specific iteration",
+    ),
+    sa.Column(
+        "result",
+        sa.Enum(StateType),
+        nullable=False,
+        server_default=StateType.NOT_STARTED.value,
+        doc="The result of the run entry",
+    ),
+    # dag node id and class
+    sa.Column(
+        "created",
+        sa.DateTime(),
+        nullable=False,
+        server_default=func.now(),
+        doc="When the run entry was created",
+    ),
+    sa.Column(
+        "modified",
+        sa.DateTime(),
+        nullable=False,
+        server_default=func.now(),
+        onupdate=func.now(),  # this will auto-update on modification
+        doc="When the run entry was last modified",
+    ),
+    # utc timestamps for submission/start/end
+    sa.Column(
+        "started",
+        sa.DateTime,
+        nullable=True,
+        doc="When the run was started",
+    ),
+    sa.Column(
+        "ended",
+        sa.DateTime,
+        nullable=True,
+        doc="When the run was finished",
+    ),
+    sa.UniqueConstraint("project_uuid", "user_id", "iteration"),
+)

services/api-server/src/simcore_service_api_server/api/dependencies/database.py

@@ -24,7 +24,8 @@ async def _get_repo(
         # the max amount of connections is reached
         # now the current solution is to acquire connection when needed.
 
-        if engine.freesize <= 1:
+        available_engines = engine.maxsize - (engine.size - engine.freesize)
+        if available_engines <= 1:
             logger.warning(
                 "Low pg connections available in pool: pool size=%d, acquired=%d, free=%d, reserved=[%d, %d]",
                 engine.size,

services/director-v2/requirements/_test.in

@@ -10,6 +10,7 @@
 
 
 # testing
+asgi_lifespan
 pytest
 pytest-aiohttp  # incompatible with pytest-asyncio. See https://github.com/pytest-dev/pytest-asyncio/issues/76
 pytest-cov

services/director-v2/requirements/_test.txt

@@ -22,8 +22,14 @@ amqp==5.0.2
     #   kombu
 apipkg==1.5
     # via execnet
+asgi-lifespan==1.0.1
+    # via -r requirements/_test.in
 astroid==2.5.6
     # via pylint
+async-exit-stack==1.0.1 ; python_version < "3.7"
+    # via
+    #   -c requirements/_base.txt
+    #   asgi-lifespan
 async-generator==1.10 ; python_version < "3.7"
     # via
     #   -c requirements/_base.txt
@@ -292,6 +298,7 @@ six==1.15.0
 sniffio==1.2.0
     # via
     #   -c requirements/_base.txt
+    #   asgi-lifespan
     #   httpcore
     #   httpx
 sqlalchemy[postgresql_psycopg2binary]==1.3.20

services/director-v2/src/simcore_service_director_v2/api/dependencies/database.py

@@ -24,7 +24,8 @@ async def _get_repo(
         # the max amount of connections is reached
         # now the current solution is to acquire connection when needed.
 
-        if engine.freesize <= 1:
+        available_engines = engine.maxsize - (engine.size - engine.freesize)
+        if available_engines <= 1:
             logger.warning(
                 "Low pg connections available in pool: pool size=%d, acquired=%d, free=%d, reserved=[%d, %d]",
                 engine.size,

services/director-v2/src/simcore_service_director_v2/api/dependencies/scheduler.py

@@ -0,0 +1,7 @@
+from fastapi import Request
+
+from ...modules.scheduler import CeleryScheduler
+
+
+def get_scheduler(request: Request) -> CeleryScheduler:
+    return request.app.state.scheduler