I have airflow version 1.10.12 installed and I am using “SlackWebhookOperator” to send alert to slack channel with slack incoming webhook. I have created an Airflow connection for Slack with HTTP connection. I have below callback function defined in my dag file.
SLACK_CONN_ID = ‘slack’
def task_fail_slack_alert(context):
slack_webhook_token = BaseHook.get_connection(SLACK_CONN_ID).password
slack_msg = “”"
Task Failed.
Task: {task}
Dag: {dag}
Execution Time: {exec_date}
Log Url: {log_url}
“”".format(
task=context.get(‘task_instance’).task_id,
dag=context.get(‘task_instance’).dag_id,
ti=context.get(‘task_instance’),
exec_date=context.get(‘execution_date’),
log_url=context.get(‘task_instance’).log_url,
)
failed_alert = SlackWebhookOperator(
task_id=‘slack_test’,
http_conn_id=‘slack’,
webhook_token=slack_webhook_token,
channel=‘airflow-alerts’,
message=slack_msg,
username=‘airflow’,
dag=dag)
return failed_alert.execute(context=context)
default_args = {
‘owner’: ‘airflow’,
‘depends_on_past’: False,
‘start_date’: datetime(2020, 5, 21),
‘email’: [‘airflow@example.com’],
‘email_on_failure’: False,
‘email_on_retry’: False,
‘retries’: 1,
‘retry_delay’: timedelta(minutes=5),
‘on_failure_callback’: task_fail_slack_alert
}
with DAG(
‘slack_usage’,
default_args=default_args,
description=‘A simple DAG with Slack Alerts’,
schedule_interval=timedelta(days=1),
) as dag:
I have a task that fails so that the callback function would send an alert to slack channel, but couldn’t achieve the same. I am getting below error:
HTTPSConnectionPool(host=‘hooks.slack.com’, port=443): Max retries exceeded with url: /services/T000000000/B000000000000/sGrU1bco8trMOFvj94XXXXXXX (Caused by NewConnectionError(’<urllib3.connection.HTTPSConnection object at 0x7fcca985c0b8>: Failed to establish a new connection: [Errno 110] Connection timed out’,))
It works fine when I post a message from airflow scheduler and webserver pods to slack with the slack webhook.
Can someone please help me on this.
Thanks.