Airflow - Уведомление по электронной почте не работает с SES

Question

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

• Ссылка 1
• Ссылка 2

DAG One: (успешное задание)

from datetime import datetime
from datetime import timedelta
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator

default_args = {
    'owner': 'Airflow',
    'depends_on_past': False,
    'start_date': datetime(2015, 6, 1),
    'email': ['firstnamelastname@company.com','firstnamelastname@company.com'],
    'email_on_failure': True,
    'email_on_retry': True,
    'retries': 1,
    'retry_delay': timedelta(seconds=5),
    'email_on_success': True
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}


def print_hello():
    return 'Hello world!'

dag = DAG('success', description='Simple tutorial DAG',
          schedule_interval='0 12 * * *',default_args=default_args,
          start_date=datetime(2017, 3, 20), catchup=False)

dummy_operator = DummyOperator(task_id='dummy_task', retries=3, dag=dag)

hello_operator = PythonOperator(task_id='hello_task', python_callable=print_hello, dag=dag)

dummy_operator >> hello_operator

DAG Two: (невыполненное задание)

from datetime import datetime
from datetime import timedelta
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator

default_args = {
    'owner': 'Airflow',
    'depends_on_past': False,
    'start_date': datetime(2015, 6, 1),
    'email': ['firstnamelastname@company.com','firstnamelastname@company.com'],
    'email_on_failure': True,
    'email_on_retry': True,
    'retries': 1,
    'retry_delay': timedelta(seconds=5),
    'email_on_success': True
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}


def print_hello():
    xxxx
    return 'Hello world!'

dag = DAG('success', description='Simple tutorial DAG',
          schedule_interval='0 12 * * *',default_args=default_args,
          start_date=datetime(2017, 3, 20), catchup=False)

dummy_operator = DummyOperator(task_id='dummy_task', retries=3, dag=dag)

hello_operator = PythonOperator(task_id='hello_task', python_callable=print_hello, dag=dag)

dummy_operator >> hello_operator

Я ожидал получить электронное письмо для обеих работ. Поскольку оба задания содержат конфигурацию для email_on_success и email_on_failure Но я не получил ни одного письма.

Пожалуйста, посмотрите статистику выполнения задания:

Вот моя Конфигурация SMTP под airflow.cfg:

smtp_host = email-smtp.ap-south-1.amazonaws.com
smtp_starttls = True
smtp_ssl = False
# Uncomment and set the user/pass settings if you want to use SMTP AUTH
smtp_user = XXXXXXXXXXXXXXXXXXX
smtp_password = XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
smtp_port = 587
smtp_mail_from = firstnamelastname@company.com 

Я получил имя пользователя и пароль от Create My SMTP Credentials в Сервисе SES. У меня также есть подтвержденный адрес электронной почты. Группа безопасности для моего EC2 содержит весь исходящий трафик c для всего протокола, для всех портов и для пункта назначения 0.0.0.0/0

Что еще мне здесь не хватает? Можно ли настроить / сгенерировать логи для процесса отправки электронной почты?