kubernetes - 使用KubernetesExecutor的pod_mutation_hook函数不适用于在Kubernetes中运行的 Airflow

Question

我正在尝试将在kubernetes中运行的气流部署从CeleryExecutor迁移到KubernetesExecutor。在本地开发环境中运行一切正常(在minikube上运行)，但是我需要在生产环境中加载sidecar容器以运行允许我连接到sql数据库的代理。经过一番谷歌搜索之后，似乎应该在airflow_local_settings.py上某个地方的$PYTHONPATH文件中定义pod_mutation_hook函数，这是应该如何完成的。

首先，我尝试根据this示例在配置映射中定义此设置。例如

apiVersion: v1
kind: ConfigMap
metadata:
  name: airflow-config
  namespace: dev
data:
  ...

  AIRFLOW__KUBERNETES__LOGS_VOLUME_CLAIM: "airflow-logs"

  AIRFLOW__KUBERNETES__AIRFLOW_LOCAL_SETTINGS_CONFIGMAP: "airflow-config"
  ...

  airflow_local_settings.py: |
    from airflow.contrib.kubernetes.pod import Pod

    def pod_mutation_hook(pod: Pod):
        extra_labels = {
            "test-label": "True",
        }
        pod.labels.update(extra_labels)

我在airflow.cfg文件中指定了此configmap，并对其进行了拾取和安装，所有其他env变量均正常运行，但是pod_mutation_hook似乎没有运行，因为没有标签被添加到kubernetes执行程序启动的结果pod中(请注意，日志容量声明也在此处指定，并且可以正常运行)。

接下来，我尝试按照注释here中的建议在airflow_local_settings.py下为作业启动气流的图像中定义$AIRFLOW_HOME/configs/airflow_local_settings.py文件。我还从上面的airflow-config configmap中删除了相关部分。由于它也缺少指定的标签，因此似乎对为该作业创建的结果 Pane 没有影响。

因此，我不确定如何进行此操作，因为我不知道如何指定airflow_local_settings.py文件和pod_mutation_hook函数，以便它们在运行之前实际对pod进行突变。任何帮助将不胜感激。谢谢。

Answer 1

我遇到了同样的问题，请确保可以从调度程序中导入airflow_local_settings。您必须将这些更改烘焙到图像中。

WORKDIR ${AIRFLOW_USER_HOME}
ENV PYTHONPATH  $PYTHONPATH:$AIRFLOW_HOME/config/
COPY airflow_local_settings.py $AIRFLOW_HOME/config/airflow_local_settings.py

使用上面突出显示的configmap，可以将它们带入执行程序，但此时不需要，因此是一种无用的设置。随时阅读源代码:

https://github.com/apache/airflow/blob/8465d66f05baeb73dd4479b019515c069444616e/airflow/settings.py