MWAA Airflow task_instance_mutation_hook未生效，求排查原因

以下是针对该问题的可能原因及排查步骤：

• 钩子函数定义不符合规范
  必须严格按照Airflow要求的格式定义钩子，检查是否存在以下问题：

  from airflow.models import TaskInstance
  
  def task_instance_mutation_hook(task_instance: TaskInstance) -> None:
      # 你的逻辑代码
      pass
  

  • 函数名拼写错误（比如写成复数形式task_instance_mutation_hooks）
  • 参数类型未正确声明或缺少必要参数
  • 返回值不符合要求（必须返回None）

• MWAA环境配置限制
  检查MWAA环境的「Airflow Configuration」中是否存在core.enable_task_instance_mutation_hook配置项，若该值设置为False会直接禁用钩子，需调整为True后生效。

• 日志级别过滤导致日志不可见
  [TASK_INSTANCE_HOOK] Invoked for TaskInstance属于DEBUG级别的日志，MWAA默认日志级别通常为INFO或更高，会过滤该日志。可以将core.logging_level配置项设置为DEBUG，重新触发任务后查看日志。

• 钩子函数内部存在语法或依赖错误
  即使dag_policy能正常加载，钩子函数内部的语法错误（如缩进错误）、未导入必要模块（比如TaskInstance或日志模块）可能导致钩子无法被正常注册执行。可以在钩子开头添加简单的日志输出验证：

  import logging
  
  def task_instance_mutation_hook(task_instance: TaskInstance) -> None:
      logging.info("Task instance mutation hook triggered")
      # 你的角色切换逻辑
  

  然后查看MWAA的任务日志或调度器日志排查是否有报错信息。

• 执行角色权限不足（针对角色切换操作）
  如果角色切换需要调用sts:AssumeRole等IAM操作，而MWAA执行角色未被授予对应权限，会导致切换操作失败。但这种情况通常会在日志中出现权限报错，若连钩子调用日志都没有，优先排查前面的原因。

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