Airflow

FileTaskHandler와 원격 로깅

Airflow 코어는 FileTaskHandler라는 인터페이스를 제공한다. 이 핸들러는 태스크 로그를 파일에 쓰고, 태스크가 실행 중일 때 워커에서 그 로그를 서빙하는 메커니즘까지 포함한다. 커뮤니티는 다양한 서비스용 프로바이더를 통해 이 로깅 기능을 확장하는 핸들러를 추가로 제공한다.

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`` 디렉터리다. 태스크 로그 파일 이름은 일반 태스크의 경우 ``dag_id={dag_id}/run_id={run_id}/task_id={task_id}/attempt={try_number}.log`` 패턴을, 동적 매핑된 태스크의 경우 여기에 ``map_index={map_index}``가 추가된 패턴을 기본으로 따르며, 이 패턴은 ``logging__log_filename_template`` 설정으로 조정할 수 있다.

코드에서 태스크 로그에 쓰기

Airflow는 표준 파이썬 logging 프레임워크를 사용하며, 태스크가 실행되는 동안에는 루트 로거가 그 태스크의 로그로 쓰도록 재설정된다. 대부분의 오퍼레이터는 BaseOperator로부터 파생되므로 자동으로 구성되는 ``log`` 속성을 통해 자동으로 태스크 로그를 남긴다.

태스크 실행 중 루트 로거가 재설정되어 있기 때문에, 기본 설정으로 루트 로거에 전파되는 표준 파이썬 로거라면 무엇이든 태스크 로그에 함께 기록된다. 커스텀 코드에서 태스크 로그에 남기고 싶다면 세 가지 방법이 있다: BaseOperator의 ``self.log`` 로거 사용, ``print`` 문 사용(권장하지 않지만 경우에 따라 유용), 또는 파이썬 모듈 이름으로 로거를 생성해 쓰는 표준적인 방식이다.

로그 그룹핑 (2.9.0 이상)

CI 파이프라인처럼 Airflow 로그도 방대해져서 읽기 어려워질 수 있다. 이 문제를 완화하기 위해 Airflow는 GitHub Actions, Azure DevOps와 호환되는 로그 메시지 그룹핑 스킴을 구현했다. ``print("::group::제목")``으로 시작하고 ``print("::endgroup::")``으로 끝나는 구간을 만들면, 웹 UI에서는 해당 구간이 접힌 형태(예: 화살표 아이콘)로 표시되고 로그 텍스트 레이블을 클릭하면 상세 로그 라인이 펼쳐진다.

로그 인터리빙: 블롭 스토리지 vs 스트리밍 핸들러

Airflow의 원격 태스크 로깅 핸들러는 크게 스트리밍 핸들러(ElasticSearch, AWS CloudWatch, GCP operations logging 등)와 블롭 스토리지 핸들러(S3, GCS, WASB 등)로 나뉜다.

블롭 스토리지 핸들러는 태스크 상태에 따라 로그가 여러 곳, 여러 파일에 나뉘어 저장될 수 있어 모든 위치를 확인하고 찾아낸 내용을 인터리빙(시간순 병합)해야 한다. 이를 위해 각 로그 라인의 타임스탬프를 파싱해야 하며, 커스텀 포매터를 쓴다면 ``[logging] interleave_timestamp_parser`` 설정으로 기본 파서를 오버라이드해야 할 수 있다.

반면 스트리밍 핸들러는 태스크 실행 단계나 위치에 관계없이 모든 로그 메시지를 동일한 식별자로 로깅 서비스에 보낼 수 있으므로, 일반적으로 여러 소스를 확인하고 인터리빙할 필요가 없다.

워커와 트리거러에서 로그 실시간 서빙

대부분의 태스크 핸들러는 태스크가 완료된 후에 로그를 전송한다. 실행 중에 실시간으로 로그를 보기 위해 Airflow는 다음 경우에 HTTP 서버를 띄워 로그를 서빙한다: LocalExecutor를 쓴다면 ``airflow scheduler``가 실행 중일 때, CeleryExecutor를 쓴다면 ``airflow worker``가 실행 중일 때. 트리거러에서는 ``--skip-serve-logs`` 옵션으로 시작하지 않는 한 로그가 서빙된다.

서버는 ``[logging]`` 섹션의 ``worker_log_server_port``(기본 8793)와 트리거러용 ``trigger_log_server_port``(기본 8794) 포트에서 동작한다. 웹서버와 워커 간 통신은 ``[api]`` 섹션의 ``secret_key`` 옵션으로 서명되므로, 문제없이 통신하려면 이 키가 서로 일치해야 한다. 서버는 WSGI 서버로 Gunicorn을 사용하며, ``GUNICORN_CMD_ARGS`` 환경변수로 설정을 오버라이드할 수 있다.

커스텀 FileTaskHandler 구현 시 트리거 관련 속성

대부분의 사용자는 프로바이더가 제공하는 기존 핸들러를 쓰면 되지만, 필요하다면 커스텀 FileTaskHandler를 구현할 수 있다. 이때 트리거러 로깅과 관련해 알아야 할 속성들이 있다. 트리거는 많은 트리거가 같은 프로세스에서 asyncio로 동작하기 때문에 로깅 핸들러를 통해 블로킹 호출이 들어가지 않도록 주의해야 하는 특수한 맥락이다.

인스턴스나 클래스에 설정할 수 있는 속성은 다음과 같다(상속은 존중되지 않으므로 서브클래스마다 명시가 필요할 수 있다): ``trigger_should_wrap``(TriggerHandlerWrapper로 감쌀지 여부. 핸들러 인스턴스마다 파일 핸들러를 만들어 모든 메시지를 쓰는 경우 필요), ``trigger_should_queue``(트리거러가 이벤트 루프와 핸들러 사이에 QueueListener를 둘지 여부. 핸들러의 블로킹 IO가 이벤트 루프를 방해하지 않도록 함), ``trigger_send_end_marker``(트리거 완료 시 END 신호를 로거에 보낼지 여부. 래퍼가 그 트리거 전용 파일 핸들러를 닫고 제거하도록 지시), ``trigger_supported``(위 두 속성이 True가 아니면 보통 트리거를 지원하지 않는다고 간주하지만, 이 속성이 True면 트리거러 시작 시 핸들러를 루트로 옮겨 트리거 메시지를 처리하게 함. StackdriverTaskHandler가 이런 네이티브 지원 사례다).