在气流 GKEStartPodOperator 操作员中配置卷

Question

我有一个谷歌云作曲家环境。在我的 DAG 中，我想在 GKE 中创建一个 pod。当我基于不需要任何卷配置或机密的 docker 容器部署一个简单的应用程序时，一切正常，例如：

kubernetes_max = GKEStartPodOperator(
    # The ID specified for the task.
    task_id="python-simple-app",
    # Name of task you want to run, used to generate Pod ID.
    name="python-demo-app",
    project_id=PROJECT_ID,
    location=CLUSTER_REGION,
    cluster_name=CLUSTER_NAME,
    # Entrypoint of the container, if not specified the Docker container's
    # entrypoint is used. The cmds parameter is templated.
    cmds=["python", "app.py"],
    namespace="production",
    image="gcr.io/path/to/lab-python-job:latest",
)

但是当我有一个应用程序需要访问我的 GKE 集群卷时，我需要在我的 pod 中配置卷。问题是文档对此并不清楚。我曾经foud 的唯一例子是：

volume = k8s.V1Volume(
    name='test-volume',
    persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name='test-volume'),
)

虽然我的清单文件中的卷（我使用它从本地部署我的应用程序）看起来像这样：

volumes:
  - name: volume-prod
    secret:
      secretName: volume-prod
      items:
        - key: config
          path: config.json
        - key: another_config
          path: another_config.conf
        - key: random-ca
          path: random-ca.pem

因此，当我比较两个卷在控制台中的外观时（当我手动部署成功运行的清单文件时，以及当我使用失败的 clod composer 部署 pod 时）：

• 成功运行——Manifest文件：

  量产
  名称：volume-prod
  类型：秘密
  源卷标识符：volume-prod

• 失败的运行 - Composer GKEStartPodOperator:

  量产
  名称：volume-prod
  类型：emptyDir
  源卷标识符：节点的默认介质

如何通过 Cloud Composer 配置我的 pod，使其能够读取集群的卷？

Answer 1

KubernetesPodOperator/GKEStartOperator 只是 python Kubernetes sdk 的一个包装器 - 我同意它在 Airflow/Cloud Composer 文档中没有很好的记录，但用于 Kubernetes 的 Python SDK 本身有很好的记录。

从 kubernetes python sdk 文档开始：https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/V1PodSpec.md

您会注意到KubernetesPodOperator/GKEStartOperator 采用的参数与此规范相匹配。如果你深入研究算子的源代码，你会发现算子只不过是一个构建器，它创建一个kubernetes.client.models.V1Pod 对象并使用 API 来部署 pod。

operator 采用volumes 参数，该参数应为List[V1Volume] 类型，其中V1Volume 的文档为here。

因此，在您的情况下，您需要提供：

from kubernetes.client import models as k8s

kubernetes_max = GKEStartPodOperator(
    # The ID specified for the task.
    task_id="python-simple-app",
    # Name of task you want to run, used to generate Pod ID.
    name="python-demo-app",
    project_id=PROJECT_ID,
    location=CLUSTER_REGION,
    cluster_name=CLUSTER_NAME,
    # Entrypoint of the container, if not specified the Docker container's
    # entrypoint is used. The cmds parameter is templated.
    cmds=["python", "app.py"],
    namespace="production",
    image="gcr.io/path/to/lab-python-job:latest",
    volumes=[
        k8s.V1Volume(
            secret=k8s.V1SecretVolumeSource(
                secret_name="volume-prod",
                items=[
                    k8s.V1KeyToPath(key="config", path="config.json"),
                    k8s.V1KeyToPath(key="another_config", path="another_config.conf"),
                    k8s.V1KeyToPath(key="random-ca", path="random-ca.pem"),
                ],
            )
        )
    ]
)

或者，您可以将清单提供给 GKEStartPodOperator 中的 pod_template_file 参数 - 这需要供气流内部的工作人员使用。

使用此 Operator 在 Airflow 中创建 pod 有 3 种方法：

1. 使用运算符的参数指定您需要的内容，并让运算符为您构建V1Pod。
2. 通过传入 pod_template_file 参数来提供清单。
3. 使用 Kubernetes sdk 自己创建一个 V1Pod 对象并将其传递给 full_pod_spec 参数。