Merge pull request #1495 from edx/brian/remove-celery-group

Remove the use of celery.group from bulk email subtasks.

CHANGELOG.rst

@@ -5,6 +5,9 @@ These are notable changes in edx-platform.  This is a rolling list of changes,
 in roughly chronological order, most recent first.  Add your entries at or near
 the top.  Include a label indicating the component affected.
 
+LMS: Change bulk email implementation to use less memory, and to better handle
+duplicate tasks in celery.
+
 LMS: Improve forum error handling so that errors in the logs are
 clearer and HTTP status codes from the comments service indicating
 client error are correctly passed through to the client.

lms/djangoapps/bulk_email/tests/test_email.py

@@ -15,7 +15,7 @@ from student.tests.factories import UserFactory, GroupFactory, CourseEnrollmentF
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
 from bulk_email.models import Optout
-from instructor_task.subtasks import increment_subtask_status
+from instructor_task.subtasks import update_subtask_status
 
 STAFF_COUNT = 3
 STUDENT_COUNT = 10
@@ -29,13 +29,13 @@ class MockCourseEmailResult(object):
     """
     emails_sent = 0
 
-    def get_mock_increment_subtask_status(self):
+    def get_mock_update_subtask_status(self):
         """Wrapper for mock email function."""
-        def mock_increment_subtask_status(original_status, **kwargs):  # pylint: disable=W0613
+        def mock_update_subtask_status(entry_id, current_task_id, new_subtask_status):  # pylint: disable=W0613
             """Increments count of number of emails sent."""
-            self.emails_sent += kwargs.get('succeeded', 0)
-            return increment_subtask_status(original_status, **kwargs)
-        return mock_increment_subtask_status
+            self.emails_sent += new_subtask_status.succeeded
+            return update_subtask_status(entry_id, current_task_id, new_subtask_status)
+        return mock_update_subtask_status
 
 
 @override_settings(MODULESTORE=TEST_DATA_MONGO_MODULESTORE)
@@ -244,13 +244,13 @@ class TestEmailSendFromDashboard(ModuleStoreTestCase):
         )
 
     @override_settings(BULK_EMAIL_EMAILS_PER_TASK=3, BULK_EMAIL_EMAILS_PER_QUERY=7)
-    @patch('bulk_email.tasks.increment_subtask_status')
+    @patch('bulk_email.tasks.update_subtask_status')
     def test_chunked_queries_send_numerous_emails(self, email_mock):
         """
         Test sending a large number of emails, to test the chunked querying
         """
         mock_factory = MockCourseEmailResult()
-        email_mock.side_effect = mock_factory.get_mock_increment_subtask_status()
+        email_mock.side_effect = mock_factory.get_mock_update_subtask_status()
         added_users = []
         for _ in xrange(LARGE_NUM_EMAILS):
             user = UserFactory()

lms/djangoapps/bulk_email/tests/test_err_handling.py

@@ -22,8 +22,8 @@ from bulk_email.models import CourseEmail, SEND_TO_ALL
 from bulk_email.tasks import perform_delegate_email_batches, send_course_email
 from instructor_task.models import InstructorTask
 from instructor_task.subtasks import (
-    create_subtask_status,
     initialize_subtask_info,
+    SubtaskStatus,
     check_subtask_is_valid,
     update_subtask_status,
     DuplicateTaskException,
@@ -75,7 +75,7 @@ class TestEmailErrors(ModuleStoreTestCase):
         self.assertIsInstance(exc, SMTPDataError)
 
     @patch('bulk_email.tasks.get_connection', autospec=True)
-    @patch('bulk_email.tasks.increment_subtask_status')
+    @patch('bulk_email.tasks.update_subtask_status')
     @patch('bulk_email.tasks.send_course_email.retry')
     def test_data_err_fail(self, retry, result, get_conn):
         """
@@ -99,11 +99,11 @@ class TestEmailErrors(ModuleStoreTestCase):
         # We shouldn't retry when hitting a 5xx error
         self.assertFalse(retry.called)
         # Test that after the rejected email, the rest still successfully send
-        ((_initial_results), kwargs) = result.call_args
-        self.assertEquals(kwargs['skipped'], 0)
+        ((_entry_id, _current_task_id, subtask_status), _kwargs) = result.call_args
+        self.assertEquals(subtask_status.skipped, 0)
         expected_fails = int((settings.BULK_EMAIL_EMAILS_PER_TASK + 3) / 4.0)
-        self.assertEquals(kwargs['failed'], expected_fails)
-        self.assertEquals(kwargs['succeeded'], settings.BULK_EMAIL_EMAILS_PER_TASK - expected_fails)
+        self.assertEquals(subtask_status.failed, expected_fails)
+        self.assertEquals(subtask_status.succeeded, settings.BULK_EMAIL_EMAILS_PER_TASK - expected_fails)
 
     @patch('bulk_email.tasks.get_connection', autospec=True)
     @patch('bulk_email.tasks.send_course_email.retry')
@@ -146,7 +146,7 @@ class TestEmailErrors(ModuleStoreTestCase):
         exc = kwargs['exc']
         self.assertIsInstance(exc, SMTPConnectError)
 
-    @patch('bulk_email.tasks.increment_subtask_status')
+    @patch('bulk_email.tasks.SubtaskStatus.increment')
     @patch('bulk_email.tasks.log')
     def test_nonexistent_email(self, mock_log, result):
         """
@@ -216,10 +216,10 @@ class TestEmailErrors(ModuleStoreTestCase):
         to_list = ['test@test.com']
         global_email_context = {'course_title': 'dummy course'}
         subtask_id = "subtask-id-value"
-        subtask_status = create_subtask_status(subtask_id)
+        subtask_status = SubtaskStatus.create(subtask_id)
         email_id = 1001
         with self.assertRaisesRegexp(DuplicateTaskException, 'unable to find subtasks of instructor task'):
-            send_course_email(entry_id, email_id, to_list, global_email_context, subtask_status)
+            send_course_email(entry_id, email_id, to_list, global_email_context, subtask_status.to_dict())
 
     def test_send_email_missing_subtask(self):
         # test at a lower level, to ensure that the course gets checked down below too.
@@ -230,10 +230,10 @@ class TestEmailErrors(ModuleStoreTestCase):
         subtask_id = "subtask-id-value"
         initialize_subtask_info(entry, "emailed", 100, [subtask_id])
         different_subtask_id = "bogus-subtask-id-value"
-        subtask_status = create_subtask_status(different_subtask_id)
+        subtask_status = SubtaskStatus.create(different_subtask_id)
         bogus_email_id = 1001
         with self.assertRaisesRegexp(DuplicateTaskException, 'unable to find status for subtask of instructor task'):
-            send_course_email(entry_id, bogus_email_id, to_list, global_email_context, subtask_status)
+            send_course_email(entry_id, bogus_email_id, to_list, global_email_context, subtask_status.to_dict())
 
     def test_send_email_completed_subtask(self):
         # test at a lower level, to ensure that the course gets checked down below too.
@@ -241,14 +241,14 @@ class TestEmailErrors(ModuleStoreTestCase):
         entry_id = entry.id  # pylint: disable=E1101
         subtask_id = "subtask-id-value"
         initialize_subtask_info(entry, "emailed", 100, [subtask_id])
-        subtask_status = create_subtask_status(subtask_id, state=SUCCESS)
+        subtask_status = SubtaskStatus.create(subtask_id, state=SUCCESS)
         update_subtask_status(entry_id, subtask_id, subtask_status)
         bogus_email_id = 1001
         to_list = ['test@test.com']
         global_email_context = {'course_title': 'dummy course'}
-        new_subtask_status = create_subtask_status(subtask_id)
+        new_subtask_status = SubtaskStatus.create(subtask_id)
         with self.assertRaisesRegexp(DuplicateTaskException, 'already completed'):
-            send_course_email(entry_id, bogus_email_id, to_list, global_email_context, new_subtask_status)
+            send_course_email(entry_id, bogus_email_id, to_list, global_email_context, new_subtask_status.to_dict())
 
     def test_send_email_running_subtask(self):
         # test at a lower level, to ensure that the course gets checked down below too.
@@ -256,14 +256,14 @@ class TestEmailErrors(ModuleStoreTestCase):
         entry_id = entry.id  # pylint: disable=E1101
         subtask_id = "subtask-id-value"
         initialize_subtask_info(entry, "emailed", 100, [subtask_id])
-        subtask_status = create_subtask_status(subtask_id)
+        subtask_status = SubtaskStatus.create(subtask_id)
         update_subtask_status(entry_id, subtask_id, subtask_status)
         check_subtask_is_valid(entry_id, subtask_id, subtask_status)
         bogus_email_id = 1001
         to_list = ['test@test.com']
         global_email_context = {'course_title': 'dummy course'}
         with self.assertRaisesRegexp(DuplicateTaskException, 'already being executed'):
-            send_course_email(entry_id, bogus_email_id, to_list, global_email_context, subtask_status)
+            send_course_email(entry_id, bogus_email_id, to_list, global_email_context, subtask_status.to_dict())
 
     def test_send_email_retried_subtask(self):
         # test at a lower level, to ensure that the course gets checked down below too.
@@ -271,19 +271,19 @@ class TestEmailErrors(ModuleStoreTestCase):
         entry_id = entry.id  # pylint: disable=E1101
         subtask_id = "subtask-id-value"
         initialize_subtask_info(entry, "emailed", 100, [subtask_id])
-        subtask_status = create_subtask_status(subtask_id, state=RETRY, retried_nomax=2)
+        subtask_status = SubtaskStatus.create(subtask_id, state=RETRY, retried_nomax=2)
         update_subtask_status(entry_id, subtask_id, subtask_status)
         bogus_email_id = 1001
         to_list = ['test@test.com']
         global_email_context = {'course_title': 'dummy course'}
         # try running with a clean subtask:
-        new_subtask_status = create_subtask_status(subtask_id)
+        new_subtask_status = SubtaskStatus.create(subtask_id)
         with self.assertRaisesRegexp(DuplicateTaskException, 'already retried'):
-            send_course_email(entry_id, bogus_email_id, to_list, global_email_context, new_subtask_status)
+            send_course_email(entry_id, bogus_email_id, to_list, global_email_context, new_subtask_status.to_dict())
         # try again, with a retried subtask with lower count:
-        new_subtask_status = create_subtask_status(subtask_id, state=RETRY, retried_nomax=1)
+        new_subtask_status = SubtaskStatus.create(subtask_id, state=RETRY, retried_nomax=1)
         with self.assertRaisesRegexp(DuplicateTaskException, 'already retried'):
-            send_course_email(entry_id, bogus_email_id, to_list, global_email_context, new_subtask_status)
+            send_course_email(entry_id, bogus_email_id, to_list, global_email_context, new_subtask_status.to_dict())
 
     def dont_test_send_email_undefined_email(self):
         # test at a lower level, to ensure that the course gets checked down below too.
@@ -293,10 +293,10 @@ class TestEmailErrors(ModuleStoreTestCase):
         global_email_context = {'course_title': 'dummy course'}
         subtask_id = "subtask-id-value"
         initialize_subtask_info(entry, "emailed", 100, [subtask_id])
-        subtask_status = create_subtask_status(subtask_id)
+        subtask_status = SubtaskStatus.create(subtask_id)
         bogus_email_id = 1001
         with self.assertRaises(CourseEmail.DoesNotExist):
             # we skip the call that updates subtask status, since we've not set up the InstructorTask
             # for the subtask, and it's not important to the test.
             with patch('bulk_email.tasks.update_subtask_status'):
-                send_course_email(entry_id, bogus_email_id, to_list, global_email_context, subtask_status)
+                send_course_email(entry_id, bogus_email_id, to_list, global_email_context, subtask_status.to_dict())

lms/djangoapps/bulk_email/tests/test_tasks.py

@@ -31,7 +31,7 @@ from django.core.management import call_command
 from bulk_email.models import CourseEmail, Optout, SEND_TO_ALL
 
 from instructor_task.tasks import send_bulk_course_email
-from instructor_task.subtasks import update_subtask_status
+from instructor_task.subtasks import update_subtask_status, SubtaskStatus
 from instructor_task.models import InstructorTask
 from instructor_task.tests.test_base import InstructorTaskCourseTestCase
 from instructor_task.tests.factories import InstructorTaskFactory
@@ -63,16 +63,9 @@ def my_update_subtask_status(entry_id, current_task_id, new_subtask_status):
     entry = InstructorTask.objects.get(pk=entry_id)
     subtask_dict = json.loads(entry.subtasks)
     subtask_status_info = subtask_dict['status']
-    current_subtask_status = subtask_status_info[current_task_id]
-
-    def _get_retry_count(subtask_result):
-        """Return the number of retries counted for the given subtask."""
-        retry_count = subtask_result.get('retried_nomax', 0)
-        retry_count += subtask_result.get('retried_withmax', 0)
-        return retry_count
-
-    current_retry_count = _get_retry_count(current_subtask_status)
-    new_retry_count = _get_retry_count(new_subtask_status)
+    current_subtask_status = SubtaskStatus.from_dict(subtask_status_info[current_task_id])
+    current_retry_count = current_subtask_status.get_retry_count()
+    new_retry_count = new_subtask_status.get_retry_count()
     if current_retry_count <= new_retry_count:
         update_subtask_status(entry_id, current_task_id, new_subtask_status)