Kubernetes Executor -> is forbidden: exceeded quota

[devscheffer]

Apache Airflow version

Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?

2.7.1

What happened?

I’m using the Kubernetes executor along with the Kubernetes pod operator. My DAG has many tasks running in parallel.

However, when resources are not available for a task, I encounter an error. The error message indicates that the pod has exceeded its resource quota.

kubernetes.client.exceptions.ApiException: (403)
Reason: Forbidden
HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"pods \"xxx\" is forbidden: exceeded quota: airflow-dags-resource-quota, requested: limits.cpu=550m, used: limits.cpu=3300m, limited: limits.cpu=3500m","reason":"Forbidden","details":{"name":"xxx","kind":"pods"},"code":403}

What you think should happen instead?

I was under the impression that when using the Kubernetes executor, tasks would queue until sufficient resources become available, rather than failing. I’m wondering if I’ve overlooked some configuration settings. For most configurations, I’m using the default values provided by the Airflow Helm chart. Can anyone help me understand this?

How to reproduce

use kubernetes and the airflow helmchart, set some resource quota for the namespace and try to execute a lot of tasks in parallel that would normally consume more resource than it is available

Operating System

ubuntu

Versions of Apache Airflow Providers

2.7.1

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[csp33]

hi @devscheffer , I believe this issue is related to your k8s cluster rather than to Airflow.
It seems that you have a quota (airflow-dags-resource-quota) defined for the namespace you're trying to run the task in.

You can check it with kubectl describe quota airflow-dags-resource-quota -n <your namespace>

https://kubernetes.io/docs/concepts/policy/resource-quotas/

[devscheffer]

if we could change the kubernetes pod operator to instead of being a kind: pod be a kind: deployment
couldn't solve the problem?
https://stackoverflow.com/questions/69902490/kubernetes-resource-quota-have-non-schedulable-pod-staying-in-pending-state

because then the deployment would manage the deploy of the pod and we wouldn't have a problem of forbidden request

[potiuk]

No idea, not an expert in that. Likely there are many more consequences to do it. If you want to propose some changes there - likely analysing it and creating PR where you'd discuss the consequences etc. is one of the ways of doing it.

[devscheffer]

My propose would be to create a new operator KubernetesDeploymentOperator, that after the pods are completed the deployment would be deleted.
that is the simplest solution I could imagine I just don't know how can I make the PR for that

from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from kubernetes import client, config
import time

class KubernetesDeploymentOperator(KubernetesPodOperator):
    def execute(self, context):
        config.load_kube_config()
        k8s_client = client.AppsV1Api()

        # Create a Deployment spec based on the existing Pod spec
        deployment_spec = {
            "apiVersion": "apps/v1",
            "kind": "Deployment",
            "metadata": {
                "name": self.name,
            },
            "spec": {
                "replicas": 1,
                "selector": {
                    "matchLabels": {
                        "app": self.name
                    }
                },
                "template": {
                    "metadata": {
                        "labels": {
                            "app": self.name
                        }
                    },
                    "spec": self.full_pod_spec.spec
                }
            }
        }

        # Create and start the Deployment
        resp = k8s_client.create_namespaced_deployment(
            body=deployment_spec,
            namespace=self.namespace)
        self.log.info(f"Deployment created. status='{resp.status}'")

        # Monitor the status of the Pods
        while True:
            resp = k8s_client.read_namespaced_deployment(
                name=self.name,
                namespace=self.namespace)
            if resp.status.replicas == resp.status.ready_replicas:
                break
            time.sleep(10)  # Wait before checking the status again

        # Delete the Deployment
        resp = k8s_client.delete_namespaced_deployment(
            name=self.name,
            namespace=self.namespace)
        self.log.info(f"Deployment deleted. status='{resp.status}'")

Is that solution makes sense to you @potiuk ?

PS: Thank you for your time in this discussion

[potiuk]

Again - not sure about consequences, proposing it in PR is a good start though. Or if you do not know how starting a discussion specific for that proposal - possibly even starting discussion in devlist https://airflow.apache.org/community/ as KPO is one of the most used operator and posssibly there are other consequences of what you propose. I think there were also other discussions and issues about it and you can search through the issues and discussions for similar things.

[potiuk]

Correct. If your task exceeds quota, things will fail. This is normal and expected. Converting to discussion if needed.

[akarneliuk]

@potiuk i actually think it is a bug and what @devscheffer observed and suggested initially is a right expectation. I will provide some details.

I am running airflow with KPO in resource constraint environment. With airflow version of 2.6.3 i didn't have any problems with pod scheduling being automatically delayed untill resources are free. But after upgrade to a newer version (2.9), i have startes seeing exactly the same error. I was not able to see in release notes relecant change, but it must be some breaking change in KPO behavior.

[potiuk]

Maybe it's a change in K8S version behaviour. If you think you can gather all the differences and show paralllel before<>after - you might want to open a new issue. Hopefully somoene will help

[Salfiii]

Hi @devscheffer ,
we observed the same behaviour change when upgrading from airflow 2.7 to 2.10.

Did you find or propose any solution so far?

This makes it exeptionally harder to run workloads via k8s exector and we had to heavily limit the number of pararell possible to tasks to mitigate this because it´s not clear from the outside how big the requirements of one task are.

Thanks in advance.

Best regards