fix: run Course Optimizer extended-analysis export/upload as a background task

cms/djangoapps/contentstore/rest_api/v1/views/course_optimizer.py

@@ -1,17 +1,17 @@
 """API Views for the Course Optimizer extended-analysis report."""
 
-import os
-
 import edx_api_doc_tools as apidocs
 import requests
 from django.conf import settings
+from django.core.cache import cache
+from kombu.exceptions import OperationalError
 from opaque_keys.edx.keys import CourseKey
 from rest_framework import status
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
-from cms.djangoapps.contentstore.tasks import create_export_tarball
+from cms.djangoapps.contentstore.tasks import course_analysis_report_cache_key, submit_course_analysis_report
 from cms.djangoapps.contentstore.toggles import enable_course_optimizer_extended_checks
 from common.djangoapps.student.auth import has_course_author_access
 from common.djangoapps.util.json_request import JsonResponse
@@ -20,7 +20,6 @@
     verify_course_exists,
     view_auth_classes,
 )
-from xmodule.modulestore.django import modulestore
 
 
 @view_auth_classes(is_authenticated=True)
@@ -38,16 +37,22 @@ class CourseAnalysisReportView(DeveloperErrorViewMixin, APIView):
             401: "The requester is not authenticated.",
             403: "The requester cannot access the specified course.",
             404: "The requested course does not exist.",
-            502: "The Course Optimizer extended-report backend is unreachable.",
         },
     )
     @verify_course_exists()
     def post(self, request: Request, course_id: str):
         """
-        Generate a fresh export of the course and hand it to the Course
-        Optimizer extended-report backend (xpert-ai-workflows) to start a
-        new analysis run. Studio generates the export server-side -- the
-        browser never uploads anything or talks to that backend directly.
+        Queue a background task to generate a fresh export of the course
+        and hand it to the Course Optimizer extended-report backend
+        (xpert-ai-workflows) to start a new analysis run. Studio generates
+        the export server-side -- the browser never uploads anything or
+        talks to that backend directly.
+
+        Exporting and compressing a course can take a while for large
+        courses, so this runs as a Celery task rather than blocking a
+        Studio request thread on it -- this view returns as soon as the
+        task is queued. Callers should poll
+        CourseAnalysisReportStatusView for the run's progress.
 
         **Example Request**
 
@@ -56,7 +61,7 @@ def post(self, request: Request, course_id: str):
         **Response Values**
         ```json
         {
-            "run_id": <string>
+            "status": "PENDING"
         }
         ```
         """
@@ -70,28 +75,18 @@ def post(self, request: Request, course_id: str):
                 status=status.HTTP_400_BAD_REQUEST,
             )
 
-        course_block = modulestore().get_course(course_key)
-        tarball = create_export_tarball(course_block, course_key, {})
+        cache_key = course_analysis_report_cache_key(course_id)
+        cache.set(
+            cache_key,
+            {'status': 'PENDING'},
+            settings.COURSE_ANALYSIS_REPORT_CACHE_TIMEOUT_SECONDS,
+        )
         try:
-            tarball.seek(0)
-            try:
-                response = requests.post(
-                    f'{settings.COURSE_ANALYSIS_WORKFLOW_URL}/courses/{course_id}/runs',
-                    files={'file': (os.path.basename(tarball.name), tarball, 'application/gzip')},
-                    headers={'X-Api-Key': settings.COURSE_ANALYSIS_WORKFLOW_API_KEY},
-                    timeout=settings.COURSE_ANALYSIS_WORKFLOW_REQUEST_TIMEOUT_SECONDS,
-                )
-            except requests.RequestException:
-                return Response(status=status.HTTP_502_BAD_GATEWAY)
-        finally:
-            tarball.close()
-
-        try:
-            response_data = response.json()
-        except ValueError:
-            return Response(status=status.HTTP_502_BAD_GATEWAY)
-
-        return Response(response_data, status=response.status_code)
+            submit_course_analysis_report.delay(course_id)
+        except OperationalError:
+            cache.delete(cache_key)
+            raise
+        return Response({'status': 'PENDING'}, status=status.HTTP_202_ACCEPTED)
 
 
 @view_auth_classes()
@@ -131,6 +126,13 @@ def get(self, request: Request, course_id: str):
             The xpert-ai-workflows run-status response, passed through
             unchanged: `{run_id, status, report, error}`. A 404 means the
             course has no analysis runs yet.
+
+            While a background export/upload triggered by
+            CourseAnalysisReportView is in flight (or just failed), this
+            returns `{status: "PENDING"}` or `{status: "FAILED", error}`
+            instead of proxying xpert-ai-workflows -- otherwise, a course
+            with an older completed run would look done again as soon as
+            it's requeued, even though the requested run hasn't landed yet.
         """
         course_key = CourseKey.from_string(course_id)
         if not has_course_author_access(request.user, course_key):
@@ -142,6 +144,10 @@ def get(self, request: Request, course_id: str):
                 status=status.HTTP_400_BAD_REQUEST,
             )
 
+        cached_status = cache.get(course_analysis_report_cache_key(course_id))
+        if cached_status is not None:
+            return Response(cached_status, status=status.HTTP_200_OK)
+
         try:
             response = requests.get(
                 f'{settings.COURSE_ANALYSIS_WORKFLOW_URL}/courses/{course_id}/runs/latest',

cms/djangoapps/contentstore/rest_api/v1/views/tests/test_course_optimizer.py

@@ -4,39 +4,34 @@
 from unittest.mock import Mock, patch
 
 import requests
-from django.conf import settings
+from django.core.cache import cache
 from django.urls import reverse
 from edx_toggles.toggles.testutils import override_waffle_flag
+from kombu.exceptions import OperationalError
 from rest_framework import status
 
+from cms.djangoapps.contentstore.tasks import course_analysis_report_cache_key
 from cms.djangoapps.contentstore.tests.utils import CourseTestCase
 from cms.djangoapps.contentstore.toggles import ENABLE_COURSE_OPTIMIZER_EXTENDED_CHECKS
 
 
 class CourseAnalysisReportViewTest(CourseTestCase):
     """
-    Tests for CourseAnalysisReportView, which kicks off a Course Optimizer
-    extended-analysis run by generating a course export server-side and
-    handing it to the xpert-ai-workflows backend.
+    Tests for CourseAnalysisReportView, which queues a background task to
+    generate a course export and hand it to the xpert-ai-workflows backend
+    to kick off a Course Optimizer extended-analysis run.
     """
 
     def setUp(self):
         super().setUp()
+        cache.clear()
         self.url = reverse(
             'cms.djangoapps.contentstore:v1:course_analysis_report',
             kwargs={'course_id': str(self.course.id)},
         )
-        self.export_patch = (
-            'cms.djangoapps.contentstore.rest_api.v1.views.course_optimizer.create_export_tarball'
+        self.task_patch = (
+            'cms.djangoapps.contentstore.rest_api.v1.views.course_optimizer.submit_course_analysis_report'
         )
-        self.backend_post_patch = (
-            'cms.djangoapps.contentstore.rest_api.v1.views.course_optimizer.requests.post'
-        )
-
-    def _mock_tarball(self):
-        tarball = Mock()
-        tarball.name = '/tmp/whatever.tar.gz'
-        return tarball
 
     def test_unauthenticated(self):
         self.client.logout()
@@ -53,42 +48,34 @@ def test_waffle_flag_disabled_returns_400(self):
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
 
     @override_waffle_flag(ENABLE_COURSE_OPTIMIZER_EXTENDED_CHECKS, True)
-    def test_kicks_off_backend_run(self):
-        with patch(self.export_patch) as mock_export, patch(self.backend_post_patch) as mock_post:
-            mock_export.return_value = self._mock_tarball()
-            mock_post.return_value = Mock(
-                status_code=202,
-                json=Mock(return_value={'run_id': 'run-123'}),
-            )
+    def test_queues_background_task_and_returns_immediately(self):
+        with patch(self.task_patch) as mock_task:
             response = self.client.post(self.url)
 
         self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
-        self.assertEqual(response.json(), {'run_id': 'run-123'})
-        self.assertEqual(
-            mock_post.call_args.kwargs['headers']['X-Api-Key'],
-            settings.COURSE_ANALYSIS_WORKFLOW_API_KEY,
-        )
+        self.assertEqual(response.json(), {'status': 'PENDING'})
+        mock_task.delay.assert_called_once_with(str(self.course.id))
 
     @override_waffle_flag(ENABLE_COURSE_OPTIMIZER_EXTENDED_CHECKS, True)
-    def test_backend_unreachable_returns_502(self):
-        with patch(self.export_patch) as mock_export, patch(self.backend_post_patch) as mock_post:
-            mock_export.return_value = self._mock_tarball()
-            mock_post.side_effect = requests.ConnectionError()
-            response = self.client.post(self.url)
+    def test_marks_run_pending_before_queuing_task(self):
+        with patch(self.task_patch):
+            self.client.post(self.url)
 
-        self.assertEqual(response.status_code, status.HTTP_502_BAD_GATEWAY)
+        self.assertEqual(
+            cache.get(course_analysis_report_cache_key(str(self.course.id))),
+            {'status': 'PENDING'},
+        )
 
     @override_waffle_flag(ENABLE_COURSE_OPTIMIZER_EXTENDED_CHECKS, True)
-    def test_backend_returns_invalid_json_returns_502(self):
-        with patch(self.export_patch) as mock_export, patch(self.backend_post_patch) as mock_post:
-            mock_export.return_value = self._mock_tarball()
-            mock_post.return_value = Mock(
-                status_code=202,
-                json=Mock(side_effect=ValueError()),
-            )
-            response = self.client.post(self.url)
-
-        self.assertEqual(response.status_code, status.HTTP_502_BAD_GATEWAY)
+    def test_clears_pending_marker_when_enqueue_fails(self):
+        with patch(self.task_patch) as mock_task:
+            mock_task.delay.side_effect = OperationalError('broker unavailable')
+            with self.assertRaises(OperationalError):
+                self.client.post(self.url)
+
+        self.assertIsNone(
+            cache.get(course_analysis_report_cache_key(str(self.course.id))),
+        )
 
 
 class CourseAnalysisReportStatusViewTest(CourseTestCase):
@@ -100,6 +87,7 @@ class CourseAnalysisReportStatusViewTest(CourseTestCase):
 
     def setUp(self):
         super().setUp()
+        cache.clear()
         self.url = reverse(
             'cms.djangoapps.contentstore:v1:course_analysis_report_status',
             kwargs={'course_id': str(self.course.id)},
@@ -168,6 +156,32 @@ def test_backend_returns_invalid_json_returns_502(self):
 
         self.assertEqual(response.status_code, status.HTTP_502_BAD_GATEWAY)
 
+    @override_waffle_flag(ENABLE_COURSE_OPTIMIZER_EXTENDED_CHECKS, True)
+    def test_pending_task_short_circuits_backend_call(self):
+        cache.set(
+            course_analysis_report_cache_key(str(self.course.id)),
+            {'status': 'PENDING'},
+        )
+        with patch(self.backend_get_patch) as mock_get:
+            response = self.client.get(self.url)
+
+        mock_get.assert_not_called()
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.json(), {'status': 'PENDING'})
+
+    @override_waffle_flag(ENABLE_COURSE_OPTIMIZER_EXTENDED_CHECKS, True)
+    def test_failed_task_short_circuits_backend_call(self):
+        cache.set(
+            course_analysis_report_cache_key(str(self.course.id)),
+            {'status': 'FAILED', 'error': 'boom'},
+        )
+        with patch(self.backend_get_patch) as mock_get:
+            response = self.client.get(self.url)
+
+        mock_get.assert_not_called()
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.json(), {'status': 'FAILED', 'error': 'boom'})
+
     @override_waffle_flag(ENABLE_COURSE_OPTIMIZER_EXTENDED_CHECKS, True)
     def test_produces_404_when_course_does_not_exist(self):
         url = reverse(

cms/djangoapps/contentstore/tasks.py

@@ -16,11 +16,13 @@
 
 import aiohttp
 import olxcleaner
+import requests
 from ccx_keys.locator import CCXLocator
 from celery import shared_task
 from celery.utils.log import get_task_logger
 from django.conf import settings
 from django.contrib.auth import get_user_model
+from django.core.cache import cache
 from django.core.exceptions import SuspiciousOperation
 from django.core.files import File
 from django.test import RequestFactory
@@ -1632,6 +1634,70 @@ def _write_broken_links_to_file(broken_or_locked_urls, broken_links_file):
         json.dump(broken_or_locked_urls, file, indent=4)
 
 
+def course_analysis_report_cache_key(course_key_string: str) -> str:
+    """
+    Cache key tracking the in-flight/failed state of a course's Course
+    Optimizer extended-analysis run, shared between
+    submit_course_analysis_report and CourseAnalysisReportStatusView.
+    """
+    return f'course_analysis_report_status:{course_key_string}'
+
+
+@shared_task(
+    soft_time_limit=settings.COURSE_ANALYSIS_EXPORT_TASK_SOFT_TIME_LIMIT,
+    time_limit=settings.COURSE_ANALYSIS_EXPORT_TASK_TIME_LIMIT,
+)
+@set_code_owner_attribute
+def submit_course_analysis_report(course_key_string: str) -> None:
+    """
+    Generates a fresh course export and hands it to the Course Optimizer
+    extended-report backend (xpert-ai-workflows) to start a new analysis
+    run.
+
+    Runs as a background task because exporting and compressing a course
+    can take a while for large courses -- the API view that queues this
+    returns 202 immediately rather than blocking a Studio request thread
+    on it. Callers poll xpert-ai-workflows (via CourseAnalysisReportStatusView)
+    for the run's progress.
+
+    CourseAnalysisReportView marks the run pending (via
+    course_analysis_report_cache_key) before queuing this task. If export
+    or upload fails -- including this task hitting its own time limit --
+    that cache entry is updated to a terminal 'FAILED' status instead of
+    leaving the status endpoint to poll a stale or nonexistent run
+    forever. On success the entry is cleared so the status endpoint goes
+    back to proxying xpert-ai-workflows directly.
+
+    Note: a hard Celery time limit kills the worker process outright, so
+    the except block below can't run in that case -- the pending cache
+    entry just expires on its own per COURSE_ANALYSIS_REPORT_CACHE_TIMEOUT_SECONDS.
+    """
+    cache_key = course_analysis_report_cache_key(course_key_string)
+    try:
+        course_key = CourseKey.from_string(course_key_string)
+        course_block = modulestore().get_course(course_key)
+        tarball = create_export_tarball(course_block, course_key, {})
+        try:
+            tarball.seek(0)
+            response = requests.post(
+                f'{settings.COURSE_ANALYSIS_WORKFLOW_URL}/courses/{course_key_string}/runs',
+                files={'file': (os.path.basename(tarball.name), tarball, 'application/gzip')},
+                headers={'X-Api-Key': settings.COURSE_ANALYSIS_WORKFLOW_API_KEY},
+                timeout=settings.COURSE_ANALYSIS_WORKFLOW_REQUEST_TIMEOUT_SECONDS,
+            )
+            response.raise_for_status()
+        finally:
+            tarball.close()
+    except Exception as exc:
+        cache.set(
+            cache_key,
+            {'status': 'FAILED', 'error': str(exc)},
+            settings.COURSE_ANALYSIS_REPORT_CACHE_TIMEOUT_SECONDS,
+        )
+        raise
+    cache.delete(cache_key)
+
+
 @shared_task
 @set_code_owner_attribute
 def handle_create_xblock_upstream_link(usage_key):