Airflow DockerOperator访问连接变量及动态传参方案问询

DockerOperator启动的容器和Airflow运行环境完全隔离，没有元数据库访问权限，不要在容器内部尝试直接读取Airflow连接/变量。所有动态配置的读取动作全部放在Airflow Worker侧完成，再通过命令参数、环境变量的方式传给容器，既安全又不需要修改容器内的脚本逻辑。

现存代码问题 #

• 参数键名不匹配：DAG params里定义的key是SOURCE_ONE，command模板里却引用params.SOURCE_TWO，渲染时会报找不到键的错误
• 模板渲染逻辑错误：直接在Jinja模板里写{{ conn[params.SOURCE_TWO] }}输出的是Connection对象的字符串描述，不是结构化的连接配置字典，容器里的脚本拿到的是无效值
• 引号转义错误：command模板里嵌套使用单引号，会导致Jinja渲染后的命令语法错误
• 导入路径过时：airflow.operators.docker_operator.DockerOperator是旧版本导入路径，Airflow 2.x正式版已经迁移到docker provider包内，旧路径会触发废弃告警
• 配置逻辑不合理：强制force_pull=True会导致本地构建的镜像每次运行都去远程仓库拉取，本地测试时会直接报错

修正步骤 #

1. 调整DAG参数定义，明确暴露credential_id入参，支持用户触发DAG时自定义传入，设置合理的默认值
2. 增加前置Python任务，在Airflow Worker侧根据用户传入的conn_id拉取对应连接配置，转成结构化字典后和其他参数一起序列化，通过XCom传给下游任务
3. 简化DockerOperator的command模板，去掉模板内的复杂逻辑，直接读取前置任务输出的序列化参数，修复引号转义问题
4. 敏感凭证优先通过环境变量传递，避免明文出现在进程命令行参数中

修正后完整代码 #

from datetime import datetime
from json import dumps

import pendulum
from airflow import DAG
from airflow.decorators import task
from airflow.models import XCom, Connection
from airflow.models.param import Param
from airflow.providers.docker.operators.docker import DockerOperator
from airflow.utils.db import provide_session


local_tz = pendulum.timezone("America/Los_Angeles")
args = {"owner": "Airflow"}
SYNC_SCRIPT_DAG = "sync_script_dag_v1"


@provide_session
def cleanup_xcom(session=None, **context):
    print("Cleaning up!!!")
    dag = context["dag"]
    dag_id = dag._dag_id
    session.query(XCom).filter(XCom.dag_id == dag_id).delete()


with DAG(
    dag_id=SYNC_SCRIPT_DAG,
    default_args=args,
    catchup=False,
    start_date=datetime(2020, 7, 8, tzinfo=local_tz),
    max_active_runs=1,
    tags=["production"],
    schedule_interval=None,
    params={
        # 暴露连接ID参数，用户触发时可自定义传入
        "CREDENTIAL_CONN_ID": Param(
            default="source_two_credential",
            type="string",
            description="脚本使用的凭证对应的Airflow连接ID"
        ),
        "LOG_LEVEL": Param(default="INFO", type="string"),
        "NUM_WORKER_THREADS": Param(default="10", type="string"),
    },
    on_failure_callback=cleanup_xcom,
    on_success_callback=cleanup_xcom,
) as dag:

    @task(task_id="get_airflow_params_task")
    def get_airflow_params(**context):
        airflow_params = context.get("params")
        # 在Airflow侧拉取对应连接配置，不需要容器访问元库
        source_conn = Connection.get_connection_from_secrets(airflow_params["CREDENTIAL_CONN_ID"])
        # 组装脚本需要的连接配置结构
        airflow_params["source_credential"] = {
            "host": source_conn.host,
            "port": source_conn.port,
            "username": source_conn.login,
            "password": source_conn.password,
            "extra_config": source_conn.extra_dejson
        }
        return dumps(airflow_params)

    get_params_task = get_airflow_params()

    run_script_task = DockerOperator(
        task_id="run_sync_script",
        image="sync-script:latest",
        api_version="auto",
        auto_remove=True,
        # 修复引号转义问题，直接读取前置任务序列化好的全量参数
        command="python main.py --airflow_params '{{ ti.xcom_pull(task_ids=\"get_airflow_params_task\") }}'",
        environment={
            "YML_CONFIG": "yml_config",
            # 如果脚本支持读环境变量，也可以直接把敏感配置通过环境变量传入，避免命令行泄露
            # "SOURCE_USER": "{{ conn[params.CREDENTIAL_CONN_ID].login }}",
            # "SOURCE_PWD": "{{ conn[params.CREDENTIAL_CONN_ID].password }}",
        },
        docker_url="unix://var/run/docker.sock",
        network_mode="host",
        force_pull=False, # 本地镜像不需要强制拉取，推到远程仓库后再按需开启
        docker_conn_id="harbor_credentials",
    )

    get_params_task >> run_script_task

额外注意事项 #

• 禁止给Docker容器配置Airflow元数据库连接串让容器自行读取配置，该方式会导致容器获得所有Airflow存储的凭证访问权限，存在极大安全风险
• 如果传递敏感参数，建议开启Airflow的敏感字段脱敏配置，避免密码等信息明文打印在任务日志中
• 如果脚本不需要全量参数JSON，也可以直接在DockerOperator的environment字段里通过Jinja模板直接引用对应连接的属性，脚本直接读取环境变量即可，省去JSON解析步骤

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