使用CloudWatch远程日志时Airflow中Django任务的DAG卡住问题

我通过自定义Operator在Apache Airflow中运行Django任务，默认日志配置（AIRFLOW__LOGGING__REMOTE_LOGGING=False）下一切正常，但启用CloudWatch远程日志后，DAG在执行django.setup()调用Django日志配置时卡住。已确认CloudWatch日志功能本身可用，任务因长时间无进展被标记失败，想问是否需要特定的Django日志配置来适配这种场景。

相关日志信息 #

*** 从CloudWatch日志组读取远程日志: job-sentinel-image-layer-log-group-5f303da 日志流: dag_id=clear_tokens/run_id=scheduled__2023-07-03T22_00_00+00_00/task_id=execute_clear_tokens_command/attempt=1.log.
[2023-07-05, 17:23:27 CEST] 所有依赖项已满足，dep_context=non-requeueable deps ti=<TaskInstance: clear_tokens.execute_clear_tokens_command scheduled__2023-07-03T22:00:00+00:00 [queued]>
[2023-07-05, 17:23:27 CEST] 所有依赖项已满足，dep_context=requeueable deps ti=<TaskInstance: clear_tokens.execute_clear_tokens_command scheduled__2023-07-03T22:00:00+00:00 [queued]>
[2023-07-05, 17:23:27 CEST] 
--------------------------------------------------------------------------------
[2023-07-05, 17:23:27 CEST] 开始第1次尝试，共2次
[2023-07-05, 17:23:27 CEST] 
--------------------------------------------------------------------------------
[2023-07-05, 17:23:27 CEST] 在2023-07-03 22:00:00+00:00执行<Task(DjangoOperator): execute_clear_tokens_command>
[2023-07-05, 17:23:27 CEST] 启动进程24150以运行任务
[2023-07-05, 17:23:27 CEST] 运行命令: ['airflow', 'tasks', 'run', 'clear_tokens', 'execute_clear_tokens_command', 'scheduled__2023-07-03T22:00:00+00:00', '--job-id', '3', '--raw', '--subdir', 'DAGS_FOLDER/management_commands/src/clear_tokens.py', '--cfg-path', '/tmp/tmpkmehxuz0']
[2023-07-05, 17:23:27 CEST] 任务3: 子任务 execute_clear_tokens_command
[2023-07-05, 17:23:28 CEST] 在主机147b23cea447上运行<TaskInstance: clear_tokens.execute_clear_tokens_command scheduled__2023-07-03T22:00:00+00:00 [running]>
[2023-07-05, 17:23:28 CEST] 导出以下环境变量:
AIRFLOW_CTX_DAG_ID=clear_tokens
AIRFLOW_CTX_TASK_ID=execute_clear_tokens_command
AIRFLOW_CTX_EXECUTION_DATE=2023-07-03T22:00:00+00:00
AIRFLOW_CTX_TRY_NUMBER=1
AIRFLOW_CTX_DAG_RUN_ID=scheduled__2023-07-03T22:00:00+00:00
[2023-07-05, 17:23:28 CEST] 进入预执行阶段
[2023-07-05, 17:23:28 CEST] 在django.setup()中完成`from django.utils.log import configure_logging`导入
[2023-07-05, 17:30:06 CEST] 该实例状态已被外部设置为失败。正在终止实例。
[2023-07-05, 17:30:06 CEST] 向进程组24150发送Signals.SIGTERM信号。进程组中的所有PID: [24150]
[2023-07-05, 17:30:06 CEST] 向进程组24150发送Signals.SIGTERM信号
[2023-07-05, 17:30:06 CEST] 收到SIGTERM信号。正在终止子进程。
[2023-07-05, 17:30:06 CEST] 进程psutil.Process(pid=24150, status='terminated', exitcode=0, started='15:23:27') (24150) 已终止，退出码0

解决方案 #

• 禁用Django自动日志配置：在调用django.setup()前，覆盖Django的日志配置函数，避免其与Airflow的CloudWatch日志系统冲突。在自定义Operator代码中添加：

  import os
  os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目.settings')
  
  # 跳过Django默认日志配置
  from django.utils.log import configure_logging
  configure_logging = lambda: None
  
  import django
  django.setup()
  

• 简化Django日志配置：修改Django项目的settings.py，仅保留基础日志处理器，避免触发阻塞逻辑：

  import os
  LOGGING = {
      'version': 1,
      'disable_existing_loggers': False,
      'handlers': {
          'console': {
              'class': 'logging.StreamHandler',
          },
      },
      'root': {
          'handlers': ['console'],
          'level': 'INFO',
      },
      'loggers': {
          'django': {
              'handlers': ['console'],
              'level': os.getenv('DJANGO_LOG_LEVEL', 'INFO'),
              'propagate': False,
          },
      },
  }
  

• 复用Airflow日志上下文：自定义Operator继承Airflow的LoggingMixin，直接使用Airflow的日志器，无需让Django重新初始化日志系统：

  from airflow.models.baseoperator import BaseOperator
  from airflow.utils.log.logging_mixin import LoggingMixin
  
  class DjangoOperator(BaseOperator, LoggingMixin):
      def execute(self, context):
          self.log.info("启动Django任务")
          # 此处编写Django任务逻辑
  

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