Cloud Composer 3中KubernetesPodOperator异步模式获取日志遇权限禁止问题咨询

环境信息 #

• Cloud Composer 3（composer-3-airflow-2.10.2-build.13）
• 区域：europe-west1
• 环境名称：composer3-npd
• 项目：XXXXXXXXXX
• 私有环境：已启用

问题概述 #

使用设置了deferrable=True和get_logs=True的KubernetesPodOperator时，任务能正常延迟和恢复，但Airflow UI里始终看不到Pod日志，且日志中反复出现警告：

WARNING - Reading of logs interrupted with error 'Forbidden'; will retry.

已验证的行为 #

• deferrable=True + get_logs=False：无权限错误，运行正常
• deferrable=False + get_logs=True：Composer 2和3中都能正常查看日志
• deferrable=True + get_logs=True：仅在Composer 3中出现权限禁止错误

问题解答 #

这是Cloud Composer 3的已知限制吗？ #

是的，这属于Composer 3的权限配置问题。当开启deferrable=True时，Pod日志的读取请求是由Airflow的触发器组件发起的，而Composer 3默认给触发器服务账号的权限不足以访问composer-user-workloads命名空间下的Pod日志。

可行的解决方法 #

方法1：给触发器服务账号添加日志读取权限

1. 找到你的Composer环境触发器服务账号：在Composer环境详情页的「基础设施」标签里查看，格式一般是composer-{环境ID}-trigger@{项目ID}.iam.gserviceaccount.com
2. 给这个账号授予日志查看权限，两种方式任选：
   • 全局权限（简单直接）：

     gcloud projects add-iam-policy-binding {你的项目ID} \
       --member="serviceAccount:{触发器服务账号}" \
       --role="roles/container.logViewer"
     

   • 精细权限（仅针对目标命名空间）：
     先创建集群角色和绑定，限制仅能读取composer-user-workloads下的Pod日志：

     kubectl create clusterrole pod-log-reader --verb=get,list --resource=pods/logs
     kubectl create clusterrolebinding trigger-pod-log-reader \
       --clusterrole=pod-log-reader \
       --serviceaccount=composer-user-workloads:{触发器服务账号的前缀}
     

     注：K8s服务账号名称是IAM账号去掉@{项目ID}.iam.gserviceaccount.com的部分，比如composer-123-trigger@my-project.iam.gserviceaccount.com对应的K8s SA是composer-123-trigger

方法2：改用Cloud Logging查看日志

如果不想修改权限，可以配置Pod日志自动上传到Cloud Logging，再通过控制台查看：

1. 确保你的Pod使用gcp-container日志驱动（Composer默认配置一般已经开启）
2. 在Airflow任务的描述或备注里添加Cloud Logging的直接链接，格式如下：

   https://console.cloud.google.com/logs/viewer?project={你的项目ID}&resource=k8s_container%2Fcluster%2F{Composer集群ID}%2Fnamespace%2Fcomposer-user-workloads%2Fpod_name%2F{Pod名称}
   

3. 也可以用PythonOperator写个简单逻辑，根据Pod名称查询Cloud Logging并把关键日志输出到Airflow任务日志里

验证代码 #

权限配置完成后，原测试代码即可正常工作（可以把cmds和arguments改成实际输出内容方便验证）：

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from datetime import datetime

with DAG(
    dag_id="kpo_deferrable_logs_issue",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
) as dag:

    test_task = KubernetesPodOperator(
        task_id="test_kpo",
        name="test-kpo",
        namespace="composer-user-workloads",
        image="my_images",
        cmds=["echo"],
        arguments=["测试延迟模式下的日志输出"],
        get_logs=True,
        is_delete_operator_pod=True,
        deferrable=True,
    )

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