From cb80d420c0e21989f7e89b33a19282a3f46f82cd Mon Sep 17 00:00:00 2001 From: ajp_anton Date: Mon, 24 Aug 2026 15:40:31 +0000 Subject: [PATCH] Simplify python-tools task configuration --- python-tools/README.md | 27 ++++++-- python-tools/app/runner.py | 107 ++++++++++++++++++++---------- python-tools/compose.example.yaml | 3 - python-tools/tests/test_runner.py | 45 +++++++++++-- 4 files changed, 135 insertions(+), 47 deletions(-) diff --git a/python-tools/README.md b/python-tools/README.md index baeeb95..dcdccb6 100644 --- a/python-tools/README.md +++ b/python-tools/README.md @@ -18,10 +18,17 @@ The image contains application code only. Mount these paths at runtime: - `/opt/credentials` read-only: named credential files used by those scripts. - `/var/lib/server-maintenance`: persistent SQLite data, logs, and task state. -Task processes receive a minimal environment. To forward non-secret container -settings to tasks, list their names in `SERVER_MAINTENANCE_TASK_ENV`, separated -by commas. This allowlist prevents unrelated container configuration from -being exposed to every task; credentials should remain in `/opt/credentials`. +Each group's durable task data lives under +`/var/lib/server-maintenance/tasks/`. When a task group disappears +from the mounted task directory, its state is retained for 14 days and then +removed automatically. Tasks that need more focused retention should clean up +their own state within that group. + +Task processes receive only runner-provided paths for input, artifacts, +credentials, and state. Keep task-specific non-secret configuration in an +optional `task-config.json` beside that group's scripts, and credentials in +`/opt/credentials`. The runner does not interpret `task-config.json`; its +schema belongs to the task group that reads it. Tasks are never imported during discovery. Their optional description is the first line of their module docstring. A task must not require stdin or command @@ -122,6 +129,12 @@ are stored in a private per-run directory; their absolute paths are provided in the input JSON and the files are removed when the task finishes, fails, or is cancelled. +A group may also include a private `task-config.json` for stable local details +such as mounted filesystem roots. This is not part of discovery or the web +form schema: task code reads and validates it itself. Keeping it beside the +group makes a task's runtime requirements clear and avoids a global +task-environment allowlist. + ## Artifacts Tasks can write files to the directory named by @@ -154,3 +167,9 @@ stack. Set `PYTHON_TOOLS_VOLUME_ROOT` to the host directory that will hold `state`, `tasks`, and `credentials`; set `SERVER_MAINTENANCE_ALLOWED_PROXY_IPS` to the reverse proxy's direct peer address. Keep the application behind an authenticated reverse proxy. + +Tasks receive a group-specific writable directory in +`SERVER_MAINTENANCE_STATE`. Tasks which intentionally maintain application-wide +state, such as run-history maintenance, can use +`SERVER_MAINTENANCE_STATE_ROOT`; normal task state should remain +group-specific. diff --git a/python-tools/app/runner.py b/python-tools/app/runner.py index 5f08b68..a48a5a9 100644 --- a/python-tools/app/runner.py +++ b/python-tools/app/runner.py @@ -18,6 +18,10 @@ from .inputs import PendingUpload from .store import Run, RunStore +ORPHANED_TASK_STATE_FILENAME = ".orphaned-groups.json" +ORPHANED_TASK_STATE_RETENTION_SECONDS = 14 * 86400 + + class RunManager: def __init__(self, settings: Settings, catalog: TaskCatalog, store: RunStore) -> None: self.settings = settings @@ -112,36 +116,78 @@ class RunManager: except OSError: pass - def cleanup_transient_files(self, *, artifact_max_age_seconds: int = 86_400) -> None: - """Remove uploads from interrupted runs and expired undownloaded artifacts.""" + def cleanup_transient_files( + self, + *, + artifact_max_age_seconds: int = 86_400, + orphaned_task_state_retention_seconds: int = ORPHANED_TASK_STATE_RETENTION_SECONDS, + ) -> None: + """Remove transient run data and state for task groups removed long ago.""" runs_root = self.settings.state_root / "runs" - if not runs_root.is_dir(): - return cutoff = time.time() - artifact_max_age_seconds - for path in runs_root.iterdir(): - if not path.is_dir(): - continue - run_id_text, separator, kind = path.name.partition(".") - if not separator or not run_id_text.isdigit() or kind not in {"uploads", "artifacts"}: - continue - run = self.store.get(int(run_id_text)) - if kind == "uploads" and (run is None or run.status not in {"queued", "running"}): - shutil.rmtree(path, ignore_errors=True) - elif kind == "artifacts": - try: - expired = path.stat().st_mtime < cutoff - except FileNotFoundError: + if runs_root.is_dir(): + for path in runs_root.iterdir(): + if not path.is_dir(): continue - if expired: + run_id_text, separator, kind = path.name.partition(".") + if not separator or not run_id_text.isdigit() or kind not in {"uploads", "artifacts"}: + continue + run = self.store.get(int(run_id_text)) + if kind == "uploads" and (run is None or run.status not in {"queued", "running"}): shutil.rmtree(path, ignore_errors=True) - for path in runs_root.glob("*.execution-input.json"): - run_id_text = path.name.removesuffix(".execution-input.json") - if not run_id_text.isdigit(): + elif kind == "artifacts": + try: + expired = path.stat().st_mtime < cutoff + except FileNotFoundError: + continue + if expired: + shutil.rmtree(path, ignore_errors=True) + for path in runs_root.glob("*.execution-input.json"): + run_id_text = path.name.removesuffix(".execution-input.json") + if not run_id_text.isdigit(): + continue + run = self.store.get(int(run_id_text)) + if run is None or run.status not in {"queued", "running"}: + path.unlink(missing_ok=True) + self._cleanup_orphaned_task_state(orphaned_task_state_retention_seconds) + + def _cleanup_orphaned_task_state(self, retention_seconds: int) -> None: + """Delete state only after a group remains absent for a grace period.""" + + if retention_seconds < 0 or not self.settings.task_root.is_dir(): + return + root = self.settings.state_root / "tasks" + if not root.is_dir(): + return + marker_path = root / ORPHANED_TASK_STATE_FILENAME + try: + markers = json.loads(marker_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + markers = {} + if not isinstance(markers, dict): + markers = {} + active_groups = {group.id for group in self.catalog.groups()} + current_time = time.time() + updated: dict[str, float] = {} + for path in root.iterdir(): + if path.is_symlink() or not path.is_dir(): continue - run = self.store.get(int(run_id_text)) - if run is None or run.status not in {"queued", "running"}: - path.unlink(missing_ok=True) + if path.name in active_groups: + continue + first_missing_at = markers.get(path.name, current_time) + if not isinstance(first_missing_at, (int, float)): + first_missing_at = current_time + if current_time - float(first_missing_at) >= retention_seconds: + shutil.rmtree(path, ignore_errors=True) + else: + updated[path.name] = float(first_missing_at) + if updated: + temporary = marker_path.with_suffix(marker_path.suffix + ".tmp") + temporary.write_text(json.dumps(updated, sort_keys=True) + "\n", encoding="utf-8") + temporary.replace(marker_path) + else: + marker_path.unlink(missing_ok=True) def run_background_task(self, task: BackgroundTask) -> None: """Run a declared internal task without creating high-volume run history.""" @@ -335,22 +381,11 @@ class RunManager: "PATH": "/usr/local/bin:/usr/bin:/bin", "PYTHONUNBUFFERED": "1", "SERVER_MAINTENANCE_CREDENTIALS": str(self.settings.credentials_root), + "SERVER_MAINTENANCE_STATE_ROOT": str(self.settings.state_root), "SERVER_MAINTENANCE_STATE": str(task_state), } if self.settings.timezone_name: environment["TZ"] = self.settings.timezone_name - for name in os.environ.get("SERVER_MAINTENANCE_TASK_ENV", "").split(","): - name = name.strip() - if not name: - continue - if not name.isidentifier() or name.startswith("SERVER_MAINTENANCE_"): - raise ValueError( - "SERVER_MAINTENANCE_TASK_ENV entries must be environment variable names " - "outside the SERVER_MAINTENANCE_ namespace" - ) - value = os.environ.get(name) - if value is not None: - environment[name] = value return environment def _upload_root(self, run_id: int) -> Path: diff --git a/python-tools/compose.example.yaml b/python-tools/compose.example.yaml index 7aaeea7..ca67f2f 100644 --- a/python-tools/compose.example.yaml +++ b/python-tools/compose.example.yaml @@ -7,9 +7,6 @@ services: - "${PYTHON_TOOLS_PORT:-8080}:8080" environment: SERVER_MAINTENANCE_ALLOWED_PROXY_IPS: "${SERVER_MAINTENANCE_ALLOWED_PROXY_IPS:?Set the reverse proxy's direct peer address}" - # Optional task settings. Only names listed here are forwarded to task - # processes; keep credentials in the mounted credentials directory. - # SERVER_MAINTENANCE_TASK_ENV: TASK_SETTING_A,TASK_SETTING_B volumes: - /etc/localtime:/etc/localtime:ro - ${PYTHON_TOOLS_VOLUME_ROOT:?Set the persistent volume root}/state:/var/lib/server-maintenance diff --git a/python-tools/tests/test_runner.py b/python-tools/tests/test_runner.py index 023ddf9..6cf65c7 100644 --- a/python-tools/tests/test_runner.py +++ b/python-tools/tests/test_runner.py @@ -77,6 +77,19 @@ def test_tasks_in_one_group_are_serialized_and_groups_overlap(manager: RunManage ] +def test_cleanup_removes_state_for_a_removed_group_after_a_grace_period(manager: RunManager) -> None: + write_task(manager.settings.task_root, "active", "task.py", 'print("active")\n') + removed = manager.settings.state_root / "tasks" / "removed" + removed.mkdir(parents=True) + (removed / "state.json").write_text("{}\n", encoding="utf-8") + + manager.cleanup_transient_files(orphaned_task_state_retention_seconds=60) + assert removed.is_dir() + + manager.cleanup_transient_files(orphaned_task_state_retention_seconds=0) + assert not removed.exists() + + def test_group_stops_after_failed_task(manager: RunManager) -> None: root = manager.settings.task_root write_task(root, "workflow", "01_ok.py", '"""Works."""\nprint("ok")\n') @@ -170,11 +183,10 @@ data = json.loads(Path(os.environ["SERVER_MAINTENANCE_INPUT"]).read_text()) assert stored.input_json == '{"attempts":2,"identifier":"AB123"}' -def test_task_receives_only_explicitly_forwarded_environment( +def test_task_does_not_receive_parent_environment( manager: RunManager, monkeypatch, ) -> None: - monkeypatch.setenv("SERVER_MAINTENANCE_TASK_ENV", "TASK_SETTING") monkeypatch.setenv("TASK_SETTING", "available") monkeypatch.setenv("UNRELATED_SETTING", "hidden") root = manager.settings.task_root @@ -182,7 +194,7 @@ def test_task_receives_only_explicitly_forwarded_environment( root, "environment", "read_environment.py", - '''"""Reads forwarded configuration.""" + '''"""Reads the task environment.""" import os from pathlib import Path @@ -198,7 +210,32 @@ state = Path(os.environ["SERVER_MAINTENANCE_STATE"]) assert wait_for(manager, run.id, {"succeeded"}) == "succeeded" assert (manager.settings.state_root / "tasks" / "environment" / "environment.txt").read_text() == ( - "available:" + ":" + ) + + +def test_task_receives_application_state_root(manager: RunManager) -> None: + root = manager.settings.task_root + write_task( + root, + "environment", + "read_state_root.py", + '''"""Reads the application state root.""" +import os +from pathlib import Path + +Path(os.environ["SERVER_MAINTENANCE_STATE"]).joinpath("state-root.txt").write_text( + os.environ["SERVER_MAINTENANCE_STATE_ROOT"]) +''', + ) + task = TaskCatalog(root).task("environment/read_state_root.py") + assert task is not None + + run, _ = manager.run_task(task) + + assert wait_for(manager, run.id, {"succeeded"}) == "succeeded" + assert (manager.settings.state_root / "tasks" / "environment" / "state-root.txt").read_text() == str( + manager.settings.state_root )