如何使用Python Kubernetes API将K8s Secrets挂载到Airflow Pod数据卷

你的挂载目录为空主要存在三个核心问题：

• 手动创建的空空白卷挂载到了/credentials路径，覆盖了Secret自动生成的卷内容
• Secret的deploy_target参数配置错误，指定单个key挂载为文件时，该参数需要填完整的目标文件路径而非目录
• Python代码中使用相对路径读取文件，和挂载的绝对路径不匹配

第一步：调整DAG代码 #

删掉手动创建的空白Volume、VolumeMount配置，修正Secret的deploy_target参数，修改后的代码如下：

from airflow import DAG
from datetime import datetime, timedelta
from airflow.contrib.operators.kubernetes_pod_operator import KubernetesPodOperator
from airflow.kubernetes.secret import Secret
from airflow.operators.dummy_operator import DummyOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime.utcnow(),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retry_delay': timedelta(minutes=5)
}

# 修正Secret配置，deploy_target填完整的文件绝对路径
secret_jira_user = Secret(deploy_type="volume",
                          deploy_target="/credentials/jira_credentials.json",
                          secret="jira-user-secret",
                          key="jira_credentials.json")
secret_storage_credentials = Secret(deploy_type="volume",
                                    deploy_target="/credentials/cloud_storage_credentials.json",
                                    secret="jira-trans-projects-cloud-storage-creds",
                                    key="cloud_storage_credentials.json")

dag = DAG(
    dag_id="jira_translations_project",
    schedule_interval="0 1 * * MON",
    start_date=datetime(2021, 9, 5, 0, 0, 0),
    max_active_runs=1,
    default_args=default_args
)

start = DummyOperator(task_id='START', dag=dag)

passing = KubernetesPodOperator(namespace='default',
                                image="eu.gcr.io/data-engineering/jira_downloader:v0.18",
                                cmds=["/usr/local/bin/run_process.sh"],
                                name="jira-translation-projects-01",
                                task_id="jira-translation-projects-01",
                                get_logs=True,
                                dag=dag,
                                # 已删除自定义的空白volumes和volume_mounts配置
                                secrets=[
                                    secret_jira_user,
                                    secret_storage_credentials],
                                env_vars={
                                    'MIGRATION_DATETIME': '2021-01-02T03:04:05',
                                    # 可选：直接把GCP认证路径配置为环境变量，避免代码里硬编码
                                    'GOOGLE_APPLICATION_CREDENTIALS': '/credentials/cloud_storage_credentials.json'
                                }, 
                                )

start >> passing

第二步：修改Python代码中的文件路径 #

把原来的相对路径读取改为绝对路径：

with open('/credentials/jira_credentials.json', 'r') as f:
    creds = json.load(f)

如果GCP认证的环境变量已经在DAG的env_vars中配置，可以删掉代码里手动设置os.environ['GOOGLE_APPLICATION_CREDENTIALS']的逻辑，或者也同步修改为绝对路径。

第三步：快速校验方法 #

如果修改后还是有问题，可以临时修改KubernetesPodOperator的启动命令，验证挂载结果：

cmds=["sh", "-c", "ls -la /credentials && cat /credentials/jira_credentials.json"]

运行任务查看日志，即可确认两个文件是否存在、内容是否正确。

额外注意事项 #

• 确认你创建的两个Kubernetes Secret确实位于default命名空间下，Secret名称、key名称和代码中配置的完全一致
• 确保Airflow的工作服务账号有读取对应Secret的权限
• 如果你使用的是Airflow 2.x及以上版本，建议将airflow.contrib.operators.kubernetes_pod_operator替换为官方推荐的airflow.providers.cncf.kubernetes.operators.kubernetes_pod.KubernetesPodOperator

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