尝试创建命名空间 Pod 时出现异常 - 错误请求

Question

我是第一次使用 Airflow DAG 和 Kubernetes。我有一个连接到 AWS S3 并读取一些文件的 Python 脚本。如果我使用 bash 在 Docker container/image 中运行它，这很好用。但是当我尝试使用 K8s pod 从气流任务运行这个 docker 时，我收到以下错误（我用 XXXXX 替换了一些敏感值）

    [2022-02-08 22:48:55,795] {kubernetes_pod.py:365} INFO - creating pod with labels {'dag_id': 'ECO_CELLS_POLYGON_STORES', 'task_id': 'process_AR', 'execution_date': '2022-02-08T224216.4628350000-e866f2011', 'try_number': '1'} and launcher <airflow.providers.cncf.kubernetes.utils.pod_launcher.PodLauncher object at 0x7f649be71410>
[2022-02-08 22:48:55,812] {pod_launcher.py:93} ERROR - Exception when attempting to create Namespaced Pod: {
  "apiVersion": "v1",
  "kind": "Pod",
  "metadata": {
    "annotations": {},
    "labels": {
      "airflow_version": "2.0.0-astro.8",
      "kubernetes_pod_operator": "True",
      "dag_id": "ECO_CELLS_POLYGON_STORES",
      "task_id": "process_AR",
      "execution_date": "2022-02-08T224216.4628350000-e866f2011",
      "try_number": "1"
    },
    "name": "k8s-pod-ml-operator.3aada8ada8df491ea63e9319bf779d10",
    "namespace": "default"
  },
  "spec": {
    "affinity": {},
    "containers": [
      {
        "args": [],
        "command": [
          "python",
          "main.py"
        ],
        "env": {
          "AWS_ACCESS_KEY_ID": "XXXXXX",
          "AWS_SECRET_ACCESS_KEY": "***",
          "AWS_BUCKET_NAME": "XXXXXX-dev",
          "SNOWFLAKE_SERVER": "XXXXXX",
          "SNOWFLAKE_LOGIN": "XXXXXX",
          "SNOWFLAKE_PASSWORD": "***",
          "SNOWFLAKE_ACCOUNT": "XXXXXX",
          "SNOWFLAKE_DATABASE": "XXXXXX",
          "SNOWFLAKE_WAREHOUSE": "XXXXXX",
          "COUNTRY": "AR",
          "S3_PROJECT": "ecom_polygon_stores",
          "S3_TEAM_VERTICAL": "ecommerce"
        },
        "envFrom": [],
        "image": "ecom_polygon_stores:v1.0.7",
        "imagePullPolicy": "Never",
        "name": "base",
        "ports": [],
        "resources": {},
        "volumeMounts": []
      }
    ],
    "hostNetwork": false,
    "imagePullSecrets": [],
    "initContainers": [],
    "restartPolicy": "Never",
    "securityContext": {},
    "serviceAccountName": "default",
    "tolerations": [],
    "volumes": []
  }
}
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/airflow/providers/cncf/kubernetes/utils/pod_launcher.py", line 89, in run_pod_async
    body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/api/core_v1_api.py", line 6174, in create_namespaced_pod
    (data) = self.create_namespaced_pod_with_http_info(namespace, body, **kwargs)  # noqa: E501
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/api/core_v1_api.py", line 6265, in create_namespaced_pod_with_http_info
    collection_formats=collection_formats)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/api_client.py", line 345, in call_api
    _preload_content, _request_timeout)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/api_client.py", line 176, in __call_api
    _request_timeout=_request_timeout)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/api_client.py", line 388, in request
    body=body)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/rest.py", line 278, in POST
    body=body)
  File "/usr/local/lib/python3.7/site-packages/kubernetes/client/rest.py", line 231, in request
    raise ApiException(http_resp=r)
kubernetes.client.rest.ApiException: (400)
Reason: Bad Request
HTTP response headers: HTTPHeaderDict({'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'X-Kubernetes-Pf-Flowschema-Uid': '319700db-6333-4a4d-885c-1f45a0cd13a3', 'X-Kubernetes-Pf-Prioritylevel-Uid': '4d5b12e4-65e9-4ab9-ad63-de6f29ca0b6d', 'Date': 'Tue, 08 Feb 2022 22:48:55 GMT', 'Content-Length': '487'})
HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"Pod in version \"v1\" cannot be handled as a Pod: v1.Pod.Spec: v1.PodSpec.Containers: []v1.Container: v1.Container.Env: []v1.EnvVar: decode slice: expect [ or n, but found {, error found in #10 byte of ...|, \"env\": {\"AWS_ACCES|..., bigger context ...|s\": [], \"command\": [\"python\", \"main.py\"], \"env\": {\"AWS_ACCESS_KEY_ID\": \"AXXXXXXXXXXXXXXXXXX6\", \"AWS_|...","reason":"BadRequest","code":400}

我不确定从这里去哪里...从我读到的错误说它期望一个 [ 而不是 { in "env": {"AWS_ACCESS_KEY_ID".. . 但我不确定如何更正它，因为我像这样传递了这些参数：

        self.env_vars = {
            'AWS_ACCESS_KEY_ID': s3_connection.login,
            'AWS_SECRET_ACCESS_KEY': s3_connection.password,
            'AWS_BUCKET_NAME': bucket_name,
            'SNOWFLAKE_SERVER': str(snowflake_connection.host),
            'SNOWFLAKE_LOGIN': str(snowflake_connection.login),
            'SNOWFLAKE_PASSWORD': str(snowflake_connection.password),
            'SNOWFLAKE_ACCOUNT': str(snowflake_connection.extra_dejson['account']),
            'SNOWFLAKE_DATABASE': str(snowflake_connection.extra_dejson['database']),
            'SNOWFLAKE_WAREHOUSE': str(snowflake_connection.extra_dejson['warehouse']),
            'COUNTRY': code,
            'S3_PROYECT': s3_project,
            'S3_TEAM_VERTICAL': s3_team_vertical
        }

有什么建议吗？

Answer 1

您的 env: 格式不正确；人们可以通过两种不同的方式看到这一点：(1) env: in the PodSpec 是 list 的 {name: "", value: ""} 项 (2) 无论怎样，错误消息中发出的结构都是错误的: "env": {"AWS_ACCESS_KEY_ID": "ASIAWCMTKGYGDU6KEOD6", "AWS_|... 因为没有 {"":"",""

这样的数据形状

我手边没有任何 Airflow 参考文档链接，但您需要检查它们以确保 self.env_vars 是 Airflow 所期望的，因为 python 承担了全部负担程序员的正确性

Answer 2

确实env var格式不对。对于 2.0.0 气流版本，它需要是 k8s.V1EnvVar 对象的列表（因为我使用的是 k8s）。

每个 V1EnvVar 都需要有这样的格式：

from kubernetes.client import models as k8s

 my_var = k8s.V1EnvVar(name='my_var_name', value=1234)

尝试创建命名空间 Pod 时出现异常 - 错误请求

Exception when attempting to create Namespaced Pod - Bad Request

python

docker

kubernetes

airflow