[Airflow] Telegraf/InfluxDB로 구현하는 metric 수집기

어쩌다 StatsD exporter -> Prometheus 루트가 아니라 Telegraf -> InfluxDB 루트를 타게된 썰😅

• 보통 Airflow metric 적재 시에는 statsd_exporter -> Prometheus -> Grafana 아키텍처를 사용합니다만 모종의 이유로 InfluxDB를 사용해야했고, Telegraf -> InfulxDB 아키텍처로 metric을 수집한 기록을 정리하고자 합니다.
• Airflow와 Telegraf는 helm을 이용해서 구축했습니다.

StatsD

• StatsD는 애플리케이션에서 발생하는 성능 지표를 수집하고, 이를 모니터링 시스템에 전송하는 데 사용되는 프로토콜입니다.
• airflow에서 생성된 메트릭은 StatsD를 통해서 전송됩니다.
• StatsD를 사용하기 위해서 helm airflow에서 아래와 같이 설정해줍니다.

values.yaml

statsd:
  enabled: true
  ...
  
config:
  metrics:
    statsd_on: '{{ ternary "True" "False" .Values.statsd.enabled }}'
    statsd_port: 9125           # StatsD 서버와 통신할 포트
    statsd_prefix: 'airflow'    # StatsD로 전송되는 메트릭의 접두사
    statsd_host: 'telegraf'     # StatsD 데이터를 수신할 서버의 호스트 이름

airflow metric

• 위와 같이 설정하고 나면 tcpdump -i any udp port 9125 -A 등의 네트워크 조회 명령어를 통해 실제로 어떤 메트릭을 내보내고 확인할 수 있습니다.
• 제공하는 전체 메트릭은 해당 문서에서도 자세히 제공하고 있습니다.

Telegraf

• Telegraf는 여러 input plugin을 지원하고 있고, 여기서는 Airflow와 k8s 클러스터의 메트릭을 모두 수집하기 위해 stasd, kubernetes plugins을 사용합니다.
• 메트릭 저장소는 influxDB를 사용할 것이기 때문에 outputs은 influxDB로 지정해줍니다.
• StatsD exporter를 사용할 경우, statsd-exporter mapping 파일을 통해 메트릭 변환 방식을 정의하지만 여기서는 StatsD exporter를 사용하지 않고 Telegraf를 사용하기 때문에 Telegraf config에서 템플릿을 지정합니다.
• kubernetes input plugin은 Kubelet API를 통해 메트릭을 수신하는 메커니즘입니다.
  (kubernetes 메트릭)
  • 단일 호스트에 대한 수집을 지원함으로 클러스터 내, 모든 노드에 배포가 필요합니다.

values.yaml

config:
  ...
  outputs:
    - influxdb_v2:
        urls:
          - "${urls}"
        organization: "${organization}"
        bucket: "${bucket}"
        token: "${token}"
   inputs:
    - statsd:
        service_address: ":9125"
        metric_separator: "_"
        data_format: "influx"
        templates:
          - "*.dag.*.*.duration measurement.measurement.dag_id.task_id.measurement"
          - "*.dagrun.dependency-check.* measurement.measurement.measurement.dag_id"
          - "*.dag_processing.last_duration.* measurement.measurement.measurement.dag_id"
          - "*.dagrun.duration.*.* measurement.measurement.measurement.status.dag_id"
          - "*.dagrun.schedule_delay.* measurement.measurement.measurement.dag_id"
          - "*.dag_processing.last_runtime.* measurement.measurement.measurement.dag_id"
          - "*.dag.loading-duration.* measurement.measurement.measurement.dag_id"
          - "*.operator.*.* measurement.measurement.status.operator"
    - kubernetes:
        url: "https://$K8S_NODE_IP:10250"
        bearer_token: "/var/run/secrets/kubernetes.io/serviceaccount/token"
        tls_ca: "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"
        insecure_skip_verify: true

InfluxDB

• 위와 같이 설정을 하게되면 InfluxDB에서 Airflow와 k8s 두 메트릭을 모두 확인할 수 있고, Grafana에서 해당 메트릭을 시각화할 수 있습니다.