Сельдерей отправить почту за каждую неудачную задачу

Question

Я использую сельдерей в своем приложении для джанго, и у меня есть цветок сельдерея, чтобы следить за задачами сельдерея. У меня есть задачи настройки, когда электронные письма отправляются пользователю, когда они регистрируют/отправляют/FP и т. Д. События. Теперь Цветок дает мне приятные детали задания и его статус. Теперь для каждой неудачной задачи мне нужно отправить электронное письмо на мою учетную запись, чтобы я не проверял цветок каждый день для неудавшейся задачи. я сделал следующую конфигурацию в моем файле settings.py

CELERY_SEND_TASK_ERROR_EMAILS = True 

и ADMINS.

EMAIL_USE_TLS = True 
EMAIL_HOST_USER = 'noreply@xyz.com' 
EMAIL_HOST_PASSWORD = 'xyz123@' 
DEFAULT_FROM_EMAIL = EMAIL_HOST_USER 
SERVER_EMAIL = EMAIL_HOST_USER 
EMAIL_HOST = 'xyz.abc.com' 
EMAIL_BACKEND = 'django.core.mail.backends.smtp.EmailBackend' 

Это настройки адреса электронной почты «От». Несколько дней назад один из членов моей команды случайно изменил пароль вышеуказанного email_host и забыл обновить файл настроек. Это произошло не после того, как было слишком поздно, что задачи терпят неудачу из-за ошибки аутентификации SMTP.

Есть ли способ обойти это, даже если возникла ошибка аутентификации SMTP, я сразу получаю письмо от сельдерея? Я не уверен в этом.

Есть ли другие инструменты, которые будут контролировать мои задачи и для каждой неудачной задачи он пришлет мне почту.

Answer 1

Я узнал, что существует таблица, созданная для ведения задач в базе данных. Поэтому я просто создал скрипт, который каждый час проверяет неудачную задачу за прошлые записи часа в таблице, и если обнаружит, что он отправит электронное письмо.

script.py

#!venv/bin/python2 

import os 
from django.conf import settings 

if __name__ == '__main__' and __package__ is None: 
    os.sys.path.append(
     os.path.dirname(
      os.path.dirname(
       os.path.abspath(__file__)))) 

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "rest_apis.settings") 

import django 

django.setup() 
from django.core.mail import EmailMessage 
from djcelery.models import TaskMeta 
from datetime import datetime, timedelta, time 


USERS_TO_NOTIFY = ['ops@yopmail.com'] 
TIME_THRESHOLD_INTERVAL = 60 

def send_email(email_subject_line, email_body): 
    email = EmailMessage(email_subject_line, 
         email_body, 
         settings.EMAIL_HOST_USER, 
         USERS_TO_NOTIFY 
         ) 
    email.send() 


def main(): 
    current_time = datetime.now() # Get Current TimeStamp 
    time_threshold = current_time - timedelta(minutes=TIME_THRESHOLD_INTERVAL) # Get 60 minutes past current time stamp 
    celery_taskmeta_objects = TaskMeta.objects.filter(status="FAILURE", date_done__gte=time_threshold) 

    email_body = "Below are the tasks which failed : " 
    if celery_taskmeta_objects.exists(): 
     for celery_taskmeta in celery_taskmeta_objects: 
      print celery_taskmeta.task_id 
      email_body += "\n\ntask_id : %s" % celery_taskmeta.task_id 
      email_body += "\nstatus : %s" % celery_taskmeta.status 
      email_body += "\ndate : %s" % celery_taskmeta.date_done 
      email_body += "\ntraceback :" 
      email_body += "\n%s\n\n" % celery_taskmeta.traceback 
     email_subject_line = '[URGENT] Celery task failure in last %s minutes' % (TIME_THRESHOLD_INTERVAL) 
     send_email(email_subject_line, email_body) 


main() 

Теперь по электронной почте я получаю полный трассировки стека тоже и идентификатор задачи. Теперь мое требование состояло в том, чтобы проверять каждый час, поэтому я просто поставил скрипт в crontab. Теперь вы можете изменить порог времени в соответствии с вашими основными потребностями и работать соответственно.

Answer 2

Вы можете настроить Datadog https://www.datadoghq.com/

Это может контролировать и отслеживать динамическую инфраструктуру. Вы получите электронное письмо для каждой ошибки, возникшей на вашем производственном сервере.

Answer 3

Обновление для сельдерея 4. Tasks no longer sends error emails. Как сообщили в подобной проблеме сельдерея Handling error emails going forward?:

Вы можете реализовать свою собственную базовую задачу, как это для обработки ошибок однако вы хотели бы:

class MyTask(celery.Task): 
    def on_failure(self, exc, task_id, args, kwargs, einfo): 
     # here you can place mail_admins(...) call 
     print('{0!r} failed: {1!r}'.format(task_id, exc))