[Airflow] SLA(Service Layer Agreement) 서비스 수준 계약 정리

Airflow에서는 실패 시 함수를 호출하는 on_failure_callback 외에도 서비스 수준 계약 개념을 적용할 수 있습니다.

관련해서 SLA와 어떻게 적용하는 지 확인 해보겠습니다.

---

1. SLA란?

• 서비스 제공자와 고객 간에 합의된 서비스 수준을 정의하는 계약
• SLA는 서비스의 성능, 가용성, 응답 시간, 처리 시간 등 다양한 측면에서 서비스의 품질을 측정하고 보장하기 위해 사용

주요 요소

1. 서비스 설명: 제공되는 서비스의 세부 사항을 명확히 설명
2. 성능 기준: 서비스의 성능을 평가하기 위한 측정 기준을 설정
   예를 들어, 시스템의 가용성, 응답 시간, 문제 해결 시간 등이 포함될 수 있음
3. 가용성: 서비스가 사용자에게 제공되는 시간의 비율을 정의합니다. 예를 들어, 연간 99.9%의 가용성은 연간 약 8.76시간의 다운타임을 의미
4. 응답 시간: 고객 요청에 대한 서비스 제공자의 응답 시간을 정의
5. 문제 해결 시간: 서비스 장애 또는 문제 발생 시 해결까지 걸리는 최대 시간을 정의
6. 벌칙 및 보상: SLA를 준수하지 못했을 때 서비스 제공자가 고객에게 제공해야 하는 벌칙 또는 보상을 정의
7. 모니터링 및 보고: 서비스 성능을 모니터링하고 정기적으로 보고하는 방법을 정의

---

2. Airflow에서 SLA

• Apache Airflow에서는 특정 작업(Task)의 완료 시간이 SLA를 초과할 경우 이를 감지하고 알림을 보낼 수 있음
• 워크플로우의 성능을 모니터링하고 문제 발생 시 신속하게 대응
• 작업(Task) 단위로 SLA를 설정하여 성능을 모니터링

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    'sla': timedelta(hours=1),  # SLA 설정: 작업이 1시간 이내에 완료되어야 합니다.
}

with DAG('example_sla_dag', default_args=default_args, schedule_interval='@daily') as dag:
    start = DummyOperator(task_id='start')
    task_with_sla = DummyOperator(task_id='task_with_sla', sla=timedelta(minutes=30))  # 개별 작업에 대한 SLA 설정
    end = DummyOperator(task_id='end')

    start >> task_with_sla >> end

SLA정보 저장 위치

• metastore의 sla_miss 테이블에 저장

 

UI에서 확인

• Browse - SLA Misses에서 확인 가능

---

3. SLA_MISS_CALLBACK

• SLA (Service Level Agreement)를 초과하는 작업(Task)이 발생했을 때 호출되는 콜백 함수
• dag가 성공하거나 실패할 때 호출되는 함수인 on_success_callback, on_failure_callback와 비슷한 원리
• slack 및 email에 알림을 전송할 수 있게 함수를 구성할 수 있음

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2023, 1, 1),
    'sla': timedelta(minutes=30),  # 전체 DAG에 대한 SLA 설정
    'sla_miss_callback': sla_miss_callback,  # SLA miss callback 함수 설정
}

with DAG('example_sla_dag', default_args=default_args, schedule_interval='@daily') as dag:
    start = DummyOperator(task_id='start')
    
    task_with_sla = DummyOperator(
        task_id='task_with_sla',
        sla=timedelta(minutes=10)  # 개별 작업에 대한 SLA 설정
    )
    
    end = DummyOperator(task_id='end')

    start >> task_with_sla >> end