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.
- 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
- 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.