Task 로깅

Task 로깅 (Logging for Tasks)

이 페이지는 Airflow UI에서 각 Task의 로그를 따로 볼 수 있게 해주는 Task 로깅 방식을 설명해요. 기본 로깅 설정(airflow.cfg의 base_log_folder)과 로그 파일 이름 패턴, 커스텀 코드에서 Task 로그에 쓰는 방법, 로그 줄 그룹화, 워커·트리거러에서 로그 제공, 커스텀 파일 Task 핸들러 구현까지 다뤄요.

출처: 문서

본문

Airflow는 Airflow UI에서 각 Task의 로그를 따로 볼 수 있도록 Task 로그를 기록해요. Core Airflow는 FileTaskHandler 인터페이스를 제공하는데, 이 인터페이스는 Task 로그를 파일에 기록하고 Task가 실행되는 동안 worker에서 그 로그를 제공하는 매커니즘을 포함해요. Apache Airflow 커뮤니티는 많은 서비스용 provider도 배포하며(Providers), 그중 일부는 Apache Airflow의 로깅 기능을 확장하는 핸들러를 제공해요. 모든 provider는 Writing logs에서 볼 수 있어요.

S3, GCS, WASB, HDFS 또는 OSS 원격 로깅 서비스를 사용할 때, 로컬 로그 파일이 원격 위치에 업로드된 후 삭제하도록 설정할 수 있어요:

[logging]
remote_logging = True
remote_base_log_folder = schema://path/to/remote/log
delete_local_logs = True

로깅 구성

기본 핸들러인 FileTaskHandler의 경우 airflow.cfg에서 base_log_folder를 사용해 로그 파일을 둘 디렉터리를 지정할 수 있어요. 기본적으로 로그는 AIRFLOW_HOME 디렉터리에 배치돼요.

Note

설정 방법에 대한 자세한 내용은 설정 옵션 설정하기를 참고해요.

Task 로그 파일 이름을 지을 때는 기본 패턴이 따릅니다:

  • 일반 Task: dag_id={dag_id}/run_id={run_id}/task_id={task_id}/attempt={try_number}.log
  • 동적 매핑된 Task: dag_id={dag_id}/run_id={run_id}/task_id={task_id}/map_index={map_index}/attempt={try_number}.log

이 패턴은 log_filename_template으로 조정할 수 있어요.

또한 현재 로그와 백업을 저장할 원격 위치를 제공할 수 있어요.

코드에서 Task 로그에 쓰기

Airflow는 로그를 작성하기 위해 표준 Python logging 프레임워크를 사용하며, Task가 실행되는 동안 root logger는 Task의 로그에 쓰도록 구성돼요.

대부분의 Operator는 Task 로그에 자동으로 로그를 작성해요. 이는 Task 로그에 쓰는 데 사용할 수 있는 (타입이 Logger인) log 속성이 있기 때문이에요. 이 logger는 BaseOperator에서 파생된 모든 Operator에 자동으로 구성돼요.

또한 Task 실행 중 root logger 구성 덕분에, root logger로 전파되는 (기본 설정을 사용하는) 표준 Python logger도 Task 로그에 쓰게 돼요.

그래서 나만의 커스텀 코드에서 Task 로그에 로깅하고 싶다면 다음 중 어느 것이든 할 수 있어요:

  • BaseOperator의 self.log logger로 로깅
  • 표준 print 문으로 stdout에 출력 (권장되진 않지만, 어떤 경우엔 유용할 수 있음)
  • Python 모듈 이름으로 logger를 만드는 표준 logger 방식을 사용해 Task 로그에 쓰기

이것은 Python 코드에서 logger를 직접 사용하는 일반적인 방식이에요:

import logging

logger = logging.getLogger(__name__)
logger.info("This is a log message")

로그 줄 그룹화

버전 2.9.0에 추가됨.

CI 파이프라인처럼 Airflow 로그도 꽤 커져서 읽기 어려워질 수 있어요. 때로는 로그 영역의 섹션을 그룹화하고 텍스트 영역을 접어(폴딩) 관련 없는 내용을 숨기는 것이 유용해요. 그래서 Airflow는 GitHub와 Azure DevOps와 호환되는 로그 메시지 그룹화를 구현해서 텍스트 영역을 접을 수 있게 해줘요. 구현된 방식은 호환성이 있어서 CI에서 출력을 만드는 도구가 Airflow에서도 동일한 경험을 활용할 수 있어요.

시작과 끝 위치를 나타내는 로그 마커를 추가하면 아래 예제처럼 로그 메시지를 그룹화할 수 있어요:

print("Here is some standard text.")
print("::group::Non important details")
print("bla")
print("debug messages...")
print("::endgroup::")
print("Here is again some standard text.")

웹 UI에서 로그를 표시할 때 로그 표시가 압축돼요:

[2024-03-08, 23:30:18 CET] {logging_mixin.py:188} INFO - Here is some standard text.
[2024-03-08, 23:30:18 CET] {logging_mixin.py:188} ⯈ Non important details
[2024-03-08, 23:30:18 CET] {logging_mixin.py:188} INFO - Here is again some standard text.

로그 텍스트 라벨을 클릭하면 상세 로그 줄이 표시돼요.

[2024-03-08, 23:30:18 CET] {logging_mixin.py:188} INFO - Here is some standard text.
[2024-03-08, 23:30:18 CET] {logging_mixin.py:188} ⯆ Non important details
[2024-03-08, 23:30:18 CET] {logging_mixin.py:188} INFO - bla
[2024-03-08, 23:30:18 CET] {logging_mixin.py:188} INFO - debug messages...
[2024-03-08, 23:30:18 CET] {logging_mixin.py:188} ⯅⯅⯅ Log group end
[2024-03-08, 23:30:18 CET] {logging_mixin.py:188} INFO - Here is again some standard text.

로그 인터리빙(Interleaving)

Airflow의 원격 Task 로깅 핸들러는 크게 두 범주로 나눌 수 있어요: 스트리밍 핸들러(ElasticSearch, AWS Cloudwatch, GCP operations logging(구 stackdriver) 등)와 blob 저장소 핸들러(예: S3, GCS, WASB)예요.

blob 저장소 핸들러의 경우 Task의 상태에 따라 로그가 아주 여러 다른 위치와 여러 다른 파일에 있을 수 있어요. 이런 이유로 모든 위치를 확인하고 발견한 것을 인터리빙해야 해요. 그러려면 각 줄의 타임스탬프를 파싱할 수 있어야 해요. 커스텀 포매터를 사용한다면 Airflow 설정 [logging] interleave_timestamp_parser에 callable 이름을 제공해 기본 파서를 덮어써야 할 수 있어요.

스트리밍 핸들러의 경우 Task 단계나 실행 위치와 무관하게 모든 로그 메시지를 같은 식별자로 로깅 서비스에 보낼 수 있어서, 일반적으로 여러 소스를 확인해 인터리빙할 필요가 없어요.

문제 해결

현재 어떤 Task 핸들러가 설정되어 있는지 확인하려면 아래 예제처럼 airflow info 명령을 사용할 수 있어요.

$ airflow info

Apache Airflow
version                | 2.9.0.dev0
executor               | LocalExecutor
task_logging_handler   | airflow.utils.log.file_task_handler.FileTaskHandler
sql_alchemy_conn       | postgresql+psycopg2://postgres:airflow@postgres/airflow
dags_folder            | /files/dags
plugins_folder         | /root/airflow/plugins
base_log_folder        | /root/airflow/logs
remote_base_log_folder |

[skipping the remaining outputs for brevity]

위 airflow info 출력은 로깅 설정과 관련된 섹션만 표시하도록 줄여졌어요. 또한 airflow config list를 실행해 로깅 설정 옵션이 유효한 값을 갖는지 확인할 수도 있어요.

고급 설정

고급 기능을 구성할 수 있어요 — 나만의 커스텀 Task 로그 핸들러(뿐 아니라 모든 Airflow 컴포넌트용 로그 핸들러)를 추가하고, Operator·hook·task별로 커스텀 로그 핸들러를 만드는 것까지 포함해요.

worker와 triggerer에서 로그 제공하기

대부분의 Task 핸들러는 Task가 완료될 때 로그를 전송해요. 실시간으로 로그를 보기 위해 Airflow는 다음 경우에 로그를 제공하는 HTTP 서버를 시작해요:

  • LocalExecutor를 사용하는 경우, airflow scheduler가 실행 중일 때.
  • CeleryExecutor를 사용하는 경우, airflow worker가 실행 중일 때.

triggerer에서는 --skip-serve-logs 옵션으로 서비스를 시작하지 않는 한 로그가 제공돼요.

서버는 [logging] 섹션의 worker_log_server_port 옵션으로 지정된 포트에서 실행되고, triggerer의 경우 trigger_log_server_port 옵션을 사용해요. 기본값은 각각 8793과 8794예요. webserver와 worker 사이의 통신은 [api] 섹션의 secret_key 옵션으로 지정된 키로 서명돼요. 키가 일치해야 문제 없이 통신할 수 있다는 점을 확인해야 해요.

WSGI 서버로 Gunicorn을 사용해요. 그 설정 옵션은 GUNICORN_CMD_ARGS 환경 변수로 덮어쓸 수 있어요. 자세한 내용은 Gunicorn settings를 참고해요.

커스텀 파일 Task 핸들러 구현하기

Note

이것은 고급 주제이며, 대부분의 사용자는 Writing logs의 기존 핸들러를 그냥 사용하면 돼요.

우리 provider에는 주요 클라우드 공급자와 함께 충분히 다양한 옵션들이 있어요. 그러나 다른 서비스로 로깅을 구현해야 하고, 그다음 커스텀 FileTaskHandler를 구현하기로 결정했다면 알아둬야 할 설정이 몇 가지 있어요, 특히 trigger 로깅과 관련해서요.

Trigger는 로깅이 설정되는 방식을 바꿔야 해요. Task와 달리 많은 trigger가 같은 프로세스에서 실행되고, trigger는 asyncio에서 실행되므로 로깅 핸들러를 통해 블로킹 호출을 도입하지 않도록 주의해야 해요. 그리고 핸들러 동작의 다양성(어떤 것은 파일에 쓰고, 어떤 것은 blob 저장소에 업로드하고, 어떤 것은 도착하자마자 네트워크로 메시지를 보내고, 어떤 것은 스레드에서 그러함) 때문에 triggerer가 이를 어떻게 사용해야 하는지 알려줄 방법이 필요해요.

이를 위해 핸들러(인스턴스 또는 클래스)에 설정할 수 있는 몇 가지 속성이 있어요. FileTaskHandler의 서브클래스는 관련 특성에서 다를 수 있으므로, 이 파라미터에는 상속이 적용되지 않아요. 이 파라미터들은 아래에 설명돼요:

  • trigger_should_wrap: 이 핸들러를 TriggerHandlerWrapper로 감싸야 하는지 제어해요. 핸들러의 각 인스턴스가 모든 메시지를 쓰는 파일 핸들러를 만들 때 필요해요.
  • trigger_should_queue: triggerer가 이벤트 루프와 핸들러 사이에 QueueListener를 둬서 핸들러의 블로킹 I/O가 이벤트 루프를 방해하지 않도록 해야 하는지 제어해요.
  • trigger_send_end_marker: trigger가 완료될 때 logger에 END 신호를 보내야 하는지 제어해요. 방금 완료된 trigger에 특화된 개별 파일 핸들러를 wrapper가 닫고 제거하도록 알려주는 데 사용돼요.
  • trigger_supported: trigger_should_wrap과 trigger_should_queue가 True가 아니면 일반적으로 그 핸들러가 trigger를 지원하지 않는다고 가정해요. 하지만 이 경우 핸들러의 trigger_supported가 True로 설정되어 있으면, triggerer 시작 시 그 핸들러를 root로 이동시켜 trigger 메시지를 처리하게 해요. 기본적으로 이는 trigger를 "네이티브"로 지원하는 핸들러에 대해 True여야 해요. 그런 예로 StackdriverTaskHandler가 있어요.

외부 링크

원격 로깅을 사용할 때 Airflow Web UI 안에 외부 UI로 연결하는 링크를 표시하도록 Airflow를 구성할 수 있어요. 링크를 클릭하면 외부 UI로 이동해요.

일부 외부 시스템은 리다이렉션이 동작하려면 Airflow에서 특정 구성이 필요하지만, 다른 시스템은 그럴 필요가 없어요.

더 알아보기 (Learn more)