Airflow挂载Kafka Handler后任务运行日志缺失的配置问询

你遇到的核心问题是：Airflow的任务执行相关日志（如taskinstance.py、standard_task_runner.py）运行在独立的Worker进程中，且部分专属日志器默认未继承根日志器的Handler配置，仅给根日志器添加Kafka Handler无法覆盖这些场景。以下是具体修复步骤：

1. 为任务相关日志器显式绑定Kafka Handler #

Airflow针对任务、CLI、任务Runner有单独的日志器配置，需要在custom_log_config.py中为这些日志器添加Kafka Handler：

LOGGING_CONFIG = deepcopy(DEFAULT_LOGGING_CONFIG)

# Configure a new handler for publishing logs to kafka
environment = get_app_env()
LOGGING_CONFIG["handlers"]["kafka_handler"] = {
    "class": "com.test.log_handler.KafkaHandler",
    "formatter": "airflow",
    "version": environment.version,
    "log_file": log_file,
    "filters": ["mask_secrets"],
}

# Attach handler to root logger
LOGGING_CONFIG["root"]["handlers"].append("kafka_handler")

# 为任务日志器添加Kafka Handler（保留原有handler，避免丢失本地日志）
LOGGING_CONFIG["loggers"]["airflow.task"] = {
    "handlers": ["kafka_handler", "task"],
    "level": "INFO",
    "propagate": False,
}

# 为CLI日志器添加Kafka Handler
LOGGING_CONFIG["loggers"]["airflow.cli"] = {
    "handlers": ["kafka_handler", "cli"],
    "level": "INFO",
    "propagate": False,
}

# 为任务Runner日志器添加Kafka Handler
LOGGING_CONFIG["loggers"]["airflow.task_runner"] = {
    "handlers": ["kafka_handler"],
    "level": "INFO",
    "propagate": False,
}

2. 确保Worker进程加载自定义日志配置 #

Worker是独立的执行进程，需确认其启动时也使用你的自定义日志配置：

• 在Airflow全局配置（airflow.cfg）或环境变量中设置：

  airflow.logging__logging_config_class=com.test.log_handler.custom_log_config.LOGGING_CONFIG
  

• 如果用systemd等工具管理Worker进程，需检查启动脚本是否传递了该配置参数，避免Worker加载默认日志配置。

3. 验证日志级别一致性 #

• 确认根日志器级别（LOGGING_CONFIG["root"]["level"]）不高于任务日志器的级别（如airflow.task的INFO），防止低级别日志被过滤。
• 检查KafkaHandler自身是否有级别限制，确保能接收对应级别的日志输出。

4. 排查Handler初始化异常 #

• 查看Worker启动日志，确认KafkaHandler能正常初始化，无模块导入错误、Kafka连接失败等异常。
• 确保Worker的Python环境包含com.test.log_handler的依赖（如Kafka客户端库）。

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