Airflow BashOperator返回码0却标记任务失败的排查与解决

问题背景 #

使用Apache Airflow通过BashOperator调用Python脚本，此前运行正常，近期出现异常：任务日志明确显示命令执行完成且返回码为0，但Airflow仍将任务标记为失败。相关日志如下：

*** Reading local file: /opt/airflow/logs/dag_id=derin_emto_preprocess/run_id=manual__2022-10-01T13:54:50.246801+00:00/task_id=emto_preprocess-month0day0/attempt=1.log
[2022-10-01, 13:55:21 UTC] {taskinstance.py:1159} INFO - Dependencies all met for <TaskInstance: derin_emto_preprocess.emto_preprocess-month0day0 manual__2022-10-01T13:54:50.246801+00:00 [queued]>
[2022-10-01, 13:55:21 UTC] {taskinstance.py:1159} INFO - Dependencies all met for <TaskInstance: derin_emto_preprocess.emto_preprocess-month0day0 manual__2022-10-01T13:54:50.246801+00:00 [queued]>
[2022-10-01, 13:55:21 UTC] {taskinstance.py:1356} INFO -
--------------------------------------------------------------------------------
[2022-10-01, 13:55:21 UTC] {taskinstance.py:1357} INFO - Starting attempt 1 of 1
[2022-10-01, 13:55:21 UTC] {taskinstance.py:1358} INFO -
--------------------------------------------------------------------------------
[2022-10-01, 13:55:21 UTC] {taskinstance.py:1377} INFO - Executing <Task(BashOperator): emto_preprocess-month0day0> on 2022-10-01 13:54:50.246801+00:00
[2022-10-01, 13:55:21 UTC] {standard_task_runner.py:52} INFO - Started process 624 to run task
[2022-10-01, 13:55:21 UTC] {standard_task_runner.py:79} INFO - Running: ['***', 'tasks', 'run', 'derin_emto_preprocess', 'emto_preprocess-month0day0', 'manual__2022-10-01T13:54:50.246801+00:00', '--job-id', '8958', '--raw', '--subdir', 'DAGS_FOLDER/derin_emto_preprocess.py', '--cfg-path', '/tmp/tmpjn_8tmiv', '--error-file', '/tmp/tmp_jr_2w3j']
[2022-10-01, 13:55:21 UTC] {standard_task_runner.py:80} INFO - Job 8958: Subtask emto_preprocess-month0day0
[2022-10-01, 13:55:21 UTC] {task_command.py:369} INFO - Running <TaskInstance: derin_emto_preprocess.emto_preprocess-month0day0 manual__2022-10-01T13:54:50.246801+00:00 [running]> on host 5b44f8453a08
[2022-10-01, 13:55:21 UTC] {taskinstance.py:1571} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=***
AIRFLOW_CTX_DAG_ID=derin_emto_preprocess
AIRFLOW_CTX_TASK_ID=emto_preprocess-month0day0
AIRFLOW_CTX_EXECUTION_DATE=2022-10-01T13:54:50.246801+00:00
AIRFLOW_CTX_TRY_NUMBER=1
AIRFLOW_CTX_DAG_RUN_ID=manual__2022-10-01T13:54:50.246801+00:00
[2022-10-01, 13:55:21 UTC] {subprocess.py:62} INFO - Tmp dir root location:
 /tmp
[2022-10-01, 13:55:21 UTC] {subprocess.py:74} INFO - Running command: ['bash', '-c', 'python /opt/***/dags/scripts/derin/pipeline/pipeline.py --valid_from=20200101 --valid_until=20200102 --purpose=emto_preprocess --module=emto_preprocess --***=True']
[2022-10-01, 13:55:21 UTC] {subprocess.py:85} INFO - Output:
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO - 2022-10-01 13:55:22 :  Hello, world!
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO - 2022-10-01 13:55:22 :  [20200101, 20200102)
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO - 2022-10-01 13:55:22 :  Running emto_preprocess purpose
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO - Current directory : /opt/***/dags
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO - 2022-10-01 13:55:22 :  Airflow parameter passed: changing configuration..
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO - 2022-10-01 13:55:24 :  Parallel threads: 15
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO - 2022-10-01 13:55:24 :  External money transfer out: preprocess is starting..
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO -
Thread None for emto_preprocess:   0%|          | 0/1 [00:00<?, ?it/s]
Thread None for emto_preprocess: 100%|██████████| 1/1 [00:00<00:00, 12633.45it/s]
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO - 2022-10-01 13:55:24 :  DEBUG: Checking existing files
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO - 2022-10-01 13:55:24 :  This module is already processed
[2022-10-01, 13:55:24 UTC] {subprocess.py:92} INFO - 2022-10-01 13:55:24 :  Good bye!
[2022-10-01, 13:55:24 UTC] {subprocess.py:96} INFO - Command exited with return code 0
[2022-10-01, 13:55:24 UTC] {taskinstance.py:1400} INFO - Marking task as SUCCESS. dag_id=derin_emto_preprocess, task_id=emto_preprocess-month0day0, execution_date=20221001T135450, start_date=20221001T135521, end_date=20221001T135524
[2022-10-01, 13:55:24 UTC] {local_task_job.py:156} INFO - Task exited with return code 0
[2022-10-01, 13:55:25 UTC] {local_task_job.py:273} INFO - 0 downstream tasks scheduled from follow-on schedule check

排查解决步骤 #

• 校验任务实际状态
  用Airflow CLI命令查询任务真实状态：

  airflow tasks state derin_emto_preprocess emto_preprocess-month0day0 "2022-10-01T13:54:50.246801+00:00"
  

  如果返回SUCCESS但UI显示失败，说明是Webserver缓存问题，重启Airflow Webserver即可同步状态。

• 检查元数据库连接
  元数据库（如PostgreSQL/MySQL）的连接异常、锁表或事务未提交会导致状态更新失败。重启Airflow Scheduler和Worker，同时查看数据库日志，排查是否有连接超时、死锁等问题。

• 处理输出中的特殊字符
  日志中出现的进度条包含终端控制字符，部分Airflow版本对这类字符处理存在bug，会干扰状态判断。可以：

  • 修改Python脚本，禁用进度条输出；
  • 在Bash命令中过滤特殊字符：

    python /opt/***/dags/scripts/derin/pipeline/pipeline.py [参数] 2>&1 | sed 's/\x1b\[[0-9;]*m//g'
    

• 检查Airflow版本与超时配置

  • 旧版本Airflow（1.x部分版本）存在BashOperator状态判断bug，建议升级到2.2+稳定版本；
  • 核查任务的execution_timeout设置，确认超时时间大于任务实际执行时长，避免误触发超时判定。

• 查看Scheduler与Worker日志
  任务状态由Scheduler负责更新，查看Scheduler和Worker的日志文件，检查是否存在元数据库连接失败、状态更新报错等异常信息。

内容的提问来源于stack exchange，提问作者ugurtosun