호스트

Datadog 에이전트 Airflow 통합 구성

Datadog 에이전트 패키지에 포함된 Airflow 점검을 구성해 메트릭 상태와 서비스 점검을 수집합니다. 그러려면 에이전트 구성 디렉터리의 루트 수준에 있는 conf.d/ 폴더에서 airflow.d/conf.yaml 파일 내 url을 편집하세요. 그러면 Airflow 서비스 점검이 시작됩니다. 사용할 수 있는 구성 옵션 전체를 보려면 airflow.d/conf.yam 샘플을 참고하세요.

url이 내 Airflow 웹서버 base_url과 일치하는지 확인하세요. 이는 Airflow 인스턴스에 연결할 때 사용한 URL입니다.

Airflow를 DogStatsD에 연결

Airflow statsd 기능을 사용해 Airflow를 DogStatsD(Datadog 에이전트에 포함되어 있음)에 연결하여 메트릭을 수집하세요. 사용된 Airflow 버전에 따른 메트릭 보고서와 추가 구성 옵션에 관해 자세히 알아보려면 아래 Airflow 설명서를 참고하세요.

• Airflow 메트릭
• Airflow 메트릭 구성

참고: Airflow가 보고하는 StatsD 메트릭 존재 여부는 사용하는 Airflow Executor 종류에 따라 달라집니다. 예를 들어 KubernetesExecutor6의 경우 airflow.ti_failures/successes, airflow.operator_failures/successes, airflow.dag.task.duration이 보고되지 않습니다.

1. Airflow StatsD 플러그인을 설치합니다.

   pip install 'apache-airflow[statsd]'
   

2. Airflow 구성 파일 airflow.cfg에 다음 구성을 추가해 업데이트합니다.

   `statsd_datadog_enabled`를 true로 설정하지 마세요. `statsd_datadog_enabled`를 활성화하면 충돌이 발생할 수 있습니다. 문제를 예방하려면 이 변수를 `False`로 설정해야 합니다.

   [scheduler]
   statsd_on = True
   # Hostname or IP of server running the Datadog Agent
   statsd_host = localhost  
   # DogStatsD port configured in the Datadog Agent
   statsd_port = 8125
   statsd_prefix = airflow
   

3. Datadog 에이전트 주 구성 파일에 다음 구성을 추가해 업데이트합니다.

   # dogstatsd_mapper_cache_size: 1000  # default to 1000
   dogstatsd_mapper_profiles:
     - name: airflow
       prefix: "airflow."
       mappings:
         - match: "airflow.*_start"
           name: "airflow.job.start"
           tags:
             job_name: "$1"
         - match: "airflow.*_end"
           name: "airflow.job.end"
           tags:
             job_name: "$1"
         - match: "airflow.*_heartbeat_failure"
           name: airflow.job.heartbeat.failure
           tags:
             job_name: "$1"
         - match: "airflow.operator_failures_*"
           name: "airflow.operator_failures"
           tags:
             operator_name: "$1"
         - match: "airflow.operator_successes_*"
           name: "airflow.operator_successes"
           tags:
             operator_name: "$1"
         - match: 'airflow\.dag_processing\.last_runtime\.(.*)'
           match_type: "regex"
           name: "airflow.dag_processing.last_runtime"
           tags:
             dag_file: "$1"
         - match: 'airflow\.dag_processing\.last_run\.seconds_ago\.(.*)'
           match_type: "regex"
           name: "airflow.dag_processing.last_run.seconds_ago"
           tags:
             dag_file: "$1"
         - match: 'airflow\.dag\.loading-duration\.(.*)'
           match_type: "regex"
           name: "airflow.dag.loading_duration"
           tags:
             dag_file: "$1"
         - match: "airflow.dagrun.*.first_task_scheduling_delay"
           name: "airflow.dagrun.first_task_scheduling_delay"
           tags:
             dag_id: "$1"
         - match: "airflow.pool.open_slots.*"
           name: "airflow.pool.open_slots"
           tags:
             pool_name: "$1"
         - match: "airflow.pool.queued_slots.*"
           name: "airflow.pool.queued_slots"
           tags:
             pool_name: "$1"
         - match: "airflow.pool.running_slots.*"
           name: "airflow.pool.running_slots"
           tags:
             pool_name: "$1"
         - match: "airflow.pool.used_slots.*"
           name: "airflow.pool.used_slots"
           tags:
             pool_name: "$1"
         - match: "airflow.pool.starving_tasks.*"
           name: "airflow.pool.starving_tasks"
           tags:
             pool_name: "$1"
         - match: 'airflow\.dagrun\.dependency-check\.(.*)'
           match_type: "regex"
           name: "airflow.dagrun.dependency_check"
           tags:
             dag_id: "$1"
         - match: 'airflow\.dag\.(.*)\.([^.]*)\.duration'
           match_type: "regex"
           name: "airflow.dag.task.duration"
           tags:
             dag_id: "$1"
             task_id: "$2"
         - match: 'airflow\.dag_processing\.last_duration\.(.*)'
           match_type: "regex"
           name: "airflow.dag_processing.last_duration"
           tags:
             dag_file: "$1"
         - match: 'airflow\.dagrun\.duration\.success\.(.*)'
           match_type: "regex"
           name: "airflow.dagrun.duration.success"
           tags:
             dag_id: "$1"
         - match: 'airflow\.dagrun\.duration\.failed\.(.*)'
           match_type: "regex"
           name: "airflow.dagrun.duration.failed"
           tags:
             dag_id: "$1"
         - match: 'airflow\.dagrun\.schedule_delay\.(.*)'
           match_type: "regex"
           name: "airflow.dagrun.schedule_delay"
           tags:
             dag_id: "$1"
         - match: 'airflow.scheduler.tasks.running'
           name: "airflow.scheduler.tasks.running"
         - match: 'airflow.scheduler.tasks.starving'
           name: "airflow.scheduler.tasks.starving"
         - match: 'airflow.sla_email_notification_failure'
           name: 'airflow.sla_email_notification_failure'
         - match: 'airflow\.task_removed_from_dag\.(.*)'
           match_type: "regex"
           name: "airflow.dag.task_removed"
           tags:
             dag_id: "$1"
         - match: 'airflow\.task_restored_to_dag\.(.*)'
           match_type: "regex"
           name: "airflow.dag.task_restored"
           tags:
             dag_id: "$1"
         - match: "airflow.task_instance_created-*"
           name: "airflow.task.instance_created"
           tags:
             task_class: "$1"
         - match: 'airflow\.ti\.start\.(.+)\.(\w+)'
           match_type: regex
           name: airflow.ti.start
           tags: 
             dag_id: "$1"
             task_id: "$2"
         - match: 'airflow\.ti\.finish\.(\w+)\.(.+)\.(\w+)'
           name: airflow.ti.finish
           match_type: regex
           tags: 
             dag_id: "$1"
             task_id: "$2"
             state: "$3"
   

Datadog 에이전트와 Airflow 재시작

1. [에이전트를 재시작합니다] 9.
2. Airflow을 재시작하면 Airflow 메트릭이 에이전트 DogStatsD 엔드포인트로 전송되기 시작합니다.

통합 서비스 점검

airflow.d/conf.yaml 파일의 기본 구성을 사용해 Airflow 서비스 점검을 활성화하세요. 사용할 수 있는 전체 사용 옵션을 보려면 샘플airflow.d/conf.yaml을 참고하세요.

로그 수집

Agent 버전 6.0 이상에서 사용 가능

1. Datadog 에이전트에서 로그 수집은 기본적으로 사용하지 않도록 설정되어 있습니다. datadog.yaml 파일에서 로그 수집을 사용하도록 설정합니다.

   logs_enabled: true
   

2. airflow.d/conf.yaml 맨 아래에 있는 다음 구성 블록에 코멘트를 지우고 편집합니다. path와 service 파라미터 값을 변경하고 내 환경에 맞게 구성합니다.

   • DAG 프로세서 매니저와 스케쥴러 로그를 구성합니다.

     logs:
       - type: file
         path: "<PATH_TO_AIRFLOW>/logs/dag_processor_manager/dag_processor_manager.log"
         source: airflow
         log_processing_rules:
           - type: multi_line
             name: new_log_start_with_date
             pattern: \[\d{4}\-\d{2}\-\d{2}
       - type: file
         path: "<PATH_TO_AIRFLOW>/logs/scheduler/latest/*.log"
         source: airflow
         log_processing_rules:
           - type: multi_line
             name: new_log_start_with_date
             pattern: \[\d{4}\-\d{2}\-\d{2}
     

     매일 로그 회전을 통해 스케쥴러를 정기적으로 정리하는 것이 좋습니다.

   • DAG 작업 로그 추가 구성

     logs:
       - type: file
         path: "<PATH_TO_AIRFLOW>/logs/*/*/*/*.log"
         source: airflow
         log_processing_rules:
           - type: multi_line
             name: new_log_start_with_date
             pattern: \[\d{4}\-\d{2}\-\d{2}
     

     중요: 기본적으로 Airflow에서는 로그 파일 템플릿 log_filename_template = {{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log를 사용합니다. 정기적으로 제거하지 않으면 로그 파일 수가 빠른 속도로 늘어납니다. 각 실행 작업 로그를 개별적으로 Airflow UI에 표시하기 위해 이와 같은 패턴을 사용합니다.

     Airflow UI로 로그를 보지 않는 경우, Datadog에서는 airflow.cfg에 log_filename_template = dag_tasks.log 구성을 사용하기를 권장합니다. 그러면 로그가 이 파일을 회전하고 다음 구성을 사용합니다.

     logs:
       - type: file
         path: "<PATH_TO_AIRFLOW>/logs/dag_tasks.log"
         source: airflow
         log_processing_rules:
           - type: multi_line
             name: new_log_start_with_date
             pattern: \[\d{4}\-\d{2}\-\d{2}
     

3. 에이전트를 다시 시작합니다.

컨테이너화

Datadog 에이전트 Airflow 통합 구성

컨테이너화된 환경의 경우 자동탐지 통합 템플릿에 다음 파라미터를 적용하는 방법이 안내되어 있습니다.

url이 내 Airflow 웹서버 base_url과 일치하는지 확인하세요. 이는 Airflow 인스턴스에 연결할 때 사용한 URL입니다. localhost를 템플릿 변수 %%host%%로 변경하세요.

Airflow를 DogStatsD에 연결

Airflow statsd 기능을 사용해 Airflow를 DogStatsD(Datadog 에이전트에 포함되어 있음)에 연결하여 메트릭을 수집하세요. 사용된 Airflow 버전에 따른 메트릭 보고서와 추가 구성 옵션에 관해 자세히 알아보려면 아래 Airflow 설명서를 참고하세요.

• Airflow 메트릭
• Airflow 메트릭 구성

참고: Airflow가 보고하는 StatsD 메트릭 존재 여부는 사용하는 Airflow Executor 종류에 따라 달라집니다. 예를 들어 KubernetesExecutor5의 경우 airflow.ti_failures/successes, airflow.operator_failures/successes, airflow.dag.task.duration이 보고되지 않습니다.

참고: Airflow에 사용되는 환경 변수가 버전에 따라 다를 수 있습니다. 예를 들어 Airflow 2.0.0에서는 환경 변수 AIRFLOW__METRICS__STATSD_HOST를 활용하지만 Airflow 1.10.15에서는 AIRFLOW__SCHEDULER__STATSD_HOST를 사용합니다.

쿠버네티스 배포의 경우 다음 환경 변수를 사용해 Airflow StatsD 구성을 활성화할 수 있습니다.

env:
  - name: AIRFLOW__SCHEDULER__STATSD_ON
    value: "True"
  - name: AIRFLOW__SCHEDULER__STATSD_PORT
    value: "8125"
  - name: AIRFLOW__SCHEDULER__STATSD_PREFIX
    value: "airflow"
  - name: AIRFLOW__SCHEDULER__STATSD_HOST
    valueFrom:
      fieldRef:
        fieldPath: status.hostIP

호스트 엔드포인트 AIRFLOW__SCHEDULER__STATSD_HOST의 환경 변수는 노드의 호스트 IP 주소와 함께 제공되어 StatsD 데이터를 Airflow Pod와 같은 Pod인 Datadog 에이전트 Pod로 라우팅합니다. 이 설정을 사용하려면 에이전트의 hostPort인 8125 포트가 개방되어 있어야 하고 로컬이 아닌 StatsD 트래픽을 수락해야 합니다. 더 자세한 정보는 쿠버네티스에서 DogStatsD 설정을 참고하세요.

그러면 StatsD 트래픽이 수신 준비가 된 상태로 Airflow 컨테이너에서 Datadog 에이전트로 이동합니다. 마지막으로 실행할 단계는 Datadog 에이전트를 적합한 dogstatsd_mapper_profiles로 업데이트하는 것입니다. 그러려면 호스트 설치에 있는 dogstatsd_mapper_profiles를 복사해 datadog.yaml 파일에 붙여 넣으세요. 또는 환경 변수 DD_DOGSTATSD_MAPPER_PROFILES에서 동급의 JSON 구성을 사용해 Datadog 에이전트를 배포하세요. 쿠버네티스의 경우 동급 환경 변수 표기는 다음과 같습니다.

env: 
  - name: DD_DOGSTATSD_MAPPER_PROFILES
    value: >
      [{"name":"airflow","prefix":"airflow.","mappings":[{"match":"airflow.*_start","name":"airflow.job.start","tags":{"job_name":"$1"}},{"match":"airflow.*_end","name":"airflow.job.end","tags":{"job_name":"$1"}},{"match":"airflow.*_heartbeat_failure","name":"airflow.job.heartbeat.failure","tags":{"job_name":"$1"}},{"match":"airflow.operator_failures_*","name":"airflow.operator_failures","tags":{"operator_name":"$1"}},{"match":"airflow.operator_successes_*","name":"airflow.operator_successes","tags":{"operator_name":"$1"}},{"match":"airflow\\.dag_processing\\.last_runtime\\.(.*)","match_type":"regex","name":"airflow.dag_processing.last_runtime","tags":{"dag_file":"$1"}},{"match":"airflow\\.dag_processing\\.last_run\\.seconds_ago\\.(.*)","match_type":"regex","name":"airflow.dag_processing.last_run.seconds_ago","tags":{"dag_file":"$1"}},{"match":"airflow\\.dag\\.loading-duration\\.(.*)","match_type":"regex","name":"airflow.dag.loading_duration","tags":{"dag_file":"$1"}},{"match":"airflow.dagrun.*.first_task_scheduling_delay","name":"airflow.dagrun.first_task_scheduling_delay","tags":{"dag_id":"$1"}},{"match":"airflow.pool.open_slots.*","name":"airflow.pool.open_slots","tags":{"pool_name":"$1"}},{"match":"airflow.pool.queued_slots.*","name":"airflow.pool.queued_slots","tags":{"pool_name":"$1"}},{"match":"airflow.pool.running_slots.*","name":"airflow.pool.running_slots","tags":{"pool_name":"$1"}},{"match":"airflow.pool.used_slots.*","name":"airflow.pool.used_slots","tags":{"pool_name":"$1"}},{"match":"airflow.pool.starving_tasks.*","name":"airflow.pool.starving_tasks","tags":{"pool_name":"$1"}},{"match":"airflow\\.dagrun\\.dependency-check\\.(.*)","match_type":"regex","name":"airflow.dagrun.dependency_check","tags":{"dag_id":"$1"}},{"match":"airflow\\.dag\\.(.*)\\.([^.]*)\\.duration","match_type":"regex","name":"airflow.dag.task.duration","tags":{"dag_id":"$1","task_id":"$2"}},{"match":"airflow\\.dag_processing\\.last_duration\\.(.*)","match_type":"regex","name":"airflow.dag_processing.last_duration","tags":{"dag_file":"$1"}},{"match":"airflow\\.dagrun\\.duration\\.success\\.(.*)","match_type":"regex","name":"airflow.dagrun.duration.success","tags":{"dag_id":"$1"}},{"match":"airflow\\.dagrun\\.duration\\.failed\\.(.*)","match_type":"regex","name":"airflow.dagrun.duration.failed","tags":{"dag_id":"$1"}},{"match":"airflow\\.dagrun\\.schedule_delay\\.(.*)","match_type":"regex","name":"airflow.dagrun.schedule_delay","tags":{"dag_id":"$1"}},{"match":"airflow.scheduler.tasks.running","name":"airflow.scheduler.tasks.running"},{"match":"airflow.scheduler.tasks.starving","name":"airflow.scheduler.tasks.starving"},{"match":"airflow.sla_email_notification_failure","name":"airflow.sla_email_notification_failure"},{"match":"airflow\\.task_removed_from_dag\\.(.*)","match_type":"regex","name":"airflow.dag.task_removed","tags":{"dag_id":"$1"}},{"match":"airflow\\.task_restored_to_dag\\.(.*)","match_type":"regex","name":"airflow.dag.task_restored","tags":{"dag_id":"$1"}},{"match":"airflow.task_instance_created-*","name":"airflow.task.instance_created","tags":{"task_class":"$1"}},{"match":"airflow\\.ti\\.start\\.(.+)\\.(\\w+)","match_type":"regex","name":"airflow.ti.start","tags":{"dag_id":"$1","task_id":"$2"}},{"match":"airflow\\.ti\\.finish\\.(\\w+)\\.(.+)\\.(\\w+)","name":"airflow.ti.finish","match_type":"regex","tags":{"dag_id":"$1","task_id":"$2","state":"$3"}}]}]      

로그 수집

Agent 버전 6.0 이상에서 사용 가능

Datadog 에이전트에서 로그 수집은 기본값으로 비활성화되어 있습니다. 이를 활성화하려면 쿠버네티스(Kubernetes) 로그 수집을 참고하세요.