Inquiry about how cyclic tasks are handled in Open edX

Hey everyone,

I’ve recently been informed about a bug with the platform I’m working on. Namely, planned emails from the Communications MFE weren’t being sent out.

So I scoured through edx-platform’s code and realized that the Scheduled Tasks are just saved in the DB as objects and the function - process_scheduled_instructor_tasks found in lms/djangoapps/instructor_task/api.py submits them accordingly.

This function however, from what I see can ONLY be called from a django management command?!

Am I missing something or is Open edX shipped with this but requires the platform operator to wire some kind of beat service to that? I later found this related page (and another post on here regarding why celery beat was not chosen):

I am just surprised that this feature is shipped with a platform as a silent failure without explicit warnings.

I already set up celery beat for some other custom parts of the platform and am planning to wire it together with this function but I am looking for a word on whether I missed some config tutorial or this is just not stated.

@dave - do you know anything here, or could tag someone else in?

Maybe @omar would know about this area?

I haven’t seen this new feature.

This feature has traces back to this pull request which has been done by the 2U/edX team in 2022:

Not sure who’s still there in 2U from the original contributors, but it’s worth a ping.

@jhynes if you’re still active here, please take a look.

Well it seems like this inquiry has kind of lost it’s momentum so I’m just gonna post my minimal solution below in case anyone else in the future comes across this issue.

It’s resolved by two plugins.

  1. Tutor plugin on the host adding a celery beat service:

Structure:

├── pyproject.toml
└── src
    ├── compose-dev.yml
    ├── compose-prod.yml
    ├── plugin.py

pyproject.toml

requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[project]
name = "instructor_tasks"
version = "1.0.0"
description = "Process scheduled instructor tasks"

[project.entry-points."tutor.plugin.v1"]
instructor_tasks = "src.plugin"

[tool.setuptools.package-data]
instructor_tasks = [
    "src/*.yml",
]

[tool.setuptools.packages.find]
where = ["."]
include = ["src*"]

plugin.py

from pathlib import Path
from tutor import hooks

COMPOSE_DEV = "compose-dev.yml"
COMPOSE_PROD = "compose-prod.yml"

DIR = Path(__file__).resolve().parent

compose_dev_str = (DIR / COMPOSE_DEV).read_text()
compose_prod_str = (DIR / COMPOSE_PROD).read_text()

hooks.Filters.ENV_PATCHES.add_item(
    ("local-docker-compose-dev-services", compose_dev_str)
)

hooks.Filters.ENV_PATCHES.add_item(
    ("local-docker-compose-prod-services", compose_prod_str)
)

compose-dev.yml

lms-beat:
    image: {{ DOCKER_IMAGE_OPENEDX_DEV }}
    environment:
      SERVICE_VARIANT: lms
      DJANGO_SETTINGS_MODULE: lms.envs.tutor.development
    command:
      - "celery"
      - "--app=lms.celery"
      - "beat"
      - "--loglevel=info"
    restart: unless-stopped
    volumes:
      - ../apps/openedx/settings/lms:/openedx/edx-platform/lms/envs/tutor:ro
      - ../apps/openedx/settings/cms:/openedx/edx-platform/cms/envs/tutor:ro
      - ../apps/openedx/config:/openedx/config:ro
      - ../../data/lms:/openedx/data
      - ../../data/openedx-media:/openedx/media
      - ../../data/openedx-media-private:/openedx/media-private
      {%- for mount in iter_mounts(MOUNTS, "openedx", "lms-beat") %}
      - {{ mount }}
      {%- endfor %}
    depends_on:
      - lms-worker
      - lms

compose-prod.yml

lms-beat:
    image: {{ DOCKER_IMAGE_OPENEDX }}
    environment:
      SERVICE_VARIANT: lms
      DJANGO_SETTINGS_MODULE: lms.envs.tutor.production
    command:
      - "celery"
      - "--app=lms.celery"
      - "beat"
      - "--loglevel=info"
    restart: unless-stopped
    volumes:
      - ../apps/openedx/settings/lms:/openedx/edx-platform/lms/envs/tutor:ro
      - ../apps/openedx/settings/cms:/openedx/edx-platform/cms/envs/tutor:ro
      - ../apps/openedx/config:/openedx/config:ro
      - ../../data/lms:/openedx/data
      - ../../data/openedx-media:/openedx/media
      - ../../data/openedx-media-private:/openedx/media-private
      {%- for mount in iter_mounts(MOUNTS, "openedx", "lms-beat") %}
      - {{ mount }}
      {%- endfor %}
    depends_on:
      - lms-worker
      - lms

  1. Django app plugin linking a custom task to the function that is invoked from the management command plus updating the beat config upon django launch during the settings phase.

Structure:

├── instructor_tasks
│   ├── apps.py
│   ├── __init__.py
│   ├── settings.py
│   └── tasks.py
└── pyproject.toml

pyproject.toml

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[project]
name = "instructor_tasks"
version = "1.0.0"
description = "Process scheduled instructor tasks"

[project.entry-points."lms.djangoapp"]
instructor_tasks = "instructor_tasks.apps:InstructorTasksConfig"

[tool.setuptools.packages.find]
where = ["."]
include = ["instructor_tasks*"]

apps.py

import logging
from django.apps import AppConfig

log = logging.getLogger(__name__)


class InstructorTasksConfig(AppConfig):
    name = "instructor_tasks"
    verbose_name = "Instructor Task"

    plugin_app = {
        'settings_config':{
            'lms.djangoapp':{
                'common':{'relative_path':'settings'}
            }
        }
    }

    def ready(self):
        log.info("Test, the app is working!")

settings.py

from celery.schedules import crontab

def plugin_settings(settings):
   settings.CELERYBEAT_SCHEDULE.update({
      'call_process_scheduled_instructor_tasks': {
         'task': 'instructor_tasks.tasks.call_process_scheduled_instructor_tasks',
         'schedule' : crontab(), # no args - every minute just what we want as granularity of mail scheduling is 1min
         'options' : {'queue' : 'edx.lms.core.default'}
      },
   })

tasks.py

import datetime
import logging

import pytz
from celery import shared_task
from django.db import transaction
from lms.djangoapps.instructor_task.api_helper import (
    QueueConnectionError,
    submit_scheduled_task,
)
from lms.djangoapps.instructor_task.models import (
    QUEUING,
    SCHEDULED,
    InstructorTaskSchedule,
)

log = logging.getLogger(__name__)

@shared_task
def call_process_scheduled_instructor_tasks():
    log.info("Calling process_scheduled_instructor_tasks")

    with transaction.atomic():
        now = datetime.datetime.now(pytz.utc)
        due_schedules = InstructorTaskSchedule.objects.filter(
            task__task_state=SCHEDULED,
            task_due__lte=now,
        )

        log.info(f"Found {due_schedules.count()} scheduled instructor tasks")

        for schedule in due_schedules:
            try:
                log.info(f"Queueing task {schedule.task.id}")
                submit_scheduled_task(schedule)

                if schedule.task.task_state == SCHEDULED:
                    schedule.task.task_state = QUEUING
                    schedule.task.save(update_fields=["task_state"])
                    log.info(f"Updated task {schedule.task.id} state to {QUEUING}")

            except QueueConnectionError as exc:
                log.error(f"Error processing task {schedule.task.id}: {exc}")

Both plugins as .tar.gz below

plugins.tar.gz (2.0 KB)


I am sure this could be solved in a better way. The current solution is not ideal because it requires a task to run every minute, which is wasteful. However, increasing the execution interval would require an MFE change to increase the time selection granularity in the UI accordingly.