Airflow 에러 발생 시 Slack 알림받아보기 - TIL230619

📚 KDT WEEK 12 DAY 1 TIL

• Airflow에서 Slack 알림 처리하기

---

🟥 Slack 알림 처리하기

 

1. Slack API 페이지에서 알림받을 채널 설정하기

 

먼저, Slack API 페이지에 접속한다.

https://api.slack.com/messaging/webhooks

Sending messages using Incoming Webhooks

 

오른쪽 상단의 Your Apps를 누르고, 새로운 앱을 만든다.

 

1 > 2 순서대로 누른다.

 

첫 번째 옵션(From scratch)을 사용해서 만들어본다.

 

From scratch를 누른다.

 

알림 받고자하는 워크스페이스를 설정한다.

App Name은 원하는 이름을 넣어주면 된다.

 

나는 테스트용 워크스페이스를 하나파서, 설정해주었다.

 

빨간 상자안의 Incoming Webhooks를 누른다.

 

Slack에 알림을 보내기 위한 옵션이다.

 

Activate Incoming Webhooks를 On 해줘야 밑으로 채널을 설정할 수 있는 옵션이 생긴다.

On을 눌렀다면, 늘어난 창 아래로 Add New Webhook to Workspace 를 눌러 채널을 추가해보자.

 

1 > 2 순서대로 누른다.

 

아까 설정했던 워크스페이스안의 채널들 중 하나를 고를 수 있다.

알림받을 채널을 골라 설정한뒤, 허용 버튼을 눌러 추가한다.

 

알림받을 채널을 새로 생성해서 설정했다.

 

채널 설정이 완료되면, 그 채널의 Webhook URL이 생긴다.

이 URL에 json 형태의 POST를 보내면, 그 채널에 알림을 보낼 수 있게된다.

Mac 유저라면 Sample curl request to post to a channel 부분을 복사해서 터미널에서 실험해볼 수 있다.

 

이 Webhook URL을 복사해두자

 

나는 이 Webhook URL을 Airflow 웹 UI에 담아두었다.

Admin > Variables 에서 slack_url 이란 이름으로 담아두었다.

Webhook URL의 전체를 담지 않고, 그 채널의 유니크 ID로 보이는 부분만 저장했다.

 

 

2. Slack 알림을 보낼 파이썬 코드 작성하기

 

이제는 slack 알림을 보내기 위한 파이썬 코드를 하나 작성해보자.

나는 dags 폴더안에 plugins 폴더를 만들어서, slack.py 이름으로 작성했다.

 

from airflow.models import Variable
import logging
import requests

# dag 실패시 호출
def on_failure_callback(context):

    text = str(context['task_instance'])
    text += "```" + str(context.get('exception')) +"```"
    send_message_to_a_slack_channel(text, ":scream:")

# slack에 알림 보내기
def send_message_to_a_slack_channel(message, emoji):
    url = "https://hooks.slack.com/services/"+Variable.get("slack_url")
    headers = {
        'content-type': 'application/json',
    }
    data = { "username": "Data GOD", "text": message, "icon_emoji": emoji }
    r = requests.post(url, json=data, headers=headers)
    return r

 

dag 실패시 안에서 호출될 on_failure_callback 함수와, slack에 알림을 보낼 send_message_to_a_slack_channel 함수를 만들었다.

 

on_failure_callback함수는 호출되면서 airflow가 context 형태의 데이터를 담아준다.

context안에는 dag의 여러 정보가 포함되어있는데,

'task_instance' 로 dag의 이름을 가져오고, 'exception'을 통해 에러 내용을 가져온다.

 

url에 아까 부여받은 Webhook URL을 넣어준다.

헤더에 json이라는 정보를 담고, request 모듈을 이용해 post로 보내주면 끝이다.

 

 

3. 실행시킬 dag에서 Slack 알림 처리하기

 

먼저 Slack 알림의 동작확인을 위해 일부러 에러를 만든다.

중요한 점은, raise를 통해 에러를 꼭 올려야 한다는 점이다.

나는 sql의 INSERT에 I를 하나 더붙여 일부러 에러를 냈다.

일부러 INSERT 앞에 I를 하나 더 붙여서 에러를 발생시켰다.

 

그리고 같은 dag 파이썬 코드에 다음을 추가한다.

 

from plugins import slack

...

dag = DAG(
	...
    
    default_args = {
        'on_failure_callback': slack.on_failure_callback, # 이 부분을 추가
    }
    
    ...
)

...

 

아까 slack 알림을 위해 작성했던 slack.py를 import하고,

작성한 DAG 부분의 default_args에 on_failure_callback 을 추가한다.

slack 파일안에 실패시 호출될 함수인 on_failure_callback을 지정했다.

 

그리고 dag를 실행해보면..

 

CLI에서 실행 후 에러가 발생했고,

 

정상적으로 Slack에 알림이 날아왔다!

 

 

++ 오류

 

원래 내 dag의 형태는 아래와 같았다.

 

default_args = {
    'retries': 1,
    'retry_delay': timedelta(minutes=3),
    'on_failure_callback': slack.on_failure_callback,
}

 

근데 default_args에 retries가 있으면 슬랙에 알림이 가지않았다ㅠ

GPT에게 물어보니 재시도 횟수를 붙여주면 on_failure_callback이 호출되지 않는다고 한다..

 

 

챗GPT를 백프로믿는건 아니지만.. 재시도 횟수를 함께 넣을 방법은 없었을까? 🥲