[DOP-19790] Implement scheduler (#114)

Author: IlyasDevelopment

.github/workflows/unit-test.yml

@@ -56,7 +56,7 @@ jobs:
         run: |
           source .env.local
           poetry run python -m syncmaster.db.migrations upgrade head
-          poetry run coverage run -m pytest -vvv -s -m backend
+          poetry run coverage run -m pytest -vvv -s -m "backend or scheduler"
 
       - name: Shutdown
         if: always()

docker-compose.test.yml

@@ -51,6 +51,25 @@ services:
         condition: service_healthy
     profiles: [backend, all]
 
+  scheduler:
+    image: mtsrus/syncmaster-backend:${BACKEND_IMAGE_TAG:-test}
+    restart: unless-stopped
+    build:
+      dockerfile: docker/Dockerfile.backend
+      context: .
+      target: test
+    env_file: .env.docker
+    volumes:
+      - ./syncmaster:/app/syncmaster
+      - ./pyproject.toml:/app/pyproject.toml
+    depends_on:
+      db:
+        condition: service_healthy
+      rabbitmq:
+        condition: service_healthy
+    entrypoint: [python, -m, syncmaster.scheduler]
+    profiles: [scheduler, all]
+
   worker:
     image: mtsrus/syncmaster-worker:${WORKER_IMAGE_TAG:-test}
     restart: unless-stopped

docs/changelog/next_release/114.feature.rst

@@ -0,0 +1 @@
+Implement a scheduler to run celery tasks on a schedule. This can be done via the `Transfer` table by fields `is_scheduled` and `schedule` (cron-like expression). The Run model now has a `type` field with options `MANUAL` and `SCHEDULED`.

poetry.lock

@@ -66,6 +66,7 @@ uuid6 = "^2024.7.10"
 coloredlogs = {version = "*", optional = true}
 python-json-logger = {version = "*", optional = true}
 asyncpg = { version = ">=0.29,<0.31", optional = true }
+apscheduler = { version = "^3.10.4", optional = true }
 
 [tool.poetry.extras]
 backend = [
@@ -86,6 +87,7 @@ backend = [
     "asyncpg",
     # migrations only
     "celery",
+    "apscheduler",
 ]
 
 worker = [
@@ -101,6 +103,18 @@ worker = [
     "coloredlogs",
     "python-json-logger",
 ]
+
+scheduler = [
+    "apscheduler",
+    "pydantic-settings",
+    "sqlalchemy",
+    "sqlalchemy-utils",
+    "asyncpg",
+    "python-multipart",
+    "python-jose",
+    "celery",
+]
+
 [tool.poetry.group.test.dependencies]
 pandas-stubs = "^2.2.2.240909"
 pytest = "^8.3.3"
@@ -204,6 +218,7 @@ ignore_missing_imports = true
 [tool.pytest.ini_options]
 markers = [
     "backend: tests for backend (require running database)",
+    "scheduler: tests for scheduler (require running database)",
     "worker: tests for syncmaster worker",
     "hive: tests for Hive",
     "postgres: tests on Postgres",

pyproject.toml

@@ -10,7 +10,7 @@
 
 from syncmaster.backend.dependencies import Stub
 from syncmaster.backend.services import UnitOfWork, get_user
-from syncmaster.db.models import Status, User
+from syncmaster.db.models import RunType, Status, User
 from syncmaster.db.utils import Permission
 from syncmaster.errors.registration import get_error_responses
 from syncmaster.exceptions.base import ActionNotAllowedError
@@ -116,6 +116,7 @@ async def start_run(
             # the work of checking fields and removing passwords is delegated to the ReadAuthDataSchema class
             source_creds=ReadAuthDataSchema(auth_data=credentials_source).dict(),
             target_creds=ReadAuthDataSchema(auth_data=credentials_target).dict(),
+            type=RunType.MANUAL,
         )
 
         log_url = Template(settings.worker.LOG_URL_TEMPLATE).render(

syncmaster/backend/api/v1/runs.py

@@ -131,6 +131,8 @@ async def create_transfer(
             target_params=transfer_data.target_params.dict(),
             strategy_params=transfer_data.strategy_params.dict(),
             queue_id=transfer_data.queue_id,
+            is_scheduled=transfer_data.is_scheduled,
+            schedule=transfer_data.schedule,
         )
     return ReadTransferSchema.from_orm(transfer)
 

syncmaster/backend/api/v1/transfers.py

@@ -25,6 +25,7 @@ def upgrade():
         sa.Column("started_at", sa.DateTime(), nullable=True),
         sa.Column("ended_at", sa.DateTime(), nullable=True),
         sa.Column("status", sa.String(255), nullable=False),
+        sa.Column("type", sa.String(64), nullable=False),
         sa.Column("log_url", sa.String(length=512), nullable=True),
         sa.Column("transfer_dump", sa.JSON(), nullable=False),
         sa.Column("created_at", sa.DateTime(), server_default=sa.text("now()"), nullable=False),
@@ -37,11 +38,13 @@ def upgrade():
         ),
         sa.PrimaryKeyConstraint("id", name=op.f("pk__run")),
     )
+    op.create_index(op.f("ix__run__type"), "run", ["type"], unique=False)
     op.create_index(op.f("ix__run__status"), "run", ["status"], unique=False)
     op.create_index(op.f("ix__run__transfer_id"), "run", ["transfer_id"], unique=False)
 
 
 def downgrade():
     op.drop_index(op.f("ix__run__transfer_id"), table_name="run")
     op.drop_index(op.f("ix__run__status"), table_name="run")
+    op.drop_index(op.f("ix__run__type"), table_name="run")
     op.drop_table("run")

syncmaster/db/migrations/versions/2023-11-23_0008_create_run_table.py

@@ -0,0 +1,44 @@
+# SPDX-FileCopyrightText: 2023-2024 MTS PJSC
+# SPDX-License-Identifier: Apache-2.0
+"""Create apscheduler table
+
+Revision ID: 0011
+Revises: 0010
+Create Date: 2024-11-01 08:37:47.078657
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "0011"
+down_revision = "0010"
+branch_labels = None
+depends_on = None
+
+
+def exists_table(table_name: str) -> bool:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    tables = inspector.get_table_names()
+    return table_name in tables
+
+
+def upgrade() -> None:
+    if exists_table("apscheduler_jobs"):
+        return
+
+    op.create_table(
+        "apscheduler_jobs",
+        sa.Column("id", sa.VARCHAR(length=191), autoincrement=False, nullable=False),
+        sa.Column("next_run_time", sa.Float(25), autoincrement=False, nullable=True),
+        sa.Column("job_state", sa.LargeBinary(), autoincrement=False, nullable=False),
+        sa.PrimaryKeyConstraint("id", name="apscheduler_jobs_pkey"),
+    )
+
+    op.create_index("ix_apscheduler_jobs_next_run_time", "apscheduler_jobs", ["next_run_time"], unique=False)
+
+
+def downgrade() -> None:
+    op.drop_index("ix_apscheduler_jobs_next_run_time", table_name="apscheduler_jobs")
+    op.drop_table("apscheduler_jobs")

syncmaster/db/migrations/versions/2024-11-01_0011_create_apscheduler_table.py

@@ -1,10 +1,11 @@
 # SPDX-FileCopyrightText: 2023-2024 MTS PJSC
 # SPDX-License-Identifier: Apache-2.0
+from syncmaster.db.models.apscheduler_job import APSchedulerJob
 from syncmaster.db.models.auth_data import AuthData
 from syncmaster.db.models.base import Base
 from syncmaster.db.models.connection import Connection
 from syncmaster.db.models.group import Group, GroupMemberRole, UserGroup
 from syncmaster.db.models.queue import Queue
-from syncmaster.db.models.run import Run, Status
+from syncmaster.db.models.run import Run, RunType, Status
 from syncmaster.db.models.transfer import Transfer
 from syncmaster.db.models.user import User