Skip to content

Brian/subtask status obj #1511

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
brianhw merged 3 commits into from
Oct 27, 2013
Merged

Brian/subtask status obj #1511

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4fd8913
Define and use SubtaskStatus class.
brianhw Oct 25, 2013
85a291c
Make more incremental use of SubtaskStatus.increment().
brianhw Oct 27, 2013
0ddc7b0
Refactor subtask creation to be more general.
brianhw Oct 27, 2013
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
191 changes: 60 additions & 131 deletions lms/djangoapps/bulk_email/tasks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,13 +40,10 @@
from courseware.courses import get_course, course_image_url
from instructor_task.models import InstructorTask
from instructor_task.subtasks import (
create_subtask_ids,
generate_items_for_subtask,
create_subtask_status,
increment_subtask_status,
update_subtask_status,
initialize_subtask_info,
SubtaskStatus,
queue_subtasks_for_query,
check_subtask_is_valid,
update_subtask_status,
)

log = get_task_logger(__name__)
Expand Down Expand Up @@ -155,10 +152,8 @@ def _get_course_email_context(course):
def perform_delegate_email_batches(entry_id, course_id, task_input, action_name):
"""
Delegates emails by querying for the list of recipients who should
get the mail, chopping up into batches of settings.BULK_EMAIL_EMAILS_PER_TASK size,
and queueing up worker jobs.

Returns the number of batches (workers) kicked off.
get the mail, chopping up into batches of no more than settings.BULK_EMAIL_EMAILS_PER_TASK
in size, and queueing up worker jobs.
"""
entry = InstructorTask.objects.get(pk=entry_id)
# Get inputs to use in this task from the entry.
Expand Down Expand Up @@ -209,55 +204,37 @@ def perform_delegate_email_batches(entry_id, course_id, task_input, action_name)
to_option = email_obj.to_option
global_email_context = _get_course_email_context(course)

# Figure out the number of needed subtasks, getting id values to use for each.
recipient_qset = _get_recipient_queryset(user_id, to_option, course_id, course.location)
total_num_emails = recipient_qset.count()
subtask_id_list = create_subtask_ids(total_num_emails, settings.BULK_EMAIL_EMAILS_PER_QUERY, settings.BULK_EMAIL_EMAILS_PER_TASK)

# Update the InstructorTask with information about the subtasks we've defined.
log.info("Task %s: Preparing to update task for sending %d emails for course %s, email %s, to_option %s",
task_id, total_num_emails, course_id, email_id, to_option)
progress = initialize_subtask_info(entry, action_name, total_num_emails, subtask_id_list)

# Construct a generator that will return the recipients to use for each subtask.
# Pass in the desired fields to fetch for each recipient.
recipient_fields = ['profile__name', 'email']
recipient_generator = generate_items_for_subtask(
recipient_qset,
recipient_fields,
total_num_emails,
settings.BULK_EMAIL_EMAILS_PER_QUERY,
settings.BULK_EMAIL_EMAILS_PER_TASK,
)

# Now create the subtasks, and start them running. This allows all the subtasks
# in the list to be submitted at the same time.
num_subtasks = len(subtask_id_list)
log.info("Task %s: Preparing to generate and queue %s subtasks for course %s, email %s, to_option %s",
task_id, num_subtasks, course_id, email_id, to_option)
num_subtasks = 0
for recipient_list in recipient_generator:
subtask_id = subtask_id_list[num_subtasks]
num_subtasks += 1
subtask_status = create_subtask_status(subtask_id)
def _create_send_email_subtask(to_list, initial_subtask_status):
"""Creates a subtask to send email to a given recipient list."""
subtask_id = initial_subtask_status.task_id
new_subtask = send_course_email.subtask(
(
entry_id,
email_id,
recipient_list,
to_list,
global_email_context,
subtask_status,
initial_subtask_status.to_dict(),
),
task_id=subtask_id,
routing_key=settings.BULK_EMAIL_ROUTING_KEY,
)
new_subtask.apply_async()
return new_subtask

recipient_qset = _get_recipient_queryset(user_id, to_option, course_id, course.location)
recipient_fields = ['profile__name', 'email']

# Sanity check: we expect the subtask to be properly summing to the original count:
if num_subtasks != len(subtask_id_list):
error_msg = "Task {}: number of tasks generated {} not equal to original total {}".format(task_id, num_subtasks, len(subtask_id_list))
log.error(error_msg)
raise ValueError(error_msg)
log.info("Task %s: Preparing to queue subtasks for sending emails for course %s, email %s, to_option %s",
task_id, course_id, email_id, to_option)

progress = queue_subtasks_for_query(
entry,
action_name,
_create_send_email_subtask,
recipient_qset,
recipient_fields,
settings.BULK_EMAIL_EMAILS_PER_QUERY,
settings.BULK_EMAIL_EMAILS_PER_TASK
)

# We want to return progress here, as this is what will be stored in the
# AsyncResult for the parent task as its return value.
Expand All @@ -268,7 +245,7 @@ def perform_delegate_email_batches(entry_id, course_id, task_input, action_name)


@task(default_retry_delay=settings.BULK_EMAIL_DEFAULT_RETRY_DELAY, max_retries=settings.BULK_EMAIL_MAX_RETRIES) # pylint: disable=E1102
def send_course_email(entry_id, email_id, to_list, global_email_context, subtask_status):
def send_course_email(entry_id, email_id, to_list, global_email_context, subtask_status_dict):
"""
Sends an email to a list of recipients.

Expand All @@ -282,7 +259,7 @@ def send_course_email(entry_id, email_id, to_list, global_email_context, subtask
* `global_email_context`: dict containing values that are unique for this email but the same
for all recipients of this email. This dict is to be used to fill in slots in email
template. It does not include 'name' and 'email', which will be provided by the to_list.
* `subtask_status` : dict containing values representing current status. Keys are:
* `subtask_status_dict` : dict containing values representing current status. Keys are:

'task_id' : id of subtask. This is used to pass task information across retries.
'attempted' : number of attempts -- should equal succeeded plus failed
Expand All @@ -302,7 +279,8 @@ def send_course_email(entry_id, email_id, to_list, global_email_context, subtask
Emails are sent multi-part, in both plain text and html. Updates InstructorTask object
with status information (sends, failures, skips) and updates number of subtasks completed.
"""
current_task_id = subtask_status['task_id']
subtask_status = SubtaskStatus.from_dict(subtask_status_dict)
current_task_id = subtask_status.task_id
num_to_send = len(to_list)
log.info("Preparing to send email %s to %d recipients as subtask %s for instructor task %d: context = %s, status=%s",
email_id, num_to_send, current_task_id, entry_id, global_email_context, subtask_status)
Expand Down Expand Up @@ -336,8 +314,8 @@ def send_course_email(entry_id, email_id, to_list, global_email_context, subtask
# We got here for really unexpected reasons. Since we don't know how far
# the task got in emailing, we count all recipients as having failed.
# It at least keeps the counts consistent.
new_subtask_status = increment_subtask_status(subtask_status, failed=num_to_send, state=FAILURE)
update_subtask_status(entry_id, current_task_id, new_subtask_status)
subtask_status.increment(failed=num_to_send, state=FAILURE)
update_subtask_status(entry_id, current_task_id, subtask_status)
raise

if send_exception is None:
Expand Down Expand Up @@ -419,37 +397,20 @@ def _send_course_email(entry_id, email_id, to_list, global_email_context, subtas
* `global_email_context`: dict containing values that are unique for this email but the same
for all recipients of this email. This dict is to be used to fill in slots in email
template. It does not include 'name' and 'email', which will be provided by the to_list.
* `subtask_status` : dict containing values representing current status. Keys are:

'task_id' : id of subtask. This is used to pass task information across retries.
'attempted' : number of attempts -- should equal succeeded plus failed
'succeeded' : number that succeeded in processing
'skipped' : number that were not processed.
'failed' : number that failed during processing
'retried_nomax' : number of times the subtask has been retried for conditions that
should not have a maximum count applied
'retried_withmax' : number of times the subtask has been retried for conditions that
should have a maximum count applied
'state' : celery state of the subtask (e.g. QUEUING, PROGRESS, RETRY, FAILURE, SUCCESS)
* `subtask_status` : object of class SubtaskStatus representing current status.

Sends to all addresses contained in to_list that are not also in the Optout table.
Emails are sent multi-part, in both plain text and html.

Returns a tuple of two values:
* First value is a dict which represents current progress at the end of this call. Keys are
the same as for the input subtask_status.
* First value is a SubtaskStatus object which represents current progress at the end of this call.

* Second value is an exception returned by the innards of the method, indicating a fatal error.
In this case, the number of recipients that were not sent have already been added to the
'failed' count above.
"""
# Get information from current task's request:
task_id = subtask_status['task_id']

# collect stats on progress:
num_optout = 0
num_sent = 0
num_error = 0
task_id = subtask_status.task_id

try:
course_email = CourseEmail.objects.get(id=email_id)
Expand All @@ -463,8 +424,9 @@ def _send_course_email(entry_id, email_id, to_list, global_email_context, subtas
# attempt. Anyone on the to_list on a retry has already passed the filter
# that existed at that time, and we don't need to keep checking for changes
# in the Optout list.
if (subtask_status['retried_nomax'] + subtask_status['retried_withmax']) == 0:
if subtask_status.get_retry_count() == 0:
to_list, num_optout = _filter_optouts_from_recipients(to_list, course_email.course_id)
subtask_status.increment(skipped=num_optout)

course_title = global_email_context['course_title']
subject = "[" + course_title + "] " + course_email.subject
Expand Down Expand Up @@ -509,7 +471,7 @@ def _send_course_email(entry_id, email_id, to_list, global_email_context, subtas
# for a period of time between all emails within this task. Choice of
# the value depends on the number of workers that might be sending email in
# parallel, and what the SES throttle rate is.
if subtask_status['retried_nomax'] > 0:
if subtask_status.retried_nomax > 0:
sleep(settings.BULK_EMAIL_RETRY_DELAY_BETWEEN_SENDS)

try:
Expand All @@ -527,21 +489,21 @@ def _send_course_email(entry_id, email_id, to_list, global_email_context, subtas
# This will fall through and not retry the message.
log.warning('Task %s: email with id %s not delivered to %s due to error %s', task_id, email_id, email, exc.smtp_error)
dog_stats_api.increment('course_email.error', tags=[_statsd_tag(course_title)])
num_error += 1
subtask_status.increment(failed=1)

except SINGLE_EMAIL_FAILURE_ERRORS as exc:
# This will fall through and not retry the message.
log.warning('Task %s: email with id %s not delivered to %s due to error %s', task_id, email_id, email, exc)
dog_stats_api.increment('course_email.error', tags=[_statsd_tag(course_title)])
num_error += 1
subtask_status.increment(failed=1)

else:
dog_stats_api.increment('course_email.sent', tags=[_statsd_tag(course_title)])
if settings.BULK_EMAIL_LOG_SENT_EMAILS:
log.info('Email with id %s sent to %s', email_id, email)
else:
log.debug('Email with id %s sent to %s', email_id, email)
num_sent += 1
subtask_status.increment(succeeded=1)

# Pop the user that was emailed off the end of the list only once they have
# successfully been processed. (That way, if there were a failure that
Expand All @@ -552,16 +514,9 @@ def _send_course_email(entry_id, email_id, to_list, global_email_context, subtas
dog_stats_api.increment('course_email.infinite_retry', tags=[_statsd_tag(course_title)])
# Increment the "retried_nomax" counter, update other counters with progress to date,
# and set the state to RETRY:
subtask_progress = increment_subtask_status(
subtask_status,
succeeded=num_sent,
failed=num_error,
skipped=num_optout,
retried_nomax=1,
state=RETRY
)
subtask_status.increment(retried_nomax=1, state=RETRY)
return _submit_for_retry(
entry_id, email_id, to_list, global_email_context, exc, subtask_progress, skip_retry_max=True
entry_id, email_id, to_list, global_email_context, exc, subtask_status, skip_retry_max=True
)

except LIMITED_RETRY_ERRORS as exc:
Expand All @@ -571,16 +526,9 @@ def _send_course_email(entry_id, email_id, to_list, global_email_context, subtas
dog_stats_api.increment('course_email.limited_retry', tags=[_statsd_tag(course_title)])
# Increment the "retried_withmax" counter, update other counters with progress to date,
# and set the state to RETRY:
subtask_progress = increment_subtask_status(
subtask_status,
succeeded=num_sent,
failed=num_error,
skipped=num_optout,
retried_withmax=1,
state=RETRY
)
subtask_status.increment(retried_withmax=1, state=RETRY)
return _submit_for_retry(
entry_id, email_id, to_list, global_email_context, exc, subtask_progress, skip_retry_max=False
entry_id, email_id, to_list, global_email_context, exc, subtask_status, skip_retry_max=False
)

except BULK_EMAIL_FAILURE_ERRORS as exc:
Expand All @@ -590,14 +538,8 @@ def _send_course_email(entry_id, email_id, to_list, global_email_context, subtas
task_id, email_id, num_pending)
# Update counters with progress to date, counting unsent emails as failures,
# and set the state to FAILURE:
subtask_progress = increment_subtask_status(
subtask_status,
succeeded=num_sent,
failed=(num_error + num_pending),
skipped=num_optout,
state=FAILURE
)
return subtask_progress, exc
subtask_status.increment(failed=num_pending, state=FAILURE)
return subtask_status, exc

except Exception as exc:
# Errors caught here cause the email to be retried. The entire task is actually retried
Expand All @@ -609,30 +551,17 @@ def _send_course_email(entry_id, email_id, to_list, global_email_context, subtas
task_id, email_id)
# Increment the "retried_withmax" counter, update other counters with progress to date,
# and set the state to RETRY:
subtask_progress = increment_subtask_status(
subtask_status,
succeeded=num_sent,
failed=num_error,
skipped=num_optout,
retried_withmax=1,
state=RETRY
)
subtask_status.increment(retried_withmax=1, state=RETRY)
return _submit_for_retry(
entry_id, email_id, to_list, global_email_context, exc, subtask_progress, skip_retry_max=False
entry_id, email_id, to_list, global_email_context, exc, subtask_status, skip_retry_max=False
)

else:
# All went well. Update counters with progress to date,
# and set the state to SUCCESS:
subtask_progress = increment_subtask_status(
subtask_status,
succeeded=num_sent,
failed=num_error,
skipped=num_optout,
state=SUCCESS
)
subtask_status.increment(state=SUCCESS)
# Successful completion is marked by an exception value of None.
return subtask_progress, None
return subtask_status, None
finally:
# Clean up at the end.
connection.close()
Expand Down Expand Up @@ -678,26 +607,26 @@ def _submit_for_retry(entry_id, email_id, to_list, global_email_context, current
successfully submitted, this value will be the RetryTaskError that retry() returns.
Otherwise, it (ought to be) the current_exception passed in.
"""
task_id = subtask_status['task_id']
task_id = subtask_status.task_id
log.info("Task %s: Successfully sent to %s users; failed to send to %s users (and skipped %s users)",
task_id, subtask_status['succeeded'], subtask_status['failed'], subtask_status['skipped'])
task_id, subtask_status.succeeded, subtask_status.failed, subtask_status.skipped)

# Calculate time until we retry this task (in seconds):
# The value for max_retries is increased by the number of times an "infinite-retry" exception
# has been retried. We want the regular retries to trigger max-retry checking, but not these
# special retries. So we count them separately.
max_retries = _get_current_task().max_retries + subtask_status['retried_nomax']
max_retries = _get_current_task().max_retries + subtask_status.retried_nomax
base_delay = _get_current_task().default_retry_delay
if skip_retry_max:
# once we reach five retries, don't increase the countdown further.
retry_index = min(subtask_status['retried_nomax'], 5)
retry_index = min(subtask_status.retried_nomax, 5)
exception_type = 'sending-rate'
# if we have a cap, after all, apply it now:
if hasattr(settings, 'BULK_EMAIL_INFINITE_RETRY_CAP'):
retry_cap = settings.BULK_EMAIL_INFINITE_RETRY_CAP + subtask_status['retried_withmax']
retry_cap = settings.BULK_EMAIL_INFINITE_RETRY_CAP + subtask_status.retried_withmax
max_retries = min(max_retries, retry_cap)
else:
retry_index = subtask_status['retried_withmax']
retry_index = subtask_status.retried_withmax
exception_type = 'transient'

# Skew the new countdown value by a random factor, so that not all
Expand All @@ -722,7 +651,7 @@ def _submit_for_retry(entry_id, email_id, to_list, global_email_context, current
email_id,
to_list,
global_email_context,
subtask_status,
subtask_status.to_dict(),
],
exc=current_exception,
countdown=countdown,
Expand All @@ -743,8 +672,8 @@ def _submit_for_retry(entry_id, email_id, to_list, global_email_context, current
log.exception('Task %s: email with id %d caused send_course_email task to fail to retry. To list: %s',
task_id, email_id, [i['email'] for i in to_list])
num_failed = len(to_list)
new_subtask_progress = increment_subtask_status(subtask_status, failed=num_failed, state=FAILURE)
return new_subtask_progress, retry_exc
subtask_status.increment(subtask_status, failed=num_failed, state=FAILURE)
return subtask_status, retry_exc


def _statsd_tag(course_title):
Expand Down
Loading