diff --git a/cms/djangoapps/contentstore/rest_api/v1/views/course_optimizer.py b/cms/djangoapps/contentstore/rest_api/v1/views/course_optimizer.py index b97b83acf669..0df0d8ce6454 100644 --- a/cms/djangoapps/contentstore/rest_api/v1/views/course_optimizer.py +++ b/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": + "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', diff --git a/cms/djangoapps/contentstore/rest_api/v1/views/tests/test_course_optimizer.py b/cms/djangoapps/contentstore/rest_api/v1/views/tests/test_course_optimizer.py index f4b9b2b9ead7..2cd20b594f26 100644 --- a/cms/djangoapps/contentstore/rest_api/v1/views/tests/test_course_optimizer.py +++ b/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( diff --git a/cms/djangoapps/contentstore/tasks.py b/cms/djangoapps/contentstore/tasks.py index 71b86acca201..c37987399893 100644 --- a/cms/djangoapps/contentstore/tasks.py +++ b/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): diff --git a/cms/djangoapps/contentstore/tests/test_tasks.py b/cms/djangoapps/contentstore/tests/test_tasks.py index 5a76b7d67fe7..c6b61ebd140c 100644 --- a/cms/djangoapps/contentstore/tests/test_tasks.py +++ b/cms/djangoapps/contentstore/tests/test_tasks.py @@ -10,8 +10,10 @@ from celery import Task import pytest +import requests from django.conf import settings from django.contrib.auth.models import User # lint-amnesty, pylint: disable=imported-auth-user +from django.core.cache import cache from django.test.utils import override_settings from edx_toggles.toggles.testutils import override_waffle_flag from opaque_keys.edx.keys import CourseKey @@ -32,7 +34,9 @@ from xmodule.modulestore.tests.factories import CourseFactory, BlockFactory # lint-amnesty, pylint: disable=wrong-import-order from ..tasks import ( LinkState, + course_analysis_report_cache_key, export_olx, + submit_course_analysis_report, update_special_exams_and_publish, rerun_course, _validate_urls_access_in_batches, @@ -668,3 +672,103 @@ def test_extract_content_URLs_from_course(self): "https://another-valid.com" ] self.assertEqual(extract_content_URLs_from_course(content), set(expected)) + + +class SubmitCourseAnalysisReportTaskTest(CourseTestCase): + """ + Tests for submit_course_analysis_report, the background task that + exports a course and uploads it to the Course Optimizer extended-report + backend (xpert-ai-workflows). + """ + + def setUp(self): + super().setUp() + cache.clear() + self.course_key_string = str(self.course.id) + self.cache_key = course_analysis_report_cache_key(self.course_key_string) + + def _mock_tarball(self): + tarball = mock.Mock() + tarball.name = '/tmp/whatever.tar.gz' + return tarball + + def test_has_explicit_celery_time_limits(self): + self.assertEqual( + submit_course_analysis_report.soft_time_limit, + settings.COURSE_ANALYSIS_EXPORT_TASK_SOFT_TIME_LIMIT, + ) + self.assertEqual( + submit_course_analysis_report.time_limit, + settings.COURSE_ANALYSIS_EXPORT_TASK_TIME_LIMIT, + ) + + @mock.patch('cms.djangoapps.contentstore.tasks.requests.post') + @mock.patch('cms.djangoapps.contentstore.tasks.create_export_tarball') + def test_uploads_export_to_backend(self, mock_export, mock_post): + mock_export.return_value = self._mock_tarball() + mock_post.return_value = mock.Mock(status_code=202) + cache.set(self.cache_key, {'status': 'PENDING'}) + + submit_course_analysis_report(self.course_key_string) + + mock_post.assert_called_once() + self.assertIn(self.course_key_string, mock_post.call_args.args[0]) + self.assertEqual( + mock_post.call_args.kwargs['headers']['X-Api-Key'], + settings.COURSE_ANALYSIS_WORKFLOW_API_KEY, + ) + mock_post.return_value.raise_for_status.assert_called_once() + self.assertIsNone(cache.get(self.cache_key)) + + @mock.patch('cms.djangoapps.contentstore.tasks.requests.post') + @mock.patch('cms.djangoapps.contentstore.tasks.create_export_tarball') + def test_closes_tarball_even_on_request_failure(self, mock_export, mock_post): + tarball = self._mock_tarball() + mock_export.return_value = tarball + mock_post.side_effect = requests.ConnectionError() + + with self.assertRaises(requests.ConnectionError): + submit_course_analysis_report(self.course_key_string) + + tarball.close.assert_called_once() + + @mock.patch('cms.djangoapps.contentstore.tasks.requests.post') + @mock.patch('cms.djangoapps.contentstore.tasks.create_export_tarball') + def test_request_failure_marks_run_failed_in_cache(self, mock_export, mock_post): + mock_export.return_value = self._mock_tarball() + mock_post.side_effect = requests.ConnectionError('unreachable') + cache.set(self.cache_key, {'status': 'PENDING'}) + + with self.assertRaises(requests.ConnectionError): + submit_course_analysis_report(self.course_key_string) + + self.assertEqual( + cache.get(self.cache_key), + {'status': 'FAILED', 'error': 'unreachable'}, + ) + + @mock.patch('cms.djangoapps.contentstore.tasks.requests.post') + @mock.patch('cms.djangoapps.contentstore.tasks.create_export_tarball') + def test_raises_on_backend_error_response(self, mock_export, mock_post): + mock_export.return_value = self._mock_tarball() + mock_post.return_value = mock.Mock(status_code=500) + mock_post.return_value.raise_for_status.side_effect = requests.HTTPError() + + with self.assertRaises(requests.HTTPError): + submit_course_analysis_report(self.course_key_string) + + @mock.patch('cms.djangoapps.contentstore.tasks.requests.post') + @mock.patch('cms.djangoapps.contentstore.tasks.create_export_tarball') + def test_backend_error_response_marks_run_failed_in_cache(self, mock_export, mock_post): + mock_export.return_value = self._mock_tarball() + mock_post.return_value = mock.Mock(status_code=500) + mock_post.return_value.raise_for_status.side_effect = requests.HTTPError('server error') + cache.set(self.cache_key, {'status': 'PENDING'}) + + with self.assertRaises(requests.HTTPError): + submit_course_analysis_report(self.course_key_string) + + self.assertEqual( + cache.get(self.cache_key), + {'status': 'FAILED', 'error': 'server error'}, + ) diff --git a/cms/envs/common.py b/cms/envs/common.py index 7a8bde7dce96..7de7b719da4c 100644 --- a/cms/envs/common.py +++ b/cms/envs/common.py @@ -1800,6 +1800,32 @@ def _should_send_xblock_events(settings): # .. COURSE_ANALYSIS_WORKFLOW_URL before giving up. A slow/unreachable # .. backend shouldn't tie up a Studio request thread waiting on it. COURSE_ANALYSIS_WORKFLOW_REQUEST_TIMEOUT_SECONDS = 5 +# .. setting_name: COURSE_ANALYSIS_EXPORT_TASK_SOFT_TIME_LIMIT +# .. setting_default: 600 +# .. setting_description: Soft Celery time limit, in seconds, for the +# .. background task that exports and uploads a course to the Course +# .. Optimizer extended-report backend. When exceeded, the task gets a +# .. chance to record the run as failed before it's terminated, so a +# .. pathological or very large course can't tie up a worker indefinitely. +COURSE_ANALYSIS_EXPORT_TASK_SOFT_TIME_LIMIT = 600 +# .. setting_name: COURSE_ANALYSIS_EXPORT_TASK_TIME_LIMIT +# .. setting_default: 660 +# .. setting_description: Hard Celery time limit, in seconds, for the +# .. background task that exports and uploads a course to the Course +# .. Optimizer extended-report backend. Should be a little higher than +# .. COURSE_ANALYSIS_EXPORT_TASK_SOFT_TIME_LIMIT to give the soft limit a +# .. chance to run first. +COURSE_ANALYSIS_EXPORT_TASK_TIME_LIMIT = 660 +# .. setting_name: COURSE_ANALYSIS_REPORT_CACHE_TIMEOUT_SECONDS +# .. setting_default: 900 +# .. setting_description: How long Studio remembers that a Course Optimizer +# .. extended-analysis run is pending or failed, keyed by course. This +# .. covers the window between queuing the export/upload task and +# .. xpert-ai-workflows actually creating the run, during which the status +# .. endpoint would otherwise have nothing new to report and could +# .. mistake a stale prior run for the one just requested. Should be a +# .. little higher than COURSE_ANALYSIS_EXPORT_TASK_TIME_LIMIT. +COURSE_ANALYSIS_REPORT_CACHE_TIMEOUT_SECONDS = 900 # .. setting_name: LIBRARY_ENABLED_BLOCKS # .. setting_default: ['problem', 'video', 'html', 'drag-and-drop-v2']