Menggunakan KubernetesPodOperator

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Halaman ini menjelaskan cara menggunakan KubernetesPodOperator untuk men-deploy Pod Kubernetes dari Cloud Composer ke Google Kubernetes Engine yang merupakan bagian dari lingkungan Cloud Composer Anda dan untuk memastikan bahwa lingkungan Anda memiliki sumber daya yang sesuai.

KubernetesPodOperator peluncuran Pod Kubernetes di cluster lingkungan Anda. Sebagai perbandingan, Operator Google Kubernetes Engine menjalankan pod Kubernetes dalam pod yang ditentukan , yang dapat berupa klaster terpisah yang tidak terkait dengan lingkungan fleksibel App Engine. Anda juga dapat membuat dan menghapus cluster menggunakan Operator Google Kubernetes Engine.

KubernetesPodOperator adalah opsi yang bagus jika Anda memerlukan:

  • Dependensi Python kustom yang tidak tersedia melalui PyPI publik repositori resource.
  • Dependensi biner yang tidak tersedia Image pekerja Cloud Composer.

Halaman ini memandu Anda melalui contoh DAG Airflow yang menyertakan hal berikut Konfigurasi KubernetesPodOperator:

Sebelum memulai

  • Di Cloud Composer 3, cluster lingkungan Anda melakukan penskalaan secara otomatis. Workload tambahan yang Anda jalankan menggunakan skala KubernetesPodOperator secara terpisah dari lingkungan Anda. Lingkungan Anda tidak terpengaruh oleh peningkatan permintaan sumber daya, tetapi cluster lingkungan Anda dapat menaikkan dan menurunkan skala bergantung pada resource permintaan tinggi. Harga untuk beban kerja tambahan yang Anda jalankan di mengikuti cluster Model penetapan harga Cloud Composer 3 dan penggunaan SKU Cloud Composer 3.

  • Di Cloud Composer 3, cluster lingkungan Anda berada dalam project tenant. Namun, KubernetesPodOperator berfungsi dengan cara yang sama, tanpa perlu mengubah kode dibandingkan dengan Cloud Composer 2. Pod dijalankan di cluster lingkungan, dalam namespace terisolasi, tetapi dengan akses ke jaringan VPC Anda (jika diaktifkan).

  • Cloud Composer 3 menggunakan cluster GKE dengan Workload Identity Federation untuk GKE. Secara default, Pod yang berjalan di namespace yang baru dibuat atau composer-user-workloads tidak dapat mengakses resource Google Cloud. Saat menggunakan Workload Identity Federation for GKE, akun layanan Kubernetes yang terkait dengan namespace harus dipetakan ke akun layanan Google Cloud, untuk mengaktifkan otorisasi identitas layanan untuk permintaan ke Google API dan layanan IT perusahaan mereka.

    Oleh karena itu, jika Anda menjalankan Pod di namespace composer-user-workloads atau namespace yang baru dibuat di cluster lingkungan Anda, maka Binding IAM antara Kubernetes dan Google Cloud akun layanan tidak dibuat, dan Pod tersebut tidak dapat mengakses resource dari project Google Cloud Anda.

    Jika ingin Pod Anda memiliki akses ke resource Google Cloud, lalu gunakan namespace composer-user-workloads atau buat sendiri seperti yang dijelaskan lebih lanjut.

    Untuk memberikan akses ke resource project, ikuti panduan di Workload Identity Federation for GKE dan menyiapkan binding:

    1. Buat namespace terpisah di cluster lingkungan Anda.
    2. Buat binding antara composer-user-workloads/<namespace_name> Akun Layanan Kubernetes dan akun layanan lingkungan Anda.
    3. Menambahkan anotasi akun layanan lingkungan Anda ke Kubernetes akun layanan Anda.
    4. Saat menggunakan KubernetesPodOperator, tentukan namespace dan Akun layanan Kubernetes di namespace dan Parameter service_account_name.

  • Cloud Composer 3 menggunakan cluster GKE dengan Workload Identitas. Server metadata GKE memerlukan waktu beberapa detik untuk mulai menerima permintaan pada Pod yang baru dibuat. Oleh karena itu, upaya untuk mengautentikasi menggunakan Workload Identity dalam beberapa detik pertama Masa pakai pod mungkin akan gagal. Untuk informasi selengkapnya tentang batasan ini, lihat Batasan Workload Identity.

  • Cloud Composer 3 menggunakan cluster Autopilot yang memperkenalkan konsep class komputasi:

    • Secara default, jika tidak ada class yang dipilih, class general-purpose akan diasumsikan saat Anda membuat Pod menggunakan KubernetesPodOperator.

    • Setiap class dikaitkan dengan properti dan batas resource tertentu, Anda dapat membacanya di Dokumentasi Autopilot. Misalnya, Pod yang berjalan dalam class general-purpose dapat menggunakan memori hingga 110 GiB.

Konfigurasi KubernetesPodOperator

Untuk mengikuti contoh ini, tempatkan seluruh kubernetes_pod_operator.py dalam folder dags/ lingkungan Anda atau tambahkan kode KubernetesPodOperator yang relevan ke DAG.

Bagian berikut menjelaskan setiap konfigurasi KubernetesPodOperator dalam contoh. Untuk informasi tentang setiap variabel konfigurasi, lihat referensi Airflow.

"""Example DAG using KubernetesPodOperator."""
import datetime

from airflow import models
from airflow.kubernetes.secret import Secret
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import (
    KubernetesPodOperator,
)
from kubernetes.client import models as k8s_models

# A Secret is an object that contains a small amount of sensitive data such as
# a password, a token, or a key. Such information might otherwise be put in a
# Pod specification or in an image; putting it in a Secret object allows for
# more control over how it is used, and reduces the risk of accidental
# exposure.
secret_env = Secret(
    # Expose the secret as environment variable.
    deploy_type="env",
    # The name of the environment variable, since deploy_type is `env` rather
    # than `volume`.
    deploy_target="SQL_CONN",
    # Name of the Kubernetes Secret
    secret="airflow-secrets",
    # Key of a secret stored in this Secret object
    key="sql_alchemy_conn",
)
secret_volume = Secret(
    deploy_type="volume",
    # Path where we mount the secret as volume
    deploy_target="/var/secrets/google",
    # Name of Kubernetes Secret
    secret="service-account",
    # Key in the form of service account file name
    key="service-account.json",
)
# If you are running Airflow in more than one time zone
# see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
# for best practices
YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

# If a Pod fails to launch, or has an error occur in the container, Airflow
# will show the task as failed, as well as contain all of the task logs
# required to debug.
with models.DAG(
    dag_id="composer_sample_kubernetes_pod",
    schedule_interval=datetime.timedelta(days=1),
    start_date=YESTERDAY,
) as dag:
    # Only name, namespace, image, and task_id are required to create a
    # KubernetesPodOperator. In Cloud Composer, the config file found at
    # `/home/airflow/composer_kube_config` contains credentials for
    # Cloud Composer's Google Kubernetes Engine cluster that is created
    # upon environment creation.
    kubernetes_min_pod = KubernetesPodOperator(
        # The ID specified for the task.
        task_id="pod-ex-minimum",
        # Name of task you want to run, used to generate Pod ID.
        name="pod-ex-minimum",
        # Entrypoint of the container, if not specified the Docker container's
        # entrypoint is used. The cmds parameter is templated.
        cmds=["echo"],
        # The namespace to run within Kubernetes. In Composer 2 environments
        # after December 2022, the default namespace is
        # `composer-user-workloads`.
        namespace="composer-user-workloads",
        # Docker image specified. Defaults to hub.docker.com, but any fully
        # qualified URLs will point to a custom repository. Supports private
        # gcr.io images if the Composer Environment is under the same
        # project-id as the gcr.io images and the service account that Composer
        # uses has permission to access the Google Container Registry
        # (the default service account has permission)
        image="gcr.io/gcp-runtimes/ubuntu_20_0_4",
        # Specifies path to kubernetes config. The config_file is templated.
        config_file="/home/airflow/composer_kube_config",
        # Identifier of connection that should be used
        kubernetes_conn_id="kubernetes_default",
    )
    kubernetes_template_ex = KubernetesPodOperator(
        task_id="ex-kube-templates",
        name="ex-kube-templates",
        namespace="composer-user-workloads",
        image="bash",
        # All parameters below are able to be templated with jinja -- cmds,
        # arguments, env_vars, and config_file. For more information visit:
        # https://airflow.apache.org/docs/apache-airflow/stable/macros-ref.html
        # Entrypoint of the container, if not specified the Docker container's
        # entrypoint is used. The cmds parameter is templated.
        cmds=["echo"],
        # DS in jinja is the execution date as YYYY-MM-DD, this docker image
        # will echo the execution date. Arguments to the entrypoint. The docker
        # image's CMD is used if this is not provided. The arguments parameter
        # is templated.
        arguments=["{{ ds }}"],
        # The var template variable allows you to access variables defined in
        # Airflow UI. In this case we are getting the value of my_value and
        # setting the environment variable `MY_VALUE`. The pod will fail if
        # `my_value` is not set in the Airflow UI.
        env_vars={"MY_VALUE": "{{ var.value.my_value }}"},
        # Sets the config file to a kubernetes config file specified in
        # airflow.cfg. If the configuration file does not exist or does
        # not provide validcredentials the pod will fail to launch. If not
        # specified, config_file defaults to ~/.kube/config
        config_file="{{ conf.get('core', 'kube_config') }}",
        # Identifier of connection that should be used
        kubernetes_conn_id="kubernetes_default",
    )
    kubernetes_secret_vars_ex = KubernetesPodOperator(
        task_id="ex-kube-secrets",
        name="ex-kube-secrets",
        namespace="composer-user-workloads",
        image="gcr.io/gcp-runtimes/ubuntu_20_0_4",
        startup_timeout_seconds=300,
        # The secrets to pass to Pod, the Pod will fail to create if the
        # secrets you specify in a Secret object do not exist in Kubernetes.
        secrets=[secret_env, secret_volume],
        cmds=["echo"],
        # env_vars allows you to specify environment variables for your
        # container to use. env_vars is templated.
        env_vars={
            "EXAMPLE_VAR": "/example/value",
            "GOOGLE_APPLICATION_CREDENTIALS": "/var/secrets/google/service-account.json",
        },
        # Specifies path to kubernetes config. The config_file is templated.
        config_file="/home/airflow/composer_kube_config",
        # Identifier of connection that should be used
        kubernetes_conn_id="kubernetes_default",
    )
    kubernetes_full_pod = KubernetesPodOperator(
        task_id="ex-all-configs",
        name="pi",
        namespace="composer-user-workloads",
        image="perl:5.34.0",
        # Entrypoint of the container, if not specified the Docker container's
        # entrypoint is used. The cmds parameter is templated.
        cmds=["perl"],
        # Arguments to the entrypoint. The docker image's CMD is used if this
        # is not provided. The arguments parameter is templated.
        arguments=["-Mbignum=bpi", "-wle", "print bpi(2000)"],
        # The secrets to pass to Pod, the Pod will fail to create if the
        # secrets you specify in a Secret object do not exist in Kubernetes.
        secrets=[],
        # Labels to apply to the Pod.
        labels={"pod-label": "label-name"},
        # Timeout to start up the Pod, default is 600.
        startup_timeout_seconds=600,
        # The environment variables to be initialized in the container
        # env_vars are templated.
        env_vars={"EXAMPLE_VAR": "/example/value"},
        # If true, logs stdout output of container. Defaults to True.
        get_logs=True,
        # Determines when to pull a fresh image, if 'IfNotPresent' will cause
        # the Kubelet to skip pulling an image if it already exists. If you
        # want to always pull a new image, set it to 'Always'.
        image_pull_policy="Always",
        # Annotations are non-identifying metadata you can attach to the Pod.
        # Can be a large range of data, and can include characters that are not
        # permitted by labels.
        annotations={"key1": "value1"},
        # Optional resource specifications for Pod, this will allow you to
        # set both cpu and memory limits and requirements.
        # Prior to Airflow 2.3 and the cncf providers package 5.0.0
        # resources were passed as a dictionary. This change was made in
        # https://github.com/apache/airflow/pull/27197
        # Additionally, "memory" and "cpu" were previously named
        # "limit_memory" and "limit_cpu"
        # resources={'limit_memory': "250M", 'limit_cpu': "100m"},
        container_resources=k8s_models.V1ResourceRequirements(
            requests={"cpu": "1000m", "memory": "10G", "ephemeral-storage": "10G"},
            limits={"cpu": "1000m", "memory": "10G", "ephemeral-storage": "10G"},
        ),
        # Specifies path to kubernetes config. The config_file is templated.
        config_file="/home/airflow/composer_kube_config",
        # If true, the content of /airflow/xcom/return.json from container will
        # also be pushed to an XCom when the container ends.
        do_xcom_push=False,
        # List of Volume objects to pass to the Pod.
        volumes=[],
        # List of VolumeMount objects to pass to the Pod.
        volume_mounts=[],
        # Identifier of connection that should be used
        kubernetes_conn_id="kubernetes_default",
        # Affinity determines which nodes the Pod can run on based on the
        # config. For more information see:
        # https://kubernetes.io/docs/concepts/configuration/assign-pod-node/
        # Pod affinity with the KubernetesPodOperator
        # is not supported with Composer 2
        # instead, create a cluster and use the GKEStartPodOperator
        # https://cloud.google.com/composer/docs/using-gke-operator
        affinity={},
    )

Konfigurasi minimal

Untuk membuat KubernetesPodOperator, hanya name Pod, namespace tempat untuk menjalankan pod, image untuk digunakan, dan task_id diperlukan.

Saat Anda menempatkan cuplikan kode berikut di DAG, konfigurasinya akan menggunakan default di /home/airflow/composer_kube_config. Anda tidak perlu memodifikasi kode agar tugas pod-ex-minimum berhasil.

kubernetes_min_pod = KubernetesPodOperator(
    # The ID specified for the task.
    task_id="pod-ex-minimum",
    # Name of task you want to run, used to generate Pod ID.
    name="pod-ex-minimum",
    # Entrypoint of the container, if not specified the Docker container's
    # entrypoint is used. The cmds parameter is templated.
    cmds=["echo"],
    # The namespace to run within Kubernetes. In Composer 2 environments
    # after December 2022, the default namespace is
    # `composer-user-workloads`.
    namespace="composer-user-workloads",
    # Docker image specified. Defaults to hub.docker.com, but any fully
    # qualified URLs will point to a custom repository. Supports private
    # gcr.io images if the Composer Environment is under the same
    # project-id as the gcr.io images and the service account that Composer
    # uses has permission to access the Google Container Registry
    # (the default service account has permission)
    image="gcr.io/gcp-runtimes/ubuntu_20_0_4",
    # Specifies path to kubernetes config. The config_file is templated.
    config_file="/home/airflow/composer_kube_config",
    # Identifier of connection that should be used
    kubernetes_conn_id="kubernetes_default",
)

Konfigurasi template

Airflow mendukung penggunaan Template Jinja. Anda harus mendeklarasikan variabel yang diperlukan (task_id, name, namespace, dan image) dengan operator. Seperti yang ditunjukkan dalam contoh berikut, Anda dapat membuat template semua parameter lainnya dengan Jinja, termasuk cmds, arguments, env_vars, dan config_file.

kubernetes_template_ex = KubernetesPodOperator(
    task_id="ex-kube-templates",
    name="ex-kube-templates",
    namespace="composer-user-workloads",
    image="bash",
    # All parameters below are able to be templated with jinja -- cmds,
    # arguments, env_vars, and config_file. For more information visit:
    # https://airflow.apache.org/docs/apache-airflow/stable/macros-ref.html
    # Entrypoint of the container, if not specified the Docker container's
    # entrypoint is used. The cmds parameter is templated.
    cmds=["echo"],
    # DS in jinja is the execution date as YYYY-MM-DD, this docker image
    # will echo the execution date. Arguments to the entrypoint. The docker
    # image's CMD is used if this is not provided. The arguments parameter
    # is templated.
    arguments=["{{ ds }}"],
    # The var template variable allows you to access variables defined in
    # Airflow UI. In this case we are getting the value of my_value and
    # setting the environment variable `MY_VALUE`. The pod will fail if
    # `my_value` is not set in the Airflow UI.
    env_vars={"MY_VALUE": "{{ var.value.my_value }}"},
    # Sets the config file to a kubernetes config file specified in
    # airflow.cfg. If the configuration file does not exist or does
    # not provide validcredentials the pod will fail to launch. If not
    # specified, config_file defaults to ~/.kube/config
    config_file="{{ conf.get('core', 'kube_config') }}",
    # Identifier of connection that should be used
    kubernetes_conn_id="kubernetes_default",
)

Tanpa mengubah DAG atau lingkungan Anda, tugas ex-kube-templates gagal karena dua error. Log menunjukkan bahwa tugas ini gagal karena variabel yang sesuai tidak ada (my_value). Kesalahan kedua, yang Anda bisa didapatkan setelah memperbaiki {i>error<i} pertama, menunjukkan bahwa tugas gagal karena core/kube_config tidak ditemukan di config.

Untuk memperbaiki kedua kesalahan tersebut, ikuti langkah-langkah yang dijelaskan lebih lanjut.

Untuk menetapkan my_value dengan gcloud atau UI Airflow:

UI Airflow

Di UI Airflow 2:

  1. Buka UI Airflow.

  2. Di toolbar, pilih Admin > Variabel.

  3. Di halaman Daftar Variabel, klik Tambahkan data baru.

  4. Di halaman Add Variable, masukkan informasi berikut:

    • Kunci:my_value
    • Nilai: example_value

  5. Klik Simpan.

gcloud

Untuk Airflow 2, masukkan perintah berikut:

gcloud composer environments run ENVIRONMENT \
    --location LOCATION \
    variables set -- \
    my_value example_value

Ganti:

  • ENVIRONMENT dengan nama lingkungan.
  • LOCATION dengan region tempat lingkungan berada.

Untuk merujuk ke config_file kustom (file konfigurasi Kubernetes), ganti opsi konfigurasi Airflow kube_config ke konfigurasi Kubernetes yang valid:

Bagian Kunci Nilai
core kube_config /home/airflow/composer_kube_config

Tunggu beberapa menit sampai lingkungan Anda diperbarui. Selanjutnya jalankan tugas ex-kube-templates lagi dan verifikasi bahwa Tugas ex-kube-templates berhasil.

Konfigurasi penuh

Contoh ini menunjukkan semua variabel yang dapat Anda konfigurasi dalam KubernetesPodOperator. Anda tidak perlu memodifikasi kode untuk agar tugas ex-all-configs berhasil.

Untuk mengetahui detail tentang setiap variabel, lihat metode Referensi KubernetesPodOperator Airflow.

kubernetes_full_pod = KubernetesPodOperator(
    task_id="ex-all-configs",
    name="pi",
    namespace="composer-user-workloads",
    image="perl:5.34.0",
    # Entrypoint of the container, if not specified the Docker container's
    # entrypoint is used. The cmds parameter is templated.
    cmds=["perl"],
    # Arguments to the entrypoint. The docker image's CMD is used if this
    # is not provided. The arguments parameter is templated.
    arguments=["-Mbignum=bpi", "-wle", "print bpi(2000)"],
    # The secrets to pass to Pod, the Pod will fail to create if the
    # secrets you specify in a Secret object do not exist in Kubernetes.
    secrets=[],
    # Labels to apply to the Pod.
    labels={"pod-label": "label-name"},
    # Timeout to start up the Pod, default is 600.
    startup_timeout_seconds=600,
    # The environment variables to be initialized in the container
    # env_vars are templated.
    env_vars={"EXAMPLE_VAR": "/example/value"},
    # If true, logs stdout output of container. Defaults to True.
    get_logs=True,
    # Determines when to pull a fresh image, if 'IfNotPresent' will cause
    # the Kubelet to skip pulling an image if it already exists. If you
    # want to always pull a new image, set it to 'Always'.
    image_pull_policy="Always",
    # Annotations are non-identifying metadata you can attach to the Pod.
    # Can be a large range of data, and can include characters that are not
    # permitted by labels.
    annotations={"key1": "value1"},
    # Optional resource specifications for Pod, this will allow you to
    # set both cpu and memory limits and requirements.
    # Prior to Airflow 2.3 and the cncf providers package 5.0.0
    # resources were passed as a dictionary. This change was made in
    # https://github.com/apache/airflow/pull/27197
    # Additionally, "memory" and "cpu" were previously named
    # "limit_memory" and "limit_cpu"
    # resources={'limit_memory': "250M", 'limit_cpu': "100m"},
    container_resources=k8s_models.V1ResourceRequirements(
        requests={"cpu": "1000m", "memory": "10G", "ephemeral-storage": "10G"},
        limits={"cpu": "1000m", "memory": "10G", "ephemeral-storage": "10G"},
    ),
    # Specifies path to kubernetes config. The config_file is templated.
    config_file="/home/airflow/composer_kube_config",
    # If true, the content of /airflow/xcom/return.json from container will
    # also be pushed to an XCom when the container ends.
    do_xcom_push=False,
    # List of Volume objects to pass to the Pod.
    volumes=[],
    # List of VolumeMount objects to pass to the Pod.
    volume_mounts=[],
    # Identifier of connection that should be used
    kubernetes_conn_id="kubernetes_default",
    # Affinity determines which nodes the Pod can run on based on the
    # config. For more information see:
    # https://kubernetes.io/docs/concepts/configuration/assign-pod-node/
    # Pod affinity with the KubernetesPodOperator
    # is not supported with Composer 2
    # instead, create a cluster and use the GKEStartPodOperator
    # https://cloud.google.com/composer/docs/using-gke-operator
    affinity={},
)

Informasi tentang Penyedia Kubernetes CNCF

GKEStartPodOperator dan KubernetesPodOperator diimplementasikan dalam Penyedia apache-airflow-providers-cncf-kubernetes.

Untuk catatan rilis defailed untuk penyedia Kubernetes CNCF, lihat situs Penyedia Kubernetes CNCF.

Versi 6.0.0

Dalam versi 6.0.0 paket Penyedia Kubernetes CNCF, koneksi kubernetes_default digunakan secara default di KubernetesPodOperator.

Jika Anda menentukan koneksi kustom di versi 5.0.0, koneksi kustom ini masih digunakan oleh operator. Untuk beralih kembali menggunakan kubernetes_default jarak jauh, Anda mungkin perlu menyesuaikan DAG dengan tepat.

Versi 5.0.0

Versi ini memperkenalkan beberapa perubahan yang tidak kompatibel dengan versi sebelumnya dibandingkan dengan versi 4.4.0. Yang paling penting terkait dengan koneksi kubernetes_default yang tidak digunakan dalam versi 5.0.0.

  • Koneksi kubernetes_default perlu diubah. Jalur konfigurasi Kube harus disetel ke /home/airflow/composer_kube_config (seperti yang ditunjukkan pada Gambar 1) Sebagai alternatif, config_file harus ditambahkan ke konfigurasi KubernetesPodOperator (seperti yang ditunjukkan dalam kode berikut contoh).
Kolom jalur konfigurasi Kube di UI Airflow
Gambar 1. UI Airflow, mengubah koneksi kubernetes_default (klik untuk memperbesar)
  • Ubah kode tugas menggunakan KubernetesPodOperator dengan cara berikut:
KubernetesPodOperator(
  # config_file parameter - can be skipped if connection contains this setting
  config_file="/home/airflow/composer_kube_config",
  # definition of connection to be used by the operator
  kubernetes_conn_id='kubernetes_default',
  ...
)

Untuk mengetahui informasi selengkapnya tentang Versi 5.0.0, baca Catatan Rilis Penyedia Kubernetes CNCF

Pemecahan masalah

Tips untuk memecahkan masalah kegagalan Pod

Selain memeriksa log tugas di UI Airflow, periksa juga log berikut:

  • Output scheduler dan pekerja Airflow:

    1. Di Konsol Google Cloud, buka halaman Environments.

      Buka Lingkungan

    2. Ikuti link DAGs untuk lingkungan Anda.

    3. Naik satu level ke bucket di lingkungan Anda.

    4. Tinjau log di logs/<DAG_NAME>/<TASK_ID>/<EXECUTION_DATE> folder tersebut.

  • Log pod mendetail di konsol Google Cloud di bagian workload GKE. Log ini mencakup pod file YAML definisi, peristiwa pod, dan detail pod.

Kode yang ditampilkan bukan nol saat juga menggunakan GKEStartPodOperator

Saat menggunakan KubernetesPodOperator dan GKEStartPodOperator, kode pengembalian titik masuk container menentukan apakah tugas dianggap berhasil atau tidak. Kode hasil yang bukan nol menunjukkan kegagalan.

Pola yang umum saat menggunakan KubernetesPodOperator dan GKEStartPodOperator akan menjalankan skrip shell sebagai container titik entri untuk mengelompokkan beberapa operasi dalam kontainer.

Jika Anda menulis naskah seperti itu, sebaiknya sertakan perintah set -e di bagian atas skrip sehingga perintah yang gagal dalam skrip akan menghentikan skrip dan menyebarkan kegagalan ke instance tugas Airflow.

Waktu tunggu pod

Waktu tunggu default untuk KubernetesPodOperator adalah 120 detik, yang dapat mengakibatkan waktu tunggu habis sebelum download gambar yang lebih besar. Anda dapat meningkatkan waktu tunggu dengan mengubah startup_timeout_seconds saat Anda membuat KubernetesPodOperator.

Saat waktu tunggu pod habis, log spesifik per tugas tersedia di UI Airflow. Contoh:

Executing <Task(KubernetesPodOperator): ex-all-configs> on 2018-07-23 19:06:58.133811
Running: ['bash', '-c', u'airflow run kubernetes-pod-example ex-all-configs 2018-07-23T19:06:58.133811 --job_id 726 --raw -sd DAGS_FOLDER/kubernetes_pod_operator_sample.py']
Event: pod-name-9a8e9d06 had an event of type Pending
...
...
Event: pod-name-9a8e9d06 had an event of type Pending
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 27, in <module>
    args.func(args)
  File "/usr/local/lib/python2.7/site-packages/airflow/bin/cli.py", line 392, in run
    pool=args.pool,
  File "/usr/local/lib/python2.7/site-packages/airflow/utils/db.py", line 50, in wrapper
    result = func(*args, **kwargs)
  File "/usr/local/lib/python2.7/site-packages/airflow/models.py", line 1492, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python2.7/site-packages/airflow/contrib/operators/kubernetes_pod_operator.py", line 123, in execute
    raise AirflowException('Pod Launching failed: {error}'.format(error=ex))
airflow.exceptions.AirflowException: Pod Launching failed: Pod took too long to start

Waktu Tunggu Pod juga dapat terjadi saat Akun Layanan Cloud Composer tidak memiliki izin IAM yang diperlukan untuk menjalankan tugas di tangan. Untuk memverifikasinya, lihat error level pod menggunakan Dasbor GKE untuk melihat log Anda Workload tertentu, atau menggunakan Cloud Logging.

Gagal membuat koneksi baru

Upgrade otomatis diaktifkan secara default di cluster GKE. Jika kumpulan node berada dalam cluster yang sedang diupgrade, Anda mungkin melihat hal berikut {i>error<i}:

<Task(KubernetesPodOperator): gke-upgrade> Failed to establish a new
connection: [Errno 111] Connection refused

Untuk memeriksa apakah cluster Anda diupgrade, di Konsol Google Cloud, buka Halaman Cluster Kubernetes dan cari ikon pemuatan di sebelah Anda nama cluster lingkungan.

Langkah selanjutnya