Affinities and tolerations sidecar override in DAG

[marekargalas]

Hi!

I use kubernetesPodTemplate in values.yaml to set affinity and tolerations for node. With this all DAGs/KubernetesPodOperator use this setting for scheduling k8s pods (spinning/base container/sidecar and actual pod for job).
However I would like to use different affinity and tolerations per DAG - which could be done by V1Affinity and V1Toleration but it's not propagated to base container and KubernetesPodOperator is using two different kubernetes nodes. Any idea how to set it up for base container in the DAGs? I found init_containers but I can adjust sidecar as well.

Thanks!

[thesuperzapper]

@marekargalas I think you may be confusing the KuberentesPodOperator, and KubernetesExecutor.

Here is a brief overview of the difference:

• KuberentesPodOperator: is a type of operator/task, which tells airflow to submit a Pod to a Kubernetes cluster as a step in a DAG
• KubernetesExecutor: is a type of airflow executor, used by airflow to run ALL tasks (including submitting the Pods defined by KuberentesPodOperator), and is an alternative to the CeleryExecutor

The way to override tolerations (or any other Pod attribute) is different:

• KuberentesPodOperator: tolerations are inherently per-task and must be specified for each KuberentesPodOperator task
• KubernetesExecutor: if you want to override tolerations per task, you can use the pod_override field which is available on every operator/task

If you plan to use KuberentesPodOperator, consider using CeleryExecutor (so that an executor Pod is not created just to submit operator Pod), alternatively, use CeleryKubernetesExecutor so you can use CeleryExecutor for some tasks, and KubernetesExecutor for others.

[marekargalas]

@thesuperzapper Hey, thanks for answer!

Reason using KubernetesExecutor is that I don't want to have live workers all the day and just spin-up k8s pods when needed. Afaik CeleryExecutor is waiting for jobs to be passed thus I can't do scale-to-zero strategy. Or can I do it with CeleryExecutor somehow?

Regarding tolerations&affinities - indeed I set them up at KuberentesPodOperator level. However this setting is not passed to executor pods and I might use different tolerations per DAG. Example:

TOLERATIONS = [
    k8s.V1Toleration(key="dedicated", operator="Equal", value="node_2", effect="NoSchedule")
]

KubernetesPodOperator(
        task_id="task"
        name="name",
        namespace="default",
        image="ecr:xyz",
        tolerations=TOLERATIONS,
    )

With this one, executor pod is created with default toleration - e.g node_1 from values.yaml file and then actual container using toleration from TOLERATIONS.

[thesuperzapper]

Reason using KubernetesExecutor is that I don't want to have live workers all the day and just spin-up k8s pods when needed. Afaik CeleryExecutor is waiting for jobs to be passed thus I can't do scale-to-zero strategy. Or can I do it with CeleryExecutor somehow?

@marekargalas I am actually actively working on a task-aware CeleryExecutor autoscaler, that will allow scaling to/from zero replicas, and even allow multiple "queues" of workers that may themselves have different tolerations/resource-requests. (Note, you can select these "queue" per task with the queue argument of all airflow operators).

The reason why I was not able to make autoscaling celery work properly until now (Kubernetes 1.22), was that there was no way to prevent the removal of active workers (K8S did not allow operators to influence which pods were removed when downscaling a Deployment until controller.kubernetes.io/pod-deletion-cost was added).

You can see more information in issue #339 (Note, I am keeping the actual beta code slightly hidden as I don't want people using the new autoscaler until its fully ready and safe to use, which I hope will be within this month!)

---

Regarding tolerations&affinities - indeed I set them up at KuberentesPodOperator level. However this setting is not passed to executor pods and I might use different tolerations per DAG. Example:

TOLERATIONS = [
    k8s.V1Toleration(key="dedicated", operator="Equal", value="node_2", effect="NoSchedule")
]

KubernetesPodOperator(
        task_id="task"
        name="name",
        namespace="default",
        image="ecr:xyz",
        tolerations=TOLERATIONS,
    )

With this one, executor pod is created with default toleration - e.g node_1 from values.yaml file and then actual container using toleration from TOLERATIONS.

This is exactly as you would expect, because when using KubernetesExecutor with a KuberentesPodOperator task, it will execute as follows

1. Airflow creates a KubernetesExecutor Pod that will run the "kubectl apply" for your the KuberentesPodOperator
   • This Pod will use the tollerations from airflow.kubernetesPodTemplate
2. The "kubectl apply" runs your Pod as defined by your KubernetesPodOperator
   • This Pod will use the tolerations from the KubernetesPodOperator definition

[marekargalas]

Thanks for more info! News around CeleryExecutor sounds cool!

So just to conclude the toleration challenge here - there is no way to change KubernetesExecutor tolerations (or kubernetesPodTemplate respectively) from KuberentesPodOperator or DAG right?
I see there is @task(executor_config=executor_config_volume_mount) but with KuberentesPodOperator all tasks are basically hidden in docker image and not living in DAG.