From 59e7734a083d956617c165646ee2d749a3f6ce1b Mon Sep 17 00:00:00 2001 From: Andrea Piccinelli Date: Thu, 16 Jul 2026 15:13:10 +0200 Subject: [PATCH 1/4] feat(tasks): add dummy job executor --- diracx-tasks/pyproject.toml | 1 + .../src/diracx/tasks/jobs/__init__.py | 2 + .../diracx/tasks/jobs/dummy_job_executor.py | 24 ++++++++++++ diracx-tasks/tests/test_dummy_job_executor.py | 37 +++++++++++++++++++ docs/admin/how-to/tasks/run-task-manually.md | 14 +++++++ 5 files changed, 78 insertions(+) create mode 100644 diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py create mode 100644 diracx-tasks/tests/test_dummy_job_executor.py diff --git a/diracx-tasks/pyproject.toml b/diracx-tasks/pyproject.toml index def809d82..3ad3e0db8 100644 --- a/diracx-tasks/pyproject.toml +++ b/diracx-tasks/pyproject.toml @@ -37,6 +37,7 @@ TaskDB = "diracx.tasks.plumbing.persistence:TaskDB" [project.entry-points."diracx.tasks.jobs"] CleanSandboxStoreTask = "diracx.tasks.jobs:CleanSandboxStoreTask" +DummyJobExecutorTask = "diracx.tasks.jobs:DummyJobExecutorTask" [build-system] requires = ["hatchling", "hatch-vcs"] diff --git a/diracx-tasks/src/diracx/tasks/jobs/__init__.py b/diracx-tasks/src/diracx/tasks/jobs/__init__.py index 3b09465d8..87ae83e3b 100644 --- a/diracx-tasks/src/diracx/tasks/jobs/__init__.py +++ b/diracx-tasks/src/diracx/tasks/jobs/__init__.py @@ -2,6 +2,8 @@ __all__ = [ "CleanSandboxStoreTask", + "DummyJobExecutorTask", ] from .clean_sandbox_store import CleanSandboxStoreTask +from .dummy_job_executor import DummyJobExecutorTask diff --git a/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py b/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py new file mode 100644 index 000000000..69e4063c1 --- /dev/null +++ b/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py @@ -0,0 +1,24 @@ +"""Task that simulates executing a job.""" + +from __future__ import annotations + +import dataclasses +import logging + +from diracx.tasks.plumbing.base_task import BaseTask +from diracx.tasks.plumbing.enums import Size + +logger = logging.getLogger(__name__) + + +@dataclasses.dataclass +class DummyJobExecutorTask(BaseTask): + """Log execution of a single job.""" + + job_id: int + + size = Size.SMALL + + async def execute(self) -> int: + logger.info("I am executing %d", self.job_id) + return self.job_id diff --git a/diracx-tasks/tests/test_dummy_job_executor.py b/diracx-tasks/tests/test_dummy_job_executor.py new file mode 100644 index 000000000..a1f3c3ad9 --- /dev/null +++ b/diracx-tasks/tests/test_dummy_job_executor.py @@ -0,0 +1,37 @@ +"""Tests for the dummy job executor task.""" + +from __future__ import annotations + +import logging + +from diracx.tasks.jobs import DummyJobExecutorTask +from diracx.tasks.jobs.dummy_job_executor import ( + DummyJobExecutorTask as DummyJobExecutorTaskImplementation, +) + + +def test_dummy_job_executor_task_serializes_job_id(): + task = DummyJobExecutorTask(job_id=42) + + assert task.job_id == 42 + assert task.serialize() == (42,) + + +async def test_dummy_job_executor_task_execute(caplog): + task = DummyJobExecutorTask(job_id=42) + + with caplog.at_level(logging.INFO, logger="diracx.tasks.jobs.dummy_job_executor"): + result = await task.execute() + + assert result == 42 + assert caplog.record_tuples == [ + ( + "diracx.tasks.jobs.dummy_job_executor", + logging.INFO, + "I am executing 42", + ) + ] + + +def test_dummy_job_executor_task_is_publicly_exported(): + assert DummyJobExecutorTask is DummyJobExecutorTaskImplementation diff --git a/docs/admin/how-to/tasks/run-task-manually.md b/docs/admin/how-to/tasks/run-task-manually.md index c8422fbb4..037a20f00 100644 --- a/docs/admin/how-to/tasks/run-task-manually.md +++ b/docs/admin/how-to/tasks/run-task-manually.md @@ -23,6 +23,20 @@ diracx-task-run call lollygag:SyncOwnersTask --args '["alice"]' diracx-task-run call lollygag:SyncOwnersTask --args '["alice"]' --kwargs '{}' ``` +## Run the dummy job executor + +Pass a job ID to the registered `jobs:DummyJobExecutorTask` entry point: + +```bash +diracx-task-run call jobs:DummyJobExecutorTask --args '[42]' +``` + +The task logs: + +```text +I am executing 42 +``` + ## Debugging The `--debugger` flag drops into Python's debugger: From c556409c1b34c6a7b4a8e9702ec70796a3ba3aba Mon Sep 17 00:00:00 2001 From: Ryunosuke O'Neil Date: Fri, 17 Jul 2026 13:27:37 +0200 Subject: [PATCH 2/4] feat(tasks): fully implement the dummy job executor Replace the logging-only stub with a working pair of tasks: - DummyJobExecutorMonitorTask: periodic task (every 10s) that moves Received jobs to Waiting and schedules a one-shot executor per job. - DummyJobExecutorTask: simulates the execution of a single job by walking it through the state machine's valid path Waiting -> Matched -> Running -> Done via set_job_statuses, holding a per-job mutex lock. --- diracx-tasks/pyproject.toml | 1 + .../src/diracx/tasks/jobs/__init__.py | 3 +- .../diracx/tasks/jobs/dummy_job_executor.py | 135 +++++++++++++++++- diracx-tasks/tests/test_dummy_job_executor.py | 123 +++++++++++++--- docs/admin/how-to/tasks/run-task-manually.md | 18 ++- 5 files changed, 248 insertions(+), 32 deletions(-) diff --git a/diracx-tasks/pyproject.toml b/diracx-tasks/pyproject.toml index 3ad3e0db8..c70716a1e 100644 --- a/diracx-tasks/pyproject.toml +++ b/diracx-tasks/pyproject.toml @@ -37,6 +37,7 @@ TaskDB = "diracx.tasks.plumbing.persistence:TaskDB" [project.entry-points."diracx.tasks.jobs"] CleanSandboxStoreTask = "diracx.tasks.jobs:CleanSandboxStoreTask" +DummyJobExecutorMonitorTask = "diracx.tasks.jobs:DummyJobExecutorMonitorTask" DummyJobExecutorTask = "diracx.tasks.jobs:DummyJobExecutorTask" [build-system] diff --git a/diracx-tasks/src/diracx/tasks/jobs/__init__.py b/diracx-tasks/src/diracx/tasks/jobs/__init__.py index 87ae83e3b..60c37cf41 100644 --- a/diracx-tasks/src/diracx/tasks/jobs/__init__.py +++ b/diracx-tasks/src/diracx/tasks/jobs/__init__.py @@ -2,8 +2,9 @@ __all__ = [ "CleanSandboxStoreTask", + "DummyJobExecutorMonitorTask", "DummyJobExecutorTask", ] from .clean_sandbox_store import CleanSandboxStoreTask -from .dummy_job_executor import DummyJobExecutorTask +from .dummy_job_executor import DummyJobExecutorMonitorTask, DummyJobExecutorTask diff --git a/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py b/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py index 69e4063c1..1b3624f65 100644 --- a/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py +++ b/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py @@ -1,24 +1,145 @@ -"""Task that simulates executing a job.""" +"""Tasks that simulate the execution of jobs, for use in demo deployments.""" from __future__ import annotations import dataclasses import logging +from datetime import UTC, datetime, timedelta -from diracx.tasks.plumbing.base_task import BaseTask -from diracx.tasks.plumbing.enums import Size +from diracx.core.models import ( + JobStatus, + JobStatusUpdate, + ScalarSearchOperator, + ScalarSearchSpec, +) +from diracx.db.os import JobParametersDB +from diracx.db.sql import JobDB, JobLoggingDB, TaskQueueDB +from diracx.logic.jobs import set_job_statuses +from diracx.tasks.plumbing.base_task import BaseTask, PeriodicBaseTask +from diracx.tasks.plumbing.depends import Config +from diracx.tasks.plumbing.enums import Priority, Size +from diracx.tasks.plumbing.lock_registry import JOB +from diracx.tasks.plumbing.locks import BaseLock, MutexLock +from diracx.tasks.plumbing.retry_policies import ExponentialBackoff +from diracx.tasks.plumbing.schedules import IntervalSeconds logger = logging.getLogger(__name__) +MINOR_STATUS = "DummyExecutor" + + +@dataclasses.dataclass +class DummyJobExecutorMonitorTask(PeriodicBaseTask): + """Periodically pick up newly received jobs and hand them to the dummy executor. + + Every job in ``Received`` state is moved to ``Waiting`` and a one-shot + ``DummyJobExecutorTask`` is scheduled for it. + """ + + priority = Priority.BACKGROUND + size = Size.MEDIUM + retry_policy = ExponentialBackoff(base_delay_seconds=300, max_retries=3) + default_schedule = IntervalSeconds(10) + + async def execute( + self, + config: Config, + job_db: JobDB, + job_logging_db: JobLoggingDB, + task_queue_db: TaskQueueDB, + job_parameters_db: JobParametersDB, + ) -> int: + _, jobs = await job_db.search( + ["JobID"], + [ + ScalarSearchSpec( + parameter="Status", + operator=ScalarSearchOperator.EQUAL, + value=JobStatus.RECEIVED, + ) + ], + [], + ) + if not jobs: + return 0 + + job_ids = [job["JobID"] for job in jobs] + logger.info("Moving %d received job(s) to Waiting: %s", len(job_ids), job_ids) + await set_job_statuses( + { + job_id: { + datetime.now(UTC): JobStatusUpdate( + Status=JobStatus.WAITING, + MinorStatus=MINOR_STATUS, + ) + } + for job_id in job_ids + }, + config=config, + job_db=job_db, + job_logging_db=job_logging_db, + task_queue_db=task_queue_db, + job_parameters_db=job_parameters_db, + ) + + for job_id in job_ids: + await DummyJobExecutorTask(job_id=job_id).schedule() + + return len(job_ids) + @dataclasses.dataclass class DummyJobExecutorTask(BaseTask): - """Log execution of a single job.""" + """Simulate the execution of a single job. - job_id: int + The job is walked through the state machine's valid path + ``Waiting -> Matched -> Running -> Done``: a direct jump to ``Done`` would be + silently rejected. ``set_job_statuses`` applies the timestamped updates in + chronological order, so a single call with increasing timestamps walks the + whole chain. + """ + priority = Priority.NORMAL size = Size.SMALL + retry_policy = ExponentialBackoff(base_delay_seconds=300, max_retries=3) + + job_id: int + + @property + def execution_locks(self) -> list[BaseLock]: + return [MutexLock(JOB, self.job_id)] - async def execute(self) -> int: - logger.info("I am executing %d", self.job_id) + async def execute( + self, + config: Config, + job_db: JobDB, + job_logging_db: JobLoggingDB, + task_queue_db: TaskQueueDB, + job_parameters_db: JobParametersDB, + ) -> int: + logger.info("Simulating execution of job %d", self.job_id) + now = datetime.now(UTC) + await set_job_statuses( + { + self.job_id: { + now: JobStatusUpdate( + Status=JobStatus.MATCHED, + MinorStatus=MINOR_STATUS, + ), + now + timedelta(seconds=1): JobStatusUpdate( + Status=JobStatus.RUNNING, + MinorStatus=MINOR_STATUS, + ), + now + timedelta(seconds=5): JobStatusUpdate( + Status=JobStatus.DONE, + MinorStatus=MINOR_STATUS, + ), + } + }, + config=config, + job_db=job_db, + job_logging_db=job_logging_db, + task_queue_db=task_queue_db, + job_parameters_db=job_parameters_db, + ) return self.job_id diff --git a/diracx-tasks/tests/test_dummy_job_executor.py b/diracx-tasks/tests/test_dummy_job_executor.py index a1f3c3ad9..4fba2a405 100644 --- a/diracx-tasks/tests/test_dummy_job_executor.py +++ b/diracx-tasks/tests/test_dummy_job_executor.py @@ -1,37 +1,122 @@ -"""Tests for the dummy job executor task.""" +"""Tests for the dummy job executor tasks.""" from __future__ import annotations -import logging +from unittest.mock import AsyncMock, MagicMock -from diracx.tasks.jobs import DummyJobExecutorTask -from diracx.tasks.jobs.dummy_job_executor import ( - DummyJobExecutorTask as DummyJobExecutorTaskImplementation, -) +from diracx.core.models import JobStatus +from diracx.tasks.jobs import DummyJobExecutorMonitorTask, DummyJobExecutorTask +from diracx.tasks.jobs import dummy_job_executor as dummy_job_executor_module +from diracx.tasks.plumbing.locks import MutexLock +from diracx.tasks.plumbing.schedules import IntervalSeconds -def test_dummy_job_executor_task_serializes_job_id(): +def make_dependencies(): + return { + "config": MagicMock(name="config"), + "job_db": AsyncMock(name="job_db"), + "job_logging_db": MagicMock(name="job_logging_db"), + "task_queue_db": MagicMock(name="task_queue_db"), + "job_parameters_db": MagicMock(name="job_parameters_db"), + } + + +def test_executor_serializes_job_id(): task = DummyJobExecutorTask(job_id=42) assert task.job_id == 42 assert task.serialize() == (42,) -async def test_dummy_job_executor_task_execute(caplog): - task = DummyJobExecutorTask(job_id=42) +def test_monitor_serializes_to_empty_tuple(): + assert DummyJobExecutorMonitorTask().serialize() == () + + +def test_monitor_runs_periodically(): + schedule = DummyJobExecutorMonitorTask.default_schedule + + assert isinstance(schedule, IntervalSeconds) + assert schedule.seconds == 10 + + +def test_executor_takes_a_per_job_mutex(): + locks = DummyJobExecutorTask(job_id=42).execution_locks + + assert len(locks) == 1 + assert isinstance(locks[0], MutexLock) + assert locks[0].redis_key == "lock:mutex:job:42" + + +async def test_executor_walks_job_through_the_state_machine(monkeypatch): + set_job_statuses = AsyncMock() + monkeypatch.setattr(dummy_job_executor_module, "set_job_statuses", set_job_statuses) + deps = make_dependencies() - with caplog.at_level(logging.INFO, logger="diracx.tasks.jobs.dummy_job_executor"): - result = await task.execute() + result = await DummyJobExecutorTask(job_id=42).execute(**deps) assert result == 42 - assert caplog.record_tuples == [ - ( - "diracx.tasks.jobs.dummy_job_executor", - logging.INFO, - "I am executing 42", - ) + set_job_statuses.assert_awaited_once() + status_changes = set_job_statuses.await_args.args[0] + assert set(status_changes) == {42} + updates = status_changes[42] + assert list(updates) == sorted(updates), "timestamps must be increasing" + assert [update.status for update in updates.values()] == [ + JobStatus.MATCHED, + JobStatus.RUNNING, + JobStatus.DONE, ] + assert set_job_statuses.await_args.kwargs == { + "config": deps["config"], + "job_db": deps["job_db"], + "job_logging_db": deps["job_logging_db"], + "task_queue_db": deps["task_queue_db"], + "job_parameters_db": deps["job_parameters_db"], + } + + +async def test_monitor_moves_received_jobs_and_schedules_executors(monkeypatch): + set_job_statuses = AsyncMock() + monkeypatch.setattr(dummy_job_executor_module, "set_job_statuses", set_job_statuses) + scheduled = [] + + async def fake_schedule(self, **kwargs): + scheduled.append(self.job_id) + return "task-id" + + monkeypatch.setattr(DummyJobExecutorTask, "schedule", fake_schedule) + deps = make_dependencies() + deps["job_db"].search.return_value = (2, [{"JobID": 1}, {"JobID": 2}]) + + result = await DummyJobExecutorMonitorTask().execute(**deps) + + assert result == 2 + deps["job_db"].search.assert_awaited_once() + (search_spec,) = deps["job_db"].search.await_args.args[1] + assert search_spec["parameter"] == "Status" + assert search_spec["value"] == JobStatus.RECEIVED + set_job_statuses.assert_awaited_once() + status_changes = set_job_statuses.await_args.args[0] + assert set(status_changes) == {1, 2} + for updates in status_changes.values(): + assert [update.status for update in updates.values()] == [JobStatus.WAITING] + assert scheduled == [1, 2] + + +async def test_monitor_does_nothing_without_received_jobs(monkeypatch): + set_job_statuses = AsyncMock() + monkeypatch.setattr(dummy_job_executor_module, "set_job_statuses", set_job_statuses) + deps = make_dependencies() + deps["job_db"].search.return_value = (0, []) + + result = await DummyJobExecutorMonitorTask().execute(**deps) + + assert result == 0 + set_job_statuses.assert_not_awaited() -def test_dummy_job_executor_task_is_publicly_exported(): - assert DummyJobExecutorTask is DummyJobExecutorTaskImplementation +def test_tasks_are_publicly_exported(): + assert DummyJobExecutorTask is dummy_job_executor_module.DummyJobExecutorTask + assert ( + DummyJobExecutorMonitorTask + is dummy_job_executor_module.DummyJobExecutorMonitorTask + ) diff --git a/docs/admin/how-to/tasks/run-task-manually.md b/docs/admin/how-to/tasks/run-task-manually.md index 037a20f00..6b6c37989 100644 --- a/docs/admin/how-to/tasks/run-task-manually.md +++ b/docs/admin/how-to/tasks/run-task-manually.md @@ -25,16 +25,24 @@ diracx-task-run call lollygag:SyncOwnersTask --args '["alice"]' --kwargs '{}' ## Run the dummy job executor -Pass a job ID to the registered `jobs:DummyJobExecutorTask` entry point: +Demo deployments include two tasks that simulate job execution. The periodic +`jobs:DummyJobExecutorMonitorTask` moves every `Received` job to `Waiting` and +schedules a one-shot `jobs:DummyJobExecutorTask` for it, which walks the job +through `Matched` → `Running` → `Done`. + +Both tasks talk to the job databases, so the relevant `DIRACX_DB_URL_*`, +`DIRACX_OS_DB_*`, and `DIRACX_CONFIG_BACKEND_URL` variables must be set. + +Run the monitor once to pick up all `Received` jobs: ```bash -diracx-task-run call jobs:DummyJobExecutorTask --args '[42]' +diracx-task-run call jobs:DummyJobExecutorMonitorTask ``` -The task logs: +Or simulate the execution of a single job by passing its job ID: -```text -I am executing 42 +```bash +diracx-task-run call jobs:DummyJobExecutorTask --args '[42]' ``` ## Debugging From c9d9a31d1e27bd5831db0ecfa96f8f12cc61d34d Mon Sep 17 00:00:00 2001 From: Andrea Piccinelli Date: Wed, 22 Jul 2026 11:28:10 +0200 Subject: [PATCH 3/4] fix(tasks): address dummy executor review feedback --- .../diracx/tasks/jobs/dummy_job_executor.py | 35 ++++-- diracx-tasks/tests/test_dummy_job_executor.py | 103 +++++++++++++----- docs/admin/how-to/tasks/run-task-manually.md | 9 +- run_local.sh | 2 + 4 files changed, 114 insertions(+), 35 deletions(-) diff --git a/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py b/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py index 1b3624f65..400e8170d 100644 --- a/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py +++ b/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py @@ -6,12 +6,15 @@ import logging from datetime import UTC, datetime, timedelta +from pydantic import PositiveInt + from diracx.core.models import ( JobStatus, JobStatusUpdate, ScalarSearchOperator, ScalarSearchSpec, ) +from diracx.core.settings import ServiceSettingsBase from diracx.db.os import JobParametersDB from diracx.db.sql import JobDB, JobLoggingDB, TaskQueueDB from diracx.logic.jobs import set_job_statuses @@ -28,18 +31,36 @@ MINOR_STATUS = "DummyExecutor" +class _DummyJobExecutorSettings(ServiceSettingsBase): + """Settings controlling automatic dummy job execution.""" + + model_config = ServiceSettingsBase.model_config | { + "env_prefix": "DIRACX_TASKS_DUMMY_JOB_EXECUTOR_", + "use_attribute_docstrings": True, + } + + enabled: bool = False + """Whether the monitor is scheduled automatically.""" + + interval_seconds: PositiveInt = 10 + """How often the enabled monitor searches for received jobs.""" + + +_settings = _DummyJobExecutorSettings() + + @dataclasses.dataclass class DummyJobExecutorMonitorTask(PeriodicBaseTask): """Periodically pick up newly received jobs and hand them to the dummy executor. - Every job in ``Received`` state is moved to ``Waiting`` and a one-shot - ``DummyJobExecutorTask`` is scheduled for it. + When enabled, every job in ``Received`` state is moved to ``Waiting`` and a + one-shot ``DummyJobExecutorTask`` is scheduled for it. """ priority = Priority.BACKGROUND - size = Size.MEDIUM - retry_policy = ExponentialBackoff(base_delay_seconds=300, max_retries=3) - default_schedule = IntervalSeconds(10) + size = Size.SMALL + _enabled = _settings.enabled + default_schedule = IntervalSeconds(_settings.interval_seconds) async def execute( self, @@ -100,8 +121,8 @@ class DummyJobExecutorTask(BaseTask): """ priority = Priority.NORMAL - size = Size.SMALL - retry_policy = ExponentialBackoff(base_delay_seconds=300, max_retries=3) + size = Size.LARGE + retry_policy = ExponentialBackoff(base_delay_seconds=10, max_retries=3) job_id: int diff --git a/diracx-tasks/tests/test_dummy_job_executor.py b/diracx-tasks/tests/test_dummy_job_executor.py index 4fba2a405..d63ded5e9 100644 --- a/diracx-tasks/tests/test_dummy_job_executor.py +++ b/diracx-tasks/tests/test_dummy_job_executor.py @@ -2,13 +2,53 @@ from __future__ import annotations +import json +import os +import subprocess +import sys from unittest.mock import AsyncMock, MagicMock +import pytest +from pydantic import ValidationError + from diracx.core.models import JobStatus from diracx.tasks.jobs import DummyJobExecutorMonitorTask, DummyJobExecutorTask from diracx.tasks.jobs import dummy_job_executor as dummy_job_executor_module from diracx.tasks.plumbing.locks import MutexLock -from diracx.tasks.plumbing.schedules import IntervalSeconds + +FEATURE_ENABLED_ENV = "DIRACX_TASKS_DUMMY_JOB_EXECUTOR_ENABLED" +FEATURE_INTERVAL_ENV = "DIRACX_TASKS_DUMMY_JOB_EXECUTOR_INTERVAL_SECONDS" +SCHEDULER_STATE_SCRIPT = """ +import json +from datetime import UTC, datetime +from unittest.mock import MagicMock + +from diracx.tasks.plumbing.factory import load_task_registry +from diracx.tasks.plumbing.scheduler import TaskScheduler + +task_name = "jobs:DummyJobExecutorMonitorTask" +registry = load_task_registry() +task_cls = registry[task_name] +scheduler = TaskScheduler( + broker=MagicMock(), + redis_url="redis://unused", + task_registry=registry, +) +before = datetime.now(tz=UTC) +scheduler._compute_initial_schedules() +next_run = scheduler._next_runs.get((task_name, "")) +print( + json.dumps( + { + "enabled": task_cls._enabled, + "tracked": next_run is not None, + "delay_seconds": ( + (next_run - before).total_seconds() if next_run is not None else None + ), + } + ) +) +""" def make_dependencies(): @@ -21,22 +61,43 @@ def make_dependencies(): } -def test_executor_serializes_job_id(): - task = DummyJobExecutorTask(job_id=42) - - assert task.job_id == 42 - assert task.serialize() == (42,) +def get_scheduler_state(feature_env: dict[str, str]) -> dict: + env = os.environ.copy() + env.pop(FEATURE_ENABLED_ENV, None) + env.pop(FEATURE_INTERVAL_ENV, None) + env.update(feature_env) + result = subprocess.run( + [sys.executable, "-c", SCHEDULER_STATE_SCRIPT], + check=True, + capture_output=True, + env=env, + text=True, + ) + return json.loads(result.stdout) -def test_monitor_serializes_to_empty_tuple(): - assert DummyJobExecutorMonitorTask().serialize() == () +def test_monitor_schedule_activation_is_environment_controlled(): + default_state = get_scheduler_state({}) + assert default_state == { + "enabled": False, + "tracked": False, + "delay_seconds": None, + } + local_state = get_scheduler_state( + { + FEATURE_ENABLED_ENV: "true", + FEATURE_INTERVAL_ENV: "10", + } + ) + assert local_state["enabled"] is True + assert local_state["tracked"] is True + assert 9 <= local_state["delay_seconds"] <= 11 -def test_monitor_runs_periodically(): - schedule = DummyJobExecutorMonitorTask.default_schedule - assert isinstance(schedule, IntervalSeconds) - assert schedule.seconds == 10 +def test_monitor_interval_must_be_positive(): + with pytest.raises(ValidationError): + dummy_job_executor_module._DummyJobExecutorSettings(interval_seconds=0) def test_executor_takes_a_per_job_mutex(): @@ -65,13 +126,6 @@ async def test_executor_walks_job_through_the_state_machine(monkeypatch): JobStatus.RUNNING, JobStatus.DONE, ] - assert set_job_statuses.await_args.kwargs == { - "config": deps["config"], - "job_db": deps["job_db"], - "job_logging_db": deps["job_logging_db"], - "task_queue_db": deps["task_queue_db"], - "job_parameters_db": deps["job_parameters_db"], - } async def test_monitor_moves_received_jobs_and_schedules_executors(monkeypatch): @@ -105,6 +159,8 @@ async def fake_schedule(self, **kwargs): async def test_monitor_does_nothing_without_received_jobs(monkeypatch): set_job_statuses = AsyncMock() monkeypatch.setattr(dummy_job_executor_module, "set_job_statuses", set_job_statuses) + schedule_executor = AsyncMock() + monkeypatch.setattr(DummyJobExecutorTask, "schedule", schedule_executor) deps = make_dependencies() deps["job_db"].search.return_value = (0, []) @@ -112,11 +168,4 @@ async def test_monitor_does_nothing_without_received_jobs(monkeypatch): assert result == 0 set_job_statuses.assert_not_awaited() - - -def test_tasks_are_publicly_exported(): - assert DummyJobExecutorTask is dummy_job_executor_module.DummyJobExecutorTask - assert ( - DummyJobExecutorMonitorTask - is dummy_job_executor_module.DummyJobExecutorMonitorTask - ) + schedule_executor.assert_not_awaited() diff --git a/docs/admin/how-to/tasks/run-task-manually.md b/docs/admin/how-to/tasks/run-task-manually.md index 6b6c37989..32d57db36 100644 --- a/docs/admin/how-to/tasks/run-task-manually.md +++ b/docs/admin/how-to/tasks/run-task-manually.md @@ -25,11 +25,18 @@ diracx-task-run call lollygag:SyncOwnersTask --args '["alice"]' --kwargs '{}' ## Run the dummy job executor -Demo deployments include two tasks that simulate job execution. The periodic +The task package includes two tasks that simulate job execution. The periodic `jobs:DummyJobExecutorMonitorTask` moves every `Received` job to `Waiting` and schedules a one-shot `jobs:DummyJobExecutorTask` for it, which walks the job through `Matched` → `Running` → `Done`. +Automatic monitoring is disabled by default. `run_local.sh` explicitly enables +the monitor every 10 seconds with +`DIRACX_TASKS_DUMMY_JOB_EXECUTOR_ENABLED=true` and +`DIRACX_TASKS_DUMMY_JOB_EXECUTOR_INTERVAL_SECONDS=10`. Demo deployment +enablement requires the coordinated `diracx-charts` values change and is not +part of this commit. + Both tasks talk to the job databases, so the relevant `DIRACX_DB_URL_*`, `DIRACX_OS_DB_*`, and `DIRACX_CONFIG_BACKEND_URL` variables must be set. diff --git a/run_local.sh b/run_local.sh index 51f4f2c50..a6d405c13 100755 --- a/run_local.sh +++ b/run_local.sh @@ -98,6 +98,8 @@ export DIRACX_SANDBOX_STORE_BUCKET_NAME=sandboxes export DIRACX_SANDBOX_STORE_AUTO_CREATE_BUCKET=true export DIRACX_SANDBOX_STORE_S3_CLIENT_KWARGS='{"endpoint_url": "http://localhost:8333", "aws_access_key_id": "console", "aws_secret_access_key": "console123"}' export DIRACX_TASKS_REDIS_URL="redis://localhost:6379" +export DIRACX_TASKS_DUMMY_JOB_EXECUTOR_ENABLED=true +export DIRACX_TASKS_DUMMY_JOB_EXECUTOR_INTERVAL_SECONDS=10 # Write all DIRACX env vars to a sourceable file for use in other terminals script_dir="$(cd "$(dirname "$0")" && pwd)" From a0b49e2fa15f0171e50d82ce68ab6b9589641fb8 Mon Sep 17 00:00:00 2001 From: Andrea Piccinelli Date: Wed, 22 Jul 2026 16:52:18 +0200 Subject: [PATCH 4/4] fix(tasks): expose dummy executor settings --- .../diracx/tasks/jobs/dummy_job_executor.py | 4 ++-- diracx-tasks/tests/test_dummy_job_executor.py | 2 +- docs/admin/reference/env-variables.md | 18 ++++++++++++++++++ docs/admin/reference/env-variables.md.j2 | 4 ++++ 4 files changed, 25 insertions(+), 3 deletions(-) diff --git a/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py b/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py index 400e8170d..cd9a72991 100644 --- a/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py +++ b/diracx-tasks/src/diracx/tasks/jobs/dummy_job_executor.py @@ -31,7 +31,7 @@ MINOR_STATUS = "DummyExecutor" -class _DummyJobExecutorSettings(ServiceSettingsBase): +class DummyJobExecutorSettings(ServiceSettingsBase): """Settings controlling automatic dummy job execution.""" model_config = ServiceSettingsBase.model_config | { @@ -46,7 +46,7 @@ class _DummyJobExecutorSettings(ServiceSettingsBase): """How often the enabled monitor searches for received jobs.""" -_settings = _DummyJobExecutorSettings() +_settings = DummyJobExecutorSettings() @dataclasses.dataclass diff --git a/diracx-tasks/tests/test_dummy_job_executor.py b/diracx-tasks/tests/test_dummy_job_executor.py index d63ded5e9..6166a6187 100644 --- a/diracx-tasks/tests/test_dummy_job_executor.py +++ b/diracx-tasks/tests/test_dummy_job_executor.py @@ -97,7 +97,7 @@ def test_monitor_schedule_activation_is_environment_controlled(): def test_monitor_interval_must_be_positive(): with pytest.raises(ValidationError): - dummy_job_executor_module._DummyJobExecutorSettings(interval_seconds=0) + dummy_job_executor_module.DummyJobExecutorSettings(interval_seconds=0) def test_executor_takes_a_per_job_mutex(): diff --git a/docs/admin/reference/env-variables.md b/docs/admin/reference/env-variables.md index e2b2b6285..2e2a37956 100644 --- a/docs/admin/reference/env-variables.md +++ b/docs/admin/reference/env-variables.md @@ -363,3 +363,21 @@ Whether to use an insecure gRPC connection for the OpenTelemetry collector. *Optional*, default value: `None` A JSON-encoded dictionary of headers to pass to the OpenTelemetry collector, e.g. {"tenant_id": "lhcbdiracx-cert"}. + +## Tasks + +## DummyJobExecutorSettings + +Settings controlling automatic dummy job execution. + +### `DIRACX_TASKS_DUMMY_JOB_EXECUTOR_ENABLED` + +*Optional*, default value: `False` + +Whether the monitor is scheduled automatically. + +### `DIRACX_TASKS_DUMMY_JOB_EXECUTOR_INTERVAL_SECONDS` + +*Optional*, default value: `10` + +How often the enabled monitor searches for received jobs. diff --git a/docs/admin/reference/env-variables.md.j2 b/docs/admin/reference/env-variables.md.j2 index 8f17665e1..4630da850 100644 --- a/docs/admin/reference/env-variables.md.j2 +++ b/docs/admin/reference/env-variables.md.j2 @@ -18,3 +18,7 @@ _X, where X is a number. The files will be loaded in order. {{ render_class('SandboxStoreSettings') }} {{ render_class('OTELSettings') }} + +## Tasks + +{{ render_class('DummyJobExecutorSettings') }}