Utilizzo di KubernetesPodOperator

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Questa pagina descrive come utilizzare KubernetesPodOperator per eseguire il deployment dei pod Kubernetes da Cloud Composer nel cluster Google Kubernetes Engine che fa parte del tuo ambiente Cloud Composer e per garantire che l'ambiente disponga delle risorse appropriate.

KubernetesPodOperator avvia i pod Kubernetes nel cluster del tuo ambiente. In confronto, gli operatori di Google Kubernetes Engine eseguono i pod Kubernetes in un cluster specificato, che può essere un cluster separato non correlato al tuo ambiente. Puoi anche creare ed eliminare i cluster usando gli operatori di Google Kubernetes Engine.

KubernetesPodOperator è una buona soluzione se hai bisogno di:

• Dipendenze Python personalizzate non disponibili tramite il repository PyPI pubblico.
• Dipendenze binarie che non sono disponibili nell'immagine worker di Cloud Composer di stock.

Nota: per ulteriori informazioni sull'installazione di pacchetti ospitati in un repository di pacchetti privato o per opzioni aggiuntive per la gestione delle dipendenze, consulta Installazione delle dipendenze Python.

Questa pagina illustra un esempio di DAG Airflow che include le seguenti configurazioni KubernetesPodOperator:

• Configurazione minima: imposta solo i parametri richiesti.
• Configurazione del modello: utilizza i parametri che puoi modellare con Jinja.

• Configurazione delle variabili secret: passa un oggetto Secret di Kubernetes al pod.

• In Cloud Composer 2, la configurazione dell'affinità dei pod non è disponibile. Utilizza invece gli operatori GKE per avviare i pod in un cluster diverso.

• Configurazione completa: include tutte le configurazioni.

Prima di iniziare

• In Cloud Composer 2, il cluster del tuo ambiente scala automaticamente. I carichi di lavoro aggiuntivi eseguiti utilizzando KubernetesPodOperator scalano in modo indipendente dal tuo ambiente. Il tuo ambiente non è influenzato dall'aumento della domanda di risorse, ma il cluster del tuo ambiente fa lo scale up e lo scale down a seconda della richiesta di risorse. I prezzi per i carichi di lavoro aggiuntivi eseguiti nel cluster del tuo ambiente seguono il modello di prezzi di Cloud Composer 2 e utilizzano gli SKU di computing di Cloud Composer.

• Cloud Composer 2 usa i cluster GKE con Workload Identity. Per impostazione predefinita, i pod in esecuzione in spazi dei nomi appena creati o nello spazio dei nomi composer-user-workloads non possono accedere alle risorse Google Cloud. Quando si utilizza Workload Identity, gli account di servizio Kubernetes associati agli spazi dei nomi devono essere mappati agli account di servizio Google Cloud per abilitare l'autorizzazione delle identità dei servizi per le richieste alle API di Google e ad altri servizi.

  Per questo motivo, se esegui pod nello spazio dei nomi composer-user-workloads o in uno spazio dei nomi appena creato nel cluster del tuo ambiente, le associazioni IAM appropriate tra gli account di servizio Kubernetes e Google Cloud non vengono create e questi pod non possono accedere alle risorse del tuo progetto Google Cloud.

  Se vuoi che i pod abbiano accesso alle risorse Google Cloud, utilizza lo spazio dei nomi composer-user-workloads o crea il tuo spazio dei nomi come descritto più avanti.

  Per fornire l'accesso alle risorse del progetto, segui le indicazioni in Workload Identity e configura le associazioni:

  1. Crea uno spazio dei nomi separato nel cluster del tuo ambiente.
  2. Crea un'associazione tra l'composer-user-workloads/<namespace_name>account di servizio Kubernetes e l'account di servizio del tuo ambiente.
  3. Aggiungi l'annotazione dell'account di servizio del tuo ambiente all'account di servizio Kubernetes.
  4. Quando utilizzi KubernetesPodOperator, specifica lo spazio dei nomi e l'account di servizio Kubernetes nei parametri namespace e service_account_name.

• Cloud Composer 2 utilizza i cluster GKE con Workload Identity. Il server di metadati GKE impiega alcuni secondi per iniziare ad accettare richieste su un pod appena creato. Di conseguenza, i tentativi di autenticazione utilizzando Workload Identity entro i primi secondi di vita di un pod potrebbero non riuscire. Per ulteriori informazioni su questa limitazione, consulta Restrizioni di Workload Identity.

• Cloud Composer 2 utilizza i cluster Autopilot che introducono la nozione di classi di computing:

  • Per impostazione predefinita, se non è selezionata nessuna classe, viene assunta la classe general-purpose quando crei pod utilizzando KubernetesPodOperator.

  • Ogni classe è associata a proprietà e limiti di risorse specifici. Puoi leggere informazioni in merito nella documentazione di Autopilot. Ad esempio, i pod in esecuzione nella classe general-purpose possono utilizzare fino a 110 GiB di memoria.

  Importante: al momento Cloud Composer 2 supporta solo la classe di calcolo general-purpose.

• Se viene utilizzata la versione 5.0.0 del provider Kubernetes CNCF, segui le istruzioni documentate Sezione sul provider Kubernetes CNCF.

Configurazione di KubernetesPodOperator

Per seguire questo esempio, inserisci l'intero file kubernetes_pod_operator.py nella cartella dags/ del tuo ambiente o aggiungi il codice KubernetesPodOperator pertinente a un DAG.

Le sezioni seguenti spiegano ciascuna configurazione di KubernetesPodOperator nell'esempio. Per informazioni su ogni variabile di configurazione, consulta il riferimento di 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={},
    )

Configurazione minima

Per creare un KubernetesPodOperator, sono necessari solo name, namespace del pod in cui eseguire il pod, image da utilizzare e task_id.

Quando inserisci il seguente snippet di codice in un DAG, la configurazione utilizza i valori predefiniti in /home/airflow/composer_kube_config. Non devi modificare il codice per completare l'attività pod-ex-minimum.

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",
)

Configurazione modello

Airflow supporta l'utilizzo di Jinja Templating. Devi dichiarare le variabili richieste (task_id, name, namespace e image) con l'operatore. Come mostrato nell'esempio seguente, puoi modellare tutti gli altri parametri con Jinja, inclusi cmds, arguments, env_vars e 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",
)

Senza modificare il DAG o il tuo ambiente, l'attività ex-kube-templates non va a buon fine a causa di due errori. I log mostrano che questa attività non è riuscita perché non esiste la variabile appropriata (my_value). Il secondo errore, che puoi ricevere dopo aver corretto il primo errore, indica che l'attività non riesce perché core/kube_config non è stato trovato in config.

Per correggere entrambi gli errori, segui i passaggi descritti più avanti.

Per impostare my_value con gcloud o con la UI di Airflow:

UI di Airflow

Nell'interfaccia utente di Airflow 2:

1. Vai alla UI di Airflow.

2. Nella barra degli strumenti, seleziona Amministrazione > Variabili.

3. Nella pagina Elenca variabili, fai clic su Aggiungi un nuovo record.

4. Nella pagina Aggiungi variabile, inserisci le seguenti informazioni:

   • Chiave:my_value
   • Val: example_value

5. Fai clic su Salva.

gcloud

Per Airflow 2, inserisci il comando seguente:

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

Sostituisci:

• ENVIRONMENT con il nome dell'ambiente.
• LOCATION con la regione in cui si trova l'ambiente.

Per fare riferimento a un config_file personalizzato (un file di configurazione di Kubernetes), esegui l'override dell'opzione di configurazione kube_config Airflow con una configurazione Kubernetes valida:

Sezione	Chiave	Valore
core	kube_config	/home/airflow/composer_kube_config

Attendi qualche minuto per l'aggiornamento dell'ambiente. Poi esegui di nuovo l'attività ex-kube-templates e verifica che l'attività ex-kube-templates sia riuscita.

Configurazione delle variabili secret

Un secret di Kubernetes è un oggetto che contiene dati sensibili. Puoi passare i secret ai pod Kubernetes utilizzando l'KubernetesPodOperator. I secret devono essere definiti in Kubernetes, altrimenti il pod non viene avviato.

Questo esempio mostra due modi per utilizzare i secret di Kubernetes: come variabile di ambiente e come volume montato dal pod.

Il primo secret, airflow-secrets, è impostato su una variabile di ambiente Kubernetes denominata SQL_CONN (anziché su una variabile di ambiente Airflow o Cloud Composer).

Il secondo secret, service-account, monta service-account.json, un file con un token dell'account di servizio, in /var/secrets/google.

Ecco i segreti:

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",
)

Il nome del primo secret Kubernetes è definito nella variabile secret. Questo particolare secret è denominato airflow-secrets. Viene esposta come variabile di ambiente, come dettato da deploy_type. La variabile di ambiente impostata su deploy_target è SQL_CONN. Infine, il key del secret archiviato in deploy_target è sql_alchemy_conn.

Il nome del secondo secret Kubernetes è definito nella variabile secret. Questo particolare secret è denominato service-account. Viene esposto come volume, come dettato dall'deploy_type. Il percorso del file da montare, deploy_target, è /var/secrets/google. Infine, il key del secret archiviato in deploy_target è service-account.json.

Ecco l'aspetto della configurazione dell'operatore:

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",
)

Senza apportare modifiche al DAG o al tuo ambiente, l'attività ex-kube-secrets non va a buon fine. Se esamini i log, l'attività non riesce a causa di un errore Pod took too long to start. Questo errore si verifica perché Airflow non è in grado di trovare il secret specificato nella configurazione, secret_env.

gcloud

Per impostare il secret utilizzando gcloud:

1. Ottenere informazioni sul cluster di ambiente Cloud Composer.

   1. Esegui questo comando:

      gcloud composer environments describe ENVIRONMENT \
          --location LOCATION \
          --format="value(config.gkeCluster)"
      

      Sostituisci:

      • ENVIRONMENT con il nome del tuo ambiente.
      • LOCATION con la regione in cui si trova l'ambiente Cloud Composer.

      L'output di questo comando utilizza il seguente formato: projects/<your-project-id>/locations/<location-of-composer-env>/clusters/<your-cluster-id>.

   2. Per ottenere l'ID cluster GKE, copia l'output dopo /clusters/ (termina con -gke).

2. Connettiti al cluster GKE eseguendo questo comando:

   gcloud container clusters get-credentials CLUSTER_ID \
     --project PROJECT \
     --region LOCATION
   

   Sostituisci:

   • CLUSTER_ID con l'ID cluster GKE.
   • PROJECT con l'ID del tuo progetto Google Cloud.
   • LOCATION con la regione in cui si trova l'ambiente Cloud Composer.

3. Creare secret di Kubernetes.

   Nota: questi passaggi mostrano due diversi approcci per la creazione di secret Kubernetes. L'approccio --from-literal utilizza coppie chiave/valore. L'approccio --from-file utilizza contenuti di file. Non devi usare entrambi gli approcci contemporaneamente. Se vuoi, puoi creare entrambi i secret utilizzando uno solo degli approcci.

   1. Crea un secret Kubernetes che imposti il valore di sql_alchemy_conn su test_value eseguendo questo comando:

      kubectl create secret generic airflow-secrets \
        --from-literal sql_alchemy_conn=test_value -n composer-user-workloads
      

   2. Crea un secret Kubernetes che imposta il valore di service-account.json su un percorso locale di un file di chiave dell'account di servizio denominato key.json eseguendo questo comando:

      kubectl create secret generic service-account \
        --from-file service-account.json=./key.json -n composer-user-workloads
      

4. Dopo aver impostato i secret, esegui di nuovo l'attività ex-kube-secrets nella UI di Airflow.

5. Verifica che l'attività ex-kube-secrets sia riuscita.

Configurazione completa

Questo esempio mostra tutte le variabili che puoi configurare in KubernetesPodOperator. Non è necessario modificare il codice per completare l'attività ex-all-configs.

Per maggiori dettagli su ogni variabile, consulta il riferimento di 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={},
)

Informazioni sul provider Kubernetes CNCF

GKEStartPodOperator e KubernetesPodOperator sono implementati all'interno del provider apache-airflow-providers-cncf-kubernetes.

Per le note di rilascio non superate per il provider Kubernetes CNCF, fai riferimento al sito web del provider Kubernetes CNCF.

Versione 6.0.0

Nella versione 6.0.0 del pacchetto del provider Kubernetes CNCF, la connessione kubernetes_default viene utilizzata per impostazione predefinita in KubernetesPodOperator.

Se hai specificato una connessione personalizzata nella versione 5.0.0, questa viene comunque utilizzata dall'operatore. Per tornare a utilizzare la connessione kubernetes_default, ti consigliamo di regolare i DAG di conseguenza.

Versione 5.0.0

Questa versione introduce alcune modifiche incompatibili con le versioni precedenti rispetto alla 4.4.0. I più importanti sono correlati alla connessione kubernetes_default, che non viene utilizzata nella versione 5.0.0.

Importante: sono necessarie modifiche ai DAG utilizzando KubernetesPodOperator se esegui l'upgrade da una versione di Cloud Composer che usa la 4.4.0 o una versione precedente del provider a una versione di Cloud Composer che utilizza la versione 5.0.0.

• La connessione kubernetes_default deve essere modificata. Il percorso di configurazione di Kube deve essere impostato su /home/airflow/composer_kube_config (come mostrato nella Figura 1) In alternativa, è necessario aggiungere config_file alla configurazione KubernetesPodOperator (come mostrato nel seguente esempio di codice).

Figura 1. UI di Airflow, modifica della connessione kubernetes_default (fai clic per ingrandire)

• Modifica il codice di un'attività utilizzando KubernetesPodOperator nel seguente modo:

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',
  ...
)

Per ulteriori informazioni sulla versione 5.0.0, consulta le note di rilascio del provider Kubernetes CNC.

Risoluzione dei problemi

Suggerimenti per la risoluzione degli errori dei pod

Oltre a controllare i log delle attività nella UI di Airflow, controlla anche i seguenti log:

• Output dello scheduler e dei worker di Airflow:

  1. Nella console Google Cloud, vai alla pagina Ambienti.

     Vai ad Ambienti

  2. Segui il link DAG per il tuo ambiente.

  3. Nel bucket del tuo ambiente, sali di un livello.

  4. Esamina i log nella cartella logs/<DAG_NAME>/<TASK_ID>/<EXECUTION_DATE>.

• Log dettagliati dei pod nella console Google Cloud, in Carichi di lavoro GKE. Questi log includono il file YAML di definizione dei pod, gli eventi dei pod e i dettagli dei pod.

Codici di reso diversi da zero se utilizzi anche GKEStartPodOperator

Quando utilizzi KubernetesPodOperator e GKEStartPodOperator, il codice restituito del punto di ingresso del container determina se l'attività viene considerata riuscita o meno. I codici di reso diversi da zero indicano un errore.

Un pattern comune quando si utilizzano KubernetesPodOperator e GKEStartPodOperator è l'esecuzione di uno script shell come punto di ingresso del container per raggruppare più operazioni all'interno del container.

Se stai scrivendo uno script di questo tipo, ti consigliamo di includere il comando set -e nella parte superiore dello script in modo che i comandi con errori nello script terminino lo script e propaghino l'errore all'istanza dell'attività Airflow.

Timeout pod

Il timeout predefinito per KubernetesPodOperator è 120 secondi, il che può comportare timeout prima del download di immagini più grandi. Puoi aumentare il timeout modificando il parametro startup_timeout_seconds quando crei KubernetesPodOperator.

In caso di timeout di un pod, il log specifico dell'attività è disponibile nella UI di Airflow. Ad esempio:

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

I timeout dei pod possono verificarsi anche quando l'account di servizio Cloud Composer non dispone delle autorizzazioni IAM necessarie per eseguire l'attività in questione. Per verificarlo, esamina gli errori a livello di pod utilizzando le dashboard di GKE per esaminare i log per il tuo particolare carico di lavoro oppure utilizza Cloud Logging.

Impossibile stabilire una nuova connessione

L'upgrade automatico è abilitato per impostazione predefinita nei cluster GKE. Se un pool di nodi si trova in un cluster in fase di upgrade, potresti visualizzare il seguente errore:

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

Per verificare se il cluster è in fase di upgrade, nella console Google Cloud vai alla pagina Cluster Kubernetes e cerca l'icona di caricamento accanto al nome del cluster dell'ambiente.

Passaggi successivi

• Utilizzare gli operatori GKE
• Nozioni di base di Kubernetes