气流调度器无法连接到 Kubernetes 服务 api答案

【问题标题】：Airflow scheduler can not connect to Kubernetes service api气流调度器无法连接到 Kubernetes 服务 api
【发布时间】：2020-05-08 18:01:59
【问题描述】：

我正在尝试使用 Kubernetes 执行程序设置气流，并且在调度程序容器启动时它会挂起一段时间，然后我收到 https 超时错误，如下所示。消息中的 ip 地址是正确的，在容器内我可以运行 curl kubernetes:443 或 curl 10.96.0.1:443 或 nc -zv 10.96.0.1 443 所以我假设没有防火墙或阻止访问。

我正在使用本地 kubernetes 以及 aws EKS 但同样的错误，我可以看到不同集群中的 ip 变化。

我已经查看了谷歌以找到解决方案，但没有看到类似的案例。

│   File "/usr/local/lib/python3.6/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 335, in run                                                                                                                                                                                                                               │
│     self.worker_uuid, self.kube_config)                                                                                                                                                                                                                                                                                                          │
│   File "/usr/local/lib/python3.6/site-packages/airflow/contrib/executors/kubernetes_executor.py", line 359, in _run                                                                                                                                                                                                                              │
│     **kwargs):                                                                                                                                                                                                                                                                                                                                   │
│   File "/usr/local/lib/python3.6/site-packages/kubernetes/watch/watch.py", line 144, in stream                                                                                                                                                                                                                                                   │
│     for line in iter_resp_lines(resp):                                                                                                                                                                                                                                                                                                           │
│   File "/usr/local/lib/python3.6/site-packages/kubernetes/watch/watch.py", line 48, in iter_resp_lines                                                                                                                                                                                                                                           │
│     for seg in resp.read_chunked(decode_content=False):                                                                                                                                                                                                                                                                                          │
│   File "/usr/local/lib/python3.6/site-packages/urllib3/response.py", line 781, in read_chunked                                                                                                                                                                                                                                                   │
│     self._original_response.close()                                                                                                                                                                                                                                                                                                              │
│   File "/usr/local/lib/python3.6/contextlib.py", line 99, in __exit__                                                                                                                                                                                                                                                                            │
│     self.gen.throw(type, value, traceback)                                                                                                                                                                                                                                                                                                       │
│   File "/usr/local/lib/python3.6/site-packages/urllib3/response.py", line 430, in _error_catcher                                                                                                                                                                                                                                                 │
│     raise ReadTimeoutError(self._pool, None, "Read timed out.")                                                                                                                                                                                                                                                                                  │
│ urllib3.exceptions.ReadTimeoutError: HTTPSConnectionPool(host='10.96.0.1', port=443): Read timed out.

更新：我找到了我的问题，但还没有解决方案。 https://github.com/kubernetes-client/python/issues/990

【问题讨论】：

您是否使用 HELM 图表在 kubernetes 中进行设置？提供的日志来自 k8 的容器？请提供有关实施的更多详细信息，我会尽力帮助您进行设置。
@rabello 是的，我正在使用 helm3，日志来自调度程序部署 pod。代码是这个 repo 的修改版本：github.com/duy0611/airflow-eks-helm-demo
我已经使用 Helm3 bitnami/airflow 在 GKE 中安装了 Airflow，一切运行良好。您使用这个版本有什么特别的原因吗？
我也将问题隔离到 kubernetes python 客户端。所以仍然无法理解为什么它不会发生在其他人身上。这是关于超时的事情。
@AsavPatel 我写的参数值是这样的，kube_client_request_args = {"_request_timeout" : [60,60]} 不再出现该错误。此行位于 configmap 定义中的文件中。

标签： kubernetes airflow kubernetes-helm airflow-scheduler

【解决方案1】：

有一个选项可以通过 ENV 变量设置值。在您的charts/airflow.yaml 文件中，您可以按如下方式设置变量，这应该可以解决您的问题，

AIRFLOW__KUBERNETES__KUBE_CLIENT_REQUEST_ARGS: {"_request_timeout" : [50, 50]}

公关参考：https://github.com/apache/airflow/pull/6643
问题讨论：https://issues.apache.org/jira/browse/AIRFLOW-6040

airflow.yaml 完整代码

airflow:
  image:
     repository: airflow-docker-local
     tag: 1
  executor: Kubernetes
  service:
    type: LoadBalancer
  config:
    AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://postgres:airflow@airflow-postgresql:5432/airflow
    AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://postgres:airflow@airflow-postgresql:5432/airflow
    AIRFLOW__CELERY__BROKER_URL: redis://:airflow@airflow-redis-master:6379/0
    AIRFLOW__CORE__REMOTE_LOGGING: True
    AIRFLOW__CORE__REMOTE_LOG_CONN_ID: my_s3_connection
    AIRFLOW__CORE__REMOTE_BASE_LOG_FOLDER: s3://xxx-airflow/logs
    AIRFLOW__WEBSERVER__LOG_FETCH_TIMEOUT_SEC: 25
    AIRFLOW__CORE__LOAD_EXAMPLES: True
    AIRFLOW__WEBSERVER__EXPOSE_CONFIG: True
    AIRFLOW__CORE__FERNET_KEY: -xyz=
    AIRFLOW__KUBERNETES__WORKER_CONTAINER_REPOSITORY: airflow-docker-local
    AIRFLOW__KUBERNETES__WORKER_CONTAINER_TAG: 1
    AIRFLOW__KUBERNETES__WORKER_CONTAINER_IMAGE_PULL_POLICY: Never
    AIRFLOW__KUBERNETES__WORKER_SERVICE_ACCOUNT_NAME: airflow
    AIRFLOW__KUBERNETES__DAGS_VOLUME_CLAIM: airflow
    AIRFLOW__KUBERNETES__NAMESPACE: airflow
    AIRFLOW__KUBERNETES__KUBE_CLIENT_REQUEST_ARGS: {"_request_timeout" : [50, 50]}


persistence:
  enabled: true
  existingClaim: ''

workers:
  enabled: true

postgresql:
  enabled: true

redis:
  enabled: true

【讨论】：

我这样做了，request_timeout 还不够，我还添加了超时参数。但这不仅与气流有关，我有示例代码仅使用 kubernetes python 库、客户端、监视和过滤 pod 标签在没有超时参数的情况下不起作用。