Run the KubernetesPodOperator on Astro Private Cloud

​

Prerequisites

• A running Airflow Deployment on Astro Private Cloud

​

Set up the KubernetesPodOperator

​

Install the operator

pip install apache-airflow-providers-cncf-kubernetes

​

Specify parameters

from airflow.configuration import conf
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator

namespace = conf.get("kubernetes", "NAMESPACE")

KubernetesPodOperator(
    namespace=namespace,
    image="ubuntu:24.04",
    cmds=["bash", "-cx"],
    arguments=["echo", "10", "echo pwd"],
    labels={"<pod-label>": "<label-name>"},
    name="airflow-test-pod",
    is_delete_operator_pod=True,
    in_cluster=True,
    task_id="task-two",
    get_logs=True,
)

• namespace = conf.get("kubernetes", "NAMESPACE"): Every Deployment runs on its own Kubernetes namespace. Information about this namespace can be programmatically imported as long as you set this variable.
• image: This is the Docker image that the operator will use to run its defined task, commands, and arguments. The value you specify is assumed to be an image tag that’s publicly available on a public registry. To pull an image from a private registry, read Pull images from a Private Registry.
• in_cluster=True: When this value is set, your task will run within the cluster from which it’s instantiated. This ensures that the Kubernetes Pod running your task has the correct permissions within the cluster.
• is_delete_operator_pod=True: This setting ensures that once a KubernetesPodOperator task is complete, the Kubernetes Pod that ran that task is terminated. This ensures that there are no unused pods in your cluster taking up resources.

​

Add resources to your Deployment on Astro Private Cloud

Note: Your Airflow scheduler and webserver will remain necessary fixed resources that ensure the rest of your tasks can execute and that your deployment stays up and running.

ERROR - Exception when attempting to create namespace Pod.
Reason: Forbidden
"Failure","message":"pods is forbidden: User \"system:serviceaccount:astronomer-cloud-solar-orbit-4143:solar-orbit-4143-airflow-worker\" can't create pods in the namespace \"datarouter\"","reason":"Forbidden","details":{"kind":"pods"},"code":403}

Note: If you need to increase your limit range on Astro Private Cloud, contact your system admin.

​

Define resources per task

Note: If you’re using the Kubernetes Executor, note that this value is separate from the executor_config parameter. In this case, the executor_config would only define the Airflow worker that is launching your Kubernetes task.

​

Example task definition

from datetime import datetime, timedelta

from airflow import DAG
from airflow.configuration import conf
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from kubernetes.client import models as k8s

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2019, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

namespace = conf.get('kubernetes', 'NAMESPACE')

# This will detect the default namespace locally and read the
# environment namespace when deployed to Astronomer.
if namespace == 'default':
    config_file = '/usr/local/airflow/include/.kube/config'
    in_cluster = False
else:
    in_cluster = True
    config_file = None

dag = DAG("example_kubernetes_pod", schedule="@once", default_args=default_args)

# This is where you define your resource allocation.
compute_resources = k8s.V1ResourceRequirements(
    limits={"cpu": "800m", "memory": "3Gi"},
    requests={"cpu": "800m", "memory": "3Gi"}
)

with dag:
    KubernetesPodOperator(
        namespace=namespace,
        image="hello-world",
        labels={"foo": "bar"},
        name="airflow-test-pod",
        task_id="task-one",
        in_cluster=in_cluster,  # if set to true, will look in the cluster, if false, looks for file
        cluster_context="docker-for-desktop",  # is ignored when in_cluster is set to True
        config_file=config_file,
        container_resources=compute_resources,
        is_delete_operator_pod=True,
        get_logs=True,
    )

from kubernetes.client import models as k8s

compute_resources = k8s.V1ResourceRequirements(
    limits={"cpu": "800m", "memory": "3Gi"},
    requests={"cpu": "800m", "memory": "3Gi"}
)

​

Pull images from a private registry

1. Retrieve a config.json file that contains your Docker credentials by following the Docker documentation. The generated file should look something like this:
   {
       "auths": {
           "https://index.docker.io/v1/": {
               "auth": "c3R...zE2"
           }
       }
   }
   
2. Follow the Kubernetes documentation to create a secret based on your credentials.
3. In your Dag code, import models from kubernetes.client and specify image_pull_secrets with your Kubernetes secret. After configuring this value, you can pull an image as you would from a public registry like in the following example.
   from kubernetes.client import models as k8s
   
   KubernetesPodOperator(
       namespace=namespace,
       image_pull_secrets=[k8s.V1LocalObjectReference("<your-secret-name>")],
       image="<your-docker-image>",
       cmds=["<commands-for-image>"],
       arguments=["<arguments-for-image>"],
       labels={"<pod-label>": "<label-name>"},
       name="<pod-name>",
       is_delete_operator_pod=True,
       in_cluster=True,
       task_id="<task-name>",
       get_logs=True,
   )
   

​

Local testing