Merge pull request #3529 from OpenNeuroOrg/maintenance-job-scheduler

Add worker task schedules and maintenance tasks

Author: nellh

docker-compose.yml

@@ -157,6 +157,34 @@ services:
           - datalad-0
           - datalad-1
 
+  scheduler:
+    platform: linux/amd64
+    build:
+      context: services/datalad
+      platform: linux/amd64
+    volumes:
+      - ${PERSISTENT_DIR}/datalad:/datalad:z
+      - ./services/datalad/datalad_service:/srv/datalad_service
+      - ./datalad-key:/datalad-key
+    env_file: ./config.env
+    init: true
+    command: [
+      'taskiq',
+      'scheduler',
+      'datalad_service.broker.scheduler:scheduler',
+      '--tasks-pattern',
+      'datalad_service/tasks/*.py',
+      '--fs-discover',
+    ]
+    depends_on:
+      redis:
+        condition: service_started
+    networks:
+      default:
+        aliases:
+          - datalad-0
+          - datalad-1
+
   # nginx + app
   web:
     image: docker.io/library/nginx:1.16.1

helm/openneuro/Chart.yaml

@@ -1,6 +1,6 @@
 apiVersion: v1
 name: openneuro
-version: 1.5.0
+version: 1.6.0
 description: OpenNeuro production deployment chart
 home: https://openneuro.org
 sources:

helm/openneuro/templates/dataset-worker-scheduler.yaml

@@ -0,0 +1,35 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ .Release.Name }}-scheduler
+  labels:
+    app: {{ .Release.Name }}-scheduler
+    chart: '{{ .Chart.Name }}-{{ .Chart.Version }}'
+    release: '{{ .Release.Name }}'
+    heritage: '{{ .Release.Service }}'
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: {{ .Release.Name }}-scheduler
+  template:
+    metadata:
+      labels:
+        app: {{ .Release.Name }}-scheduler
+      annotations:
+        checksum/config: {{ include (print $.Template.BasePath "/configmap.yaml") . | sha256sum }}
+        checksum/secret: {{ include (print $.Template.BasePath "/secret.yaml") . | sha256sum }}
+    spec:
+      containers:
+        - name: {{ .Release.Name }}-scheduler
+          image: 'openneuro/datalad-service:v{{ .Chart.AppVersion }}'
+          command: 
+          resources:
+            requests:
+              cpu: {{ .Values.workerCpuRequests }}
+              memory: {{ .Values.workerMemoryRequests }}
+          envFrom:
+            - configMapRef:
+                name: {{ .Release.Name }}-configmap
+            - secretRef:
+                name: {{ .Release.Name }}-secret

services/datalad/datalad_service/app.py

@@ -53,6 +53,9 @@ def before_send(event):
     release=f'openneuro-datalad-service@{datalad_service.version.get_version()}',
     server_name=socket.gethostname(),
     before_send=before_send,
+    _experiments={
+        'enable_logs': True,
+    },
 )
 
 

services/datalad/datalad_service/broker/__init__.py

@@ -3,6 +3,7 @@
 from taskiq import InMemoryBroker
 from taskiq_redis import RedisAsyncResultBackend, RedisStreamBroker
 
+
 from datalad_service import config
 from datalad_service.broker.get_docker_scale import get_docker_scale
 

services/datalad/datalad_service/broker/scheduler.py

@@ -0,0 +1,10 @@
+from taskiq import TaskiqScheduler
+from taskiq.schedule_sources import LabelScheduleSource
+from taskiq_redis import RedisScheduleSource
+
+from datalad_service import config
+from datalad_service.broker import broker
+
+redis_source = RedisScheduleSource(f'redis://{config.REDIS_HOST}:{config.REDIS_PORT}/0')
+label_source = LabelScheduleSource(broker)
+scheduler = TaskiqScheduler(broker, sources=[redis_source, label_source])

services/datalad/datalad_service/tasks/maintenance.py

@@ -0,0 +1,81 @@
+import os
+import logging
+import subprocess
+import random
+
+
+from datalad_service.config import DATALAD_DATASET_PATH
+from datalad_service.broker import broker
+
+
+def dataset_factory():
+    """
+    Factory that yields dataset paths from the DATALAD_DATASET_PATH directory.
+    It shuffles the list of datasets and yields them one by one.
+    When all datasets have been yielded, it re-reads the directory and shuffles again.
+    """
+    datasets = []
+    while True:
+        if not datasets:
+            # Read the directory and shuffle if the list is empty
+            try:
+                datasets = [
+                    os.path.join(DATALAD_DATASET_PATH, d)
+                    for d in os.listdir(DATALAD_DATASET_PATH)
+                    if os.path.isdir(os.path.join(DATALAD_DATASET_PATH, d))
+                    and d.startswith('ds')
+                ]
+                if not datasets:
+                    logging.warning(
+                        f'No datasets found in {DATALAD_DATASET_PATH} for maintenance tasks.'
+                    )
+                    return
+                random.shuffle(datasets)
+            except FileNotFoundError:
+                logging.error(f'DATALAD_DATASET_PATH not found: {DATALAD_DATASET_PATH}')
+                return
+        yield datasets.pop()
+
+
+gc_dataset_generator = dataset_factory()
+fsck_dataset_generator = dataset_factory()
+
+
+@broker.task(schedule=[{'cron': '*/15 * * * *'}])
+def gc_dataset(dataset_path=None):
+    """Run git gc on a random dataset periodically."""
+    try:
+        if not dataset_path:
+            dataset_path = next(gc_dataset_generator)
+    except StopIteration:
+        logging.info('No datasets available for git gc.')
+        return
+
+    logging.info(f'Running git gc on dataset: {dataset_path}')
+
+    gc = subprocess.run(
+        ['git', 'gc', '--cruft', '--prune=1.day.ago'],
+        cwd=dataset_path,
+        capture_output=True,
+        text=True,
+    )
+    if gc.returncode != 0:
+        logging.error(f'`git gc` failed for `{dataset_path}`: {gc.stderr}')
+
+
+@broker.task(schedule=[{'cron': '7 * * * *'}])
+def git_fsck_dataset(dataset_path=None):
+    """Routinely verify git repository integrity and produce an error if any issues are reported."""
+    try:
+        if not dataset_path:
+            dataset_path = next(fsck_dataset_generator)
+    except StopIteration:
+        logging.info('No datasets available for git fsck.')
+        return
+    git_fsck = subprocess.run(
+        ['git', 'fsck', '--full'], cwd=dataset_path, capture_output=True, text=True
+    )
+    if git_fsck.returncode != 0:
+        logging.error(
+            f'`git fsck --full` failed for `{dataset_path}`: {git_fsck.stderr}'
+        )

services/datalad/tests/test_maintenance.py

@@ -0,0 +1,9 @@
+from datalad_service.tasks.maintenance import gc_dataset, git_fsck_dataset
+
+
+def test_gc_dataset(new_dataset):
+    gc_dataset(new_dataset.path)
+
+
+def test_git_fsck_dataset(new_dataset):
+    git_fsck_dataset(new_dataset.path)