You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "mrrsm (via GitHub)" <gi...@apache.org> on 2026/03/19 18:37:27 UTC
[PR] feat: Add resource quota awareness to KubernetesPodOperator [airflow]
mrrsm opened a new pull request, #63946:
URL: https://github.com/apache/airflow/pull/63946
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
This adds, in a non-breaking way, additional parameters for the KubernesPodOperator which allow the lookup of a resource quota in the namespace that is being deployed to to ensure there is proper resources available for the deployment of the pod. The user can choose what to do if the quota would be exceeded. The options if the quota is exceeded are to queue (default), fail, or ignore and try anyways. The default retry is 60 seconds but is configurable via another param.
I was able to successfully test this in a breeze setup connected to a kind cluster.
I will open the PR to update the decorator after any changes to the params are finalized in this PR.
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [x] Yes (please specify the tool below)
Generated-by: [Claude Sonnet 4.5] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add resource quota awareness to KubernetesPodOperator [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #63946:
URL: https://github.com/apache/airflow/pull/63946#discussion_r2962141989
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -629,6 +654,29 @@ def get_or_create_pod(self, pod_request_obj: k8s.V1Pod, context: Context) -> k8s
self.log.info("Deleted pod to handle rerun and create new pod!")
self.log.debug("Starting pod:\n%s", yaml.safe_dump(pod_request_obj.to_dict()))
+
+ # Check resource quotas before attempting to create the pod
+ if self.check_resource_quotas and self.on_quota_exceeded != "ignore":
+ try:
+ check_pod_quota_compliance(self.client, pod_request_obj, pod_request_obj.metadata.namespace)
+ except PodResourceQuotaExceededException as e:
+ if self.on_quota_exceeded == "queue":
+ self.log.warning(
+ "Pod creation would exceed resource quota. Task will be rescheduled to retry later. %s",
+ str(e),
+ )
+ reschedule_time = datetime.datetime.now(datetime.timezone.utc) + timedelta(
+ seconds=self.quota_check_interval
+ )
+ raise AirflowRescheduleException(reschedule_time)
+ if self.on_quota_exceeded == "fail":
+ self.log.error(
+ "Pod creation blocked due to resource quota violation. "
+ "Set on_quota_exceeded='queue' to retry or 'ignore' to skip this check."
+ )
+ raise
Review Comment:
The quota issue is not so trivial problem and I think this is somewhat over simplifying it.
I find this also somewhat confusing as reschedule is something reserved for scheduler and this is not about that.
There is also the question of how K8s executor should act on quota issues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add resource quota awareness to KubernetesPodOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #63946:
URL: https://github.com/apache/airflow/pull/63946#issuecomment-4296398037
@mrrsm This draft PR has been inactive for 33 days since the last triage comment and no response from the author. Closing to keep the queue clean.
You are welcome to reopen this PR when you resume work, or to open a new one addressing the issues previously raised. There is no rush — take your time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add resource quota awareness to KubernetesPodOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #63946:
URL: https://github.com/apache/airflow/pull/63946#issuecomment-4096391596
@mrrsm This PR has been converted to **draft** because it does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **mypy (type checking)**: Failing: CI image checks / MyPy checks (mypy-providers). Run `prek --stage manual mypy-providers --all-files` locally to reproduce. You need `breeze ci-image build --python 3.10` for Docker-based mypy. See [mypy (type checking) docs](https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#mypy-checks).
- :x: **Provider tests**: Failing: provider distributions tests / Compat 3.0.6:P3.10:. Run provider tests with `breeze run pytest <provider-test-path> -xvs`. See [Provider tests docs](https://github.com/apache/airflow/blob/main/contributing-docs/12_provider_distributions.rst).
- :x: **Other failing CI checks**: Failing: CI image checks / Build documentation (--spellcheck-only). Run `prek run --from-ref main` locally to reproduce. See [static checks docs](https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst).
- :warning: **Unresolved review comments**: This PR has 1 unresolved review thread from maintainers: **@eladkal** (MEMBER): 1 unresolved thread. Please review and resolve all inline review comments before requesting another review. You can resolve a conversation by clicking 'Resolve conversation' on each thread after addressing the feedback. See [pull request guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst).
**What to do next:**
- The comment informs you what you need to do.
- Fix each issue, then mark the PR as "Ready for review" in the GitHub UI - but only after making sure that all the issues are fixed.
- There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates.
- Maintainers will then proceed with a normal review.
Converting a PR to draft is **not** a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add resource quota awareness to KubernetesPodOperator [airflow]
Posted by "Nataneljpwd (via GitHub)" <gi...@apache.org>.
Nataneljpwd commented on code in PR #63946:
URL: https://github.com/apache/airflow/pull/63946#discussion_r2962410717
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -254,6 +265,14 @@ class KubernetesPodOperator(BaseOperator):
:param log_formatter: custom log formatter function that takes two string arguments:
the first string is the container_name and the second string is the message_to_log.
The function should return a formatted string. If None, the default formatting will be used.
+ :param check_resource_quotas: if True, check namespace resource quotas before creating the pod.
+ Default to False.
+ :param on_quota_exceeded: action to take when pod would exceed resource quota. Options:
+ "queue" (default) - reschedule the task to try again later,
+ "fail" - fail the task immediately with an exception,
Review Comment:
As of now it will happen either way no?
What is done differently here? To me it seems like the exact same behavior
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -254,6 +265,14 @@ class KubernetesPodOperator(BaseOperator):
:param log_formatter: custom log formatter function that takes two string arguments:
the first string is the container_name and the second string is the message_to_log.
The function should return a formatted string. If None, the default formatting will be used.
+ :param check_resource_quotas: if True, check namespace resource quotas before creating the pod.
+ Default to False.
+ :param on_quota_exceeded: action to take when pod would exceed resource quota. Options:
+ "queue" (default) - reschedule the task to try again later,
+ "fail" - fail the task immediately with an exception,
+ "ignore" - proceed with pod creation anyway (same as check_resource_quotas=False).
Review Comment:
Is this option needed if we already have the option to have the same behavior?
It can simplify the code and probably even remove a few tests
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -629,6 +654,29 @@ def get_or_create_pod(self, pod_request_obj: k8s.V1Pod, context: Context) -> k8s
self.log.info("Deleted pod to handle rerun and create new pod!")
self.log.debug("Starting pod:\n%s", yaml.safe_dump(pod_request_obj.to_dict()))
+
+ # Check resource quotas before attempting to create the pod
+ if self.check_resource_quotas and self.on_quota_exceeded != "ignore":
+ try:
+ check_pod_quota_compliance(self.client, pod_request_obj, pod_request_obj.metadata.namespace)
+ except PodResourceQuotaExceededException as e:
+ if self.on_quota_exceeded == "queue":
+ self.log.warning(
+ "Pod creation would exceed resource quota. Task will be rescheduled to retry later. %s",
+ str(e),
+ )
+ reschedule_time = datetime.datetime.now(datetime.timezone.utc) + timedelta(
+ seconds=self.quota_check_interval
+ )
+ raise AirflowRescheduleException(reschedule_time)
Review Comment:
This is reserved for the scheduler usually, maybe instead it is better to wait and retry later?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/resource_quota.py:
##########
@@ -0,0 +1,250 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for checking Kubernetes resource quotas."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from kubernetes.client.exceptions import ApiException
+
+from airflow.providers.common.compat.sdk import AirflowException
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api
+ from kubernetes.client.models import V1Pod
+
+logger = logging.getLogger(__name__)
+
+
+class PodResourceQuotaExceededException(AirflowException):
+ """Raised when creating a pod would exceed the namespace resource quota."""
+
+
+def parse_resource_quantity(quantity: str | None) -> float:
+ """
+ Parse Kubernetes resource quantity string to a numeric value.
+
+ Supports suffixes like Ki, Mi, Gi, Ti, Pi, Ei for binary units
+ and k, M, G, T, P, E for decimal units, as well as millicpu 'm' suffix.
+
+ :param quantity: Kubernetes resource quantity string (e.g., "100m", "1Gi", "500Mi")
+ :return: Numeric value in base units (cores for CPU, bytes for memory)
+ """
+ if not quantity:
+ return 0.0
+
+ quantity = str(quantity).strip()
+ if not quantity:
+ return 0.0
+
+ # Binary suffixes (base 1024)
+ binary_suffixes = {
+ "Ki": 1024,
+ "Mi": 1024**2,
+ "Gi": 1024**3,
+ "Ti": 1024**4,
+ "Pi": 1024**5,
+ "Ei": 1024**6,
+ }
+
+ # Decimal suffixes (base 1000)
+ decimal_suffixes = {
+ "k": 1000,
+ "M": 1000**2,
+ "G": 1000**3,
+ "T": 1000**4,
+ "P": 1000**5,
+ "E": 1000**6,
+ }
+
+ # Check for millicpu (m suffix)
+ if quantity.endswith("m"):
+ try:
+ return float(quantity[:-1]) / 1000
+ except ValueError:
+ return 0.0
+
+ # Check for binary suffixes
+ for suffix, multiplier in binary_suffixes.items():
+ if quantity.endswith(suffix):
+ try:
+ return float(quantity[: -len(suffix)]) * multiplier
+ except ValueError:
+ return 0.0
+
+ # Check for decimal suffixes
+ for suffix, multiplier in decimal_suffixes.items():
+ if quantity.endswith(suffix):
+ try:
+ return float(quantity[: -len(suffix)]) * multiplier
+ except ValueError:
+ return 0.0
Review Comment:
This check is redundant and already implemented
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -79,9 +80,19 @@
PodNotFoundException,
PodPhase,
)
+from airflow.providers.cncf.kubernetes.utils.resource_quota import (
+ PodResourceQuotaExceededException,
+ check_pod_quota_compliance,
+)
from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
from airflow.providers.common.compat.sdk import XCOM_RETURN_KEY, AirflowSkipException, TaskDeferred
+# Import AirflowRescheduleException with version compatibility
+try:
+ from airflow.sdk.exceptions import AirflowRescheduleException
+except ImportError:
+ from airflow.exceptions import AirflowRescheduleException # type: ignore[no-redef]
Review Comment:
Why not use the common compat provider here?
It does this for you
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/resource_quota.py:
##########
@@ -0,0 +1,250 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for checking Kubernetes resource quotas."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from kubernetes.client.exceptions import ApiException
+
+from airflow.providers.common.compat.sdk import AirflowException
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api
+ from kubernetes.client.models import V1Pod
+
+logger = logging.getLogger(__name__)
+
+
+class PodResourceQuotaExceededException(AirflowException):
+ """Raised when creating a pod would exceed the namespace resource quota."""
+
+
+def parse_resource_quantity(quantity: str | None) -> float:
+ """
+ Parse Kubernetes resource quantity string to a numeric value.
+
+ Supports suffixes like Ki, Mi, Gi, Ti, Pi, Ei for binary units
+ and k, M, G, T, P, E for decimal units, as well as millicpu 'm' suffix.
+
+ :param quantity: Kubernetes resource quantity string (e.g., "100m", "1Gi", "500Mi")
+ :return: Numeric value in base units (cores for CPU, bytes for memory)
+ """
+ if not quantity:
+ return 0.0
+
+ quantity = str(quantity).strip()
+ if not quantity:
+ return 0.0
+
+ # Binary suffixes (base 1024)
+ binary_suffixes = {
+ "Ki": 1024,
+ "Mi": 1024**2,
+ "Gi": 1024**3,
+ "Ti": 1024**4,
+ "Pi": 1024**5,
+ "Ei": 1024**6,
+ }
+
+ # Decimal suffixes (base 1000)
+ decimal_suffixes = {
+ "k": 1000,
+ "M": 1000**2,
+ "G": 1000**3,
+ "T": 1000**4,
+ "P": 1000**5,
+ "E": 1000**6,
+ }
Review Comment:
K8s and python has this functionality built in using format_quantity and parse_quantity
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/resource_quota.py:
##########
@@ -0,0 +1,250 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for checking Kubernetes resource quotas."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from kubernetes.client.exceptions import ApiException
+
+from airflow.providers.common.compat.sdk import AirflowException
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api
+ from kubernetes.client.models import V1Pod
+
+logger = logging.getLogger(__name__)
+
+
+class PodResourceQuotaExceededException(AirflowException):
+ """Raised when creating a pod would exceed the namespace resource quota."""
+
+
+def parse_resource_quantity(quantity: str | None) -> float:
+ """
+ Parse Kubernetes resource quantity string to a numeric value.
+
+ Supports suffixes like Ki, Mi, Gi, Ti, Pi, Ei for binary units
+ and k, M, G, T, P, E for decimal units, as well as millicpu 'm' suffix.
+
+ :param quantity: Kubernetes resource quantity string (e.g., "100m", "1Gi", "500Mi")
+ :return: Numeric value in base units (cores for CPU, bytes for memory)
+ """
+ if not quantity:
+ return 0.0
+
+ quantity = str(quantity).strip()
+ if not quantity:
+ return 0.0
Review Comment:
Why is it checked twice? Why not the same if?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/resource_quota.py:
##########
@@ -0,0 +1,250 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for checking Kubernetes resource quotas."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from kubernetes.client.exceptions import ApiException
+
+from airflow.providers.common.compat.sdk import AirflowException
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api
+ from kubernetes.client.models import V1Pod
+
+logger = logging.getLogger(__name__)
+
+
+class PodResourceQuotaExceededException(AirflowException):
+ """Raised when creating a pod would exceed the namespace resource quota."""
+
+
+def parse_resource_quantity(quantity: str | None) -> float:
+ """
+ Parse Kubernetes resource quantity string to a numeric value.
+
+ Supports suffixes like Ki, Mi, Gi, Ti, Pi, Ei for binary units
+ and k, M, G, T, P, E for decimal units, as well as millicpu 'm' suffix.
+
+ :param quantity: Kubernetes resource quantity string (e.g., "100m", "1Gi", "500Mi")
+ :return: Numeric value in base units (cores for CPU, bytes for memory)
+ """
+ if not quantity:
+ return 0.0
+
+ quantity = str(quantity).strip()
+ if not quantity:
+ return 0.0
+
+ # Binary suffixes (base 1024)
+ binary_suffixes = {
+ "Ki": 1024,
+ "Mi": 1024**2,
+ "Gi": 1024**3,
+ "Ti": 1024**4,
+ "Pi": 1024**5,
+ "Ei": 1024**6,
+ }
+
+ # Decimal suffixes (base 1000)
+ decimal_suffixes = {
+ "k": 1000,
+ "M": 1000**2,
+ "G": 1000**3,
+ "T": 1000**4,
+ "P": 1000**5,
+ "E": 1000**6,
+ }
+
+ # Check for millicpu (m suffix)
+ if quantity.endswith("m"):
+ try:
+ return float(quantity[:-1]) / 1000
+ except ValueError:
+ return 0.0
+
+ # Check for binary suffixes
+ for suffix, multiplier in binary_suffixes.items():
+ if quantity.endswith(suffix):
+ try:
+ return float(quantity[: -len(suffix)]) * multiplier
+ except ValueError:
+ return 0.0
+
+ # Check for decimal suffixes
+ for suffix, multiplier in decimal_suffixes.items():
+ if quantity.endswith(suffix):
+ try:
+ return float(quantity[: -len(suffix)]) * multiplier
+ except ValueError:
+ return 0.0
+
+ # No suffix, just a number
+ try:
+ return float(quantity)
+ except ValueError:
+ return 0.0
+
+
+def get_pod_resource_requests(pod: V1Pod) -> dict[str, float]:
+ """
+ Extract total resource requests from a pod specification.
+
+ :param pod: V1Pod object
+ :return: Dictionary with 'cpu' and 'memory' keys containing total requests
+ """
+ total_cpu = 0.0
+ total_memory = 0.0
+
+ if not pod.spec or not pod.spec.containers:
+ return {"cpu": total_cpu, "memory": total_memory}
+
+ for container in pod.spec.containers:
+ if container.resources and container.resources.requests:
+ cpu_request = container.resources.requests.get("cpu")
+ memory_request = container.resources.requests.get("memory")
+
+ if cpu_request:
+ total_cpu += parse_resource_quantity(cpu_request)
+ if memory_request:
+ total_memory += parse_resource_quantity(memory_request)
+
+ # Also check init containers
+ if pod.spec.init_containers:
+ for container in pod.spec.init_containers:
+ if container.resources and container.resources.requests:
+ cpu_request = container.resources.requests.get("cpu")
+ memory_request = container.resources.requests.get("memory")
+
+ if cpu_request:
+ total_cpu += parse_resource_quantity(cpu_request)
+ if memory_request:
+ total_memory += parse_resource_quantity(memory_request)
+
+ return {"cpu": total_cpu, "memory": total_memory}
+
+
+def get_namespace_quota_status(
+ client: CoreV1Api, namespace: str
+) -> tuple[dict[str, float], dict[str, float]] | None:
+ """
+ Get the current quota usage and limits for a namespace.
+
+ :param client: Kubernetes CoreV1Api client
+ :param namespace: Namespace name
+ :return: Tuple of (used_resources, hard_limits) dictionaries or None if no quota
+ """
+ try:
+ quota_list = client.list_namespaced_resource_quota(namespace=namespace)
+ except ApiException as e:
+ if e.status == 403:
+ logger.warning(
+ "Insufficient permissions to check resource quotas in namespace %s. "
+ "Skipping quota validation.",
+ namespace,
+ )
Review Comment:
Maybe it is better to fail here, wdyt?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/resource_quota.py:
##########
@@ -0,0 +1,250 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for checking Kubernetes resource quotas."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from kubernetes.client.exceptions import ApiException
+
+from airflow.providers.common.compat.sdk import AirflowException
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api
+ from kubernetes.client.models import V1Pod
+
+logger = logging.getLogger(__name__)
+
+
+class PodResourceQuotaExceededException(AirflowException):
+ """Raised when creating a pod would exceed the namespace resource quota."""
+
+
+def parse_resource_quantity(quantity: str | None) -> float:
+ """
+ Parse Kubernetes resource quantity string to a numeric value.
+
+ Supports suffixes like Ki, Mi, Gi, Ti, Pi, Ei for binary units
+ and k, M, G, T, P, E for decimal units, as well as millicpu 'm' suffix.
+
+ :param quantity: Kubernetes resource quantity string (e.g., "100m", "1Gi", "500Mi")
+ :return: Numeric value in base units (cores for CPU, bytes for memory)
+ """
+ if not quantity:
+ return 0.0
+
+ quantity = str(quantity).strip()
+ if not quantity:
+ return 0.0
+
+ # Binary suffixes (base 1024)
+ binary_suffixes = {
+ "Ki": 1024,
+ "Mi": 1024**2,
+ "Gi": 1024**3,
+ "Ti": 1024**4,
+ "Pi": 1024**5,
+ "Ei": 1024**6,
+ }
+
+ # Decimal suffixes (base 1000)
+ decimal_suffixes = {
+ "k": 1000,
+ "M": 1000**2,
+ "G": 1000**3,
+ "T": 1000**4,
+ "P": 1000**5,
+ "E": 1000**6,
+ }
+
+ # Check for millicpu (m suffix)
+ if quantity.endswith("m"):
+ try:
+ return float(quantity[:-1]) / 1000
+ except ValueError:
+ return 0.0
+
+ # Check for binary suffixes
+ for suffix, multiplier in binary_suffixes.items():
+ if quantity.endswith(suffix):
+ try:
+ return float(quantity[: -len(suffix)]) * multiplier
+ except ValueError:
+ return 0.0
+
+ # Check for decimal suffixes
+ for suffix, multiplier in decimal_suffixes.items():
+ if quantity.endswith(suffix):
+ try:
+ return float(quantity[: -len(suffix)]) * multiplier
+ except ValueError:
+ return 0.0
+
+ # No suffix, just a number
+ try:
+ return float(quantity)
+ except ValueError:
+ return 0.0
+
+
+def get_pod_resource_requests(pod: V1Pod) -> dict[str, float]:
+ """
+ Extract total resource requests from a pod specification.
+
+ :param pod: V1Pod object
+ :return: Dictionary with 'cpu' and 'memory' keys containing total requests
+ """
+ total_cpu = 0.0
+ total_memory = 0.0
+
+ if not pod.spec or not pod.spec.containers:
+ return {"cpu": total_cpu, "memory": total_memory}
+
+ for container in pod.spec.containers:
+ if container.resources and container.resources.requests:
+ cpu_request = container.resources.requests.get("cpu")
+ memory_request = container.resources.requests.get("memory")
+
+ if cpu_request:
+ total_cpu += parse_resource_quantity(cpu_request)
+ if memory_request:
+ total_memory += parse_resource_quantity(memory_request)
+
+ # Also check init containers
+ if pod.spec.init_containers:
+ for container in pod.spec.init_containers:
+ if container.resources and container.resources.requests:
+ cpu_request = container.resources.requests.get("cpu")
+ memory_request = container.resources.requests.get("memory")
+
+ if cpu_request:
+ total_cpu += parse_resource_quantity(cpu_request)
+ if memory_request:
+ total_memory += parse_resource_quantity(memory_request)
+
+ return {"cpu": total_cpu, "memory": total_memory}
+
+
+def get_namespace_quota_status(
+ client: CoreV1Api, namespace: str
+) -> tuple[dict[str, float], dict[str, float]] | None:
+ """
+ Get the current quota usage and limits for a namespace.
+
+ :param client: Kubernetes CoreV1Api client
+ :param namespace: Namespace name
+ :return: Tuple of (used_resources, hard_limits) dictionaries or None if no quota
+ """
+ try:
+ quota_list = client.list_namespaced_resource_quota(namespace=namespace)
+ except ApiException as e:
+ if e.status == 403:
+ logger.warning(
+ "Insufficient permissions to check resource quotas in namespace %s. "
+ "Skipping quota validation.",
+ namespace,
+ )
+ return None
+ if e.status == 404:
+ logger.debug("Namespace %s not found for quota check", namespace)
Review Comment:
Same here
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -254,6 +265,14 @@ class KubernetesPodOperator(BaseOperator):
:param log_formatter: custom log formatter function that takes two string arguments:
the first string is the container_name and the second string is the message_to_log.
The function should return a formatted string. If None, the default formatting will be used.
+ :param check_resource_quotas: if True, check namespace resource quotas before creating the pod.
+ Default to False.
+ :param on_quota_exceeded: action to take when pod would exceed resource quota. Options:
+ "queue" (default) - reschedule the task to try again later,
+ "fail" - fail the task immediately with an exception,
+ "ignore" - proceed with pod creation anyway (same as check_resource_quotas=False).
Review Comment:
If both the above and this comment is true, the code can be vastly simplified as you can remove all that is not the queue mode behavior
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/utils/resource_quota.py:
##########
@@ -0,0 +1,250 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Utilities for checking Kubernetes resource quotas."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from kubernetes.client.exceptions import ApiException
+
+from airflow.providers.common.compat.sdk import AirflowException
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api
+ from kubernetes.client.models import V1Pod
+
+logger = logging.getLogger(__name__)
+
+
+class PodResourceQuotaExceededException(AirflowException):
+ """Raised when creating a pod would exceed the namespace resource quota."""
+
+
+def parse_resource_quantity(quantity: str | None) -> float:
+ """
+ Parse Kubernetes resource quantity string to a numeric value.
+
+ Supports suffixes like Ki, Mi, Gi, Ti, Pi, Ei for binary units
+ and k, M, G, T, P, E for decimal units, as well as millicpu 'm' suffix.
+
+ :param quantity: Kubernetes resource quantity string (e.g., "100m", "1Gi", "500Mi")
+ :return: Numeric value in base units (cores for CPU, bytes for memory)
+ """
+ if not quantity:
+ return 0.0
+
+ quantity = str(quantity).strip()
+ if not quantity:
+ return 0.0
+
+ # Binary suffixes (base 1024)
+ binary_suffixes = {
+ "Ki": 1024,
+ "Mi": 1024**2,
+ "Gi": 1024**3,
+ "Ti": 1024**4,
+ "Pi": 1024**5,
+ "Ei": 1024**6,
+ }
+
+ # Decimal suffixes (base 1000)
+ decimal_suffixes = {
+ "k": 1000,
+ "M": 1000**2,
+ "G": 1000**3,
+ "T": 1000**4,
+ "P": 1000**5,
+ "E": 1000**6,
+ }
+
+ # Check for millicpu (m suffix)
+ if quantity.endswith("m"):
+ try:
+ return float(quantity[:-1]) / 1000
+ except ValueError:
+ return 0.0
+
+ # Check for binary suffixes
+ for suffix, multiplier in binary_suffixes.items():
+ if quantity.endswith(suffix):
+ try:
+ return float(quantity[: -len(suffix)]) * multiplier
+ except ValueError:
+ return 0.0
+
+ # Check for decimal suffixes
+ for suffix, multiplier in decimal_suffixes.items():
+ if quantity.endswith(suffix):
+ try:
+ return float(quantity[: -len(suffix)]) * multiplier
+ except ValueError:
+ return 0.0
+
+ # No suffix, just a number
+ try:
+ return float(quantity)
+ except ValueError:
+ return 0.0
+
+
+def get_pod_resource_requests(pod: V1Pod) -> dict[str, float]:
+ """
+ Extract total resource requests from a pod specification.
+
+ :param pod: V1Pod object
+ :return: Dictionary with 'cpu' and 'memory' keys containing total requests
+ """
+ total_cpu = 0.0
+ total_memory = 0.0
+
+ if not pod.spec or not pod.spec.containers:
+ return {"cpu": total_cpu, "memory": total_memory}
+
+ for container in pod.spec.containers:
+ if container.resources and container.resources.requests:
+ cpu_request = container.resources.requests.get("cpu")
+ memory_request = container.resources.requests.get("memory")
+
+ if cpu_request:
+ total_cpu += parse_resource_quantity(cpu_request)
+ if memory_request:
+ total_memory += parse_resource_quantity(memory_request)
+
+ # Also check init containers
+ if pod.spec.init_containers:
+ for container in pod.spec.init_containers:
+ if container.resources and container.resources.requests:
+ cpu_request = container.resources.requests.get("cpu")
+ memory_request = container.resources.requests.get("memory")
+
+ if cpu_request:
+ total_cpu += parse_resource_quantity(cpu_request)
+ if memory_request:
+ total_memory += parse_resource_quantity(memory_request)
+
+ return {"cpu": total_cpu, "memory": total_memory}
+
+
+def get_namespace_quota_status(
+ client: CoreV1Api, namespace: str
+) -> tuple[dict[str, float], dict[str, float]] | None:
+ """
+ Get the current quota usage and limits for a namespace.
+
+ :param client: Kubernetes CoreV1Api client
+ :param namespace: Namespace name
+ :return: Tuple of (used_resources, hard_limits) dictionaries or None if no quota
+ """
+ try:
+ quota_list = client.list_namespaced_resource_quota(namespace=namespace)
+ except ApiException as e:
+ if e.status == 403:
+ logger.warning(
+ "Insufficient permissions to check resource quotas in namespace %s. "
+ "Skipping quota validation.",
+ namespace,
+ )
+ return None
+ if e.status == 404:
+ logger.debug("Namespace %s not found for quota check", namespace)
+ return None
+ logger.warning("Error checking resource quotas: %s", e)
+ return None
+
+ if not quota_list.items:
+ # No resource quotas defined for this namespace
+ return None
+
+ # Aggregate all quotas in the namespace
+ total_used = {"cpu": 0.0, "memory": 0.0}
+ total_hard = {"cpu": 0.0, "memory": 0.0}
+
+ for quota in quota_list.items:
+ if quota.status and quota.status.used:
+ cpu_used = quota.status.used.get("requests.cpu") or quota.status.used.get("cpu")
+ memory_used = quota.status.used.get("requests.memory") or quota.status.used.get("memory")
+
+ if cpu_used:
+ total_used["cpu"] += parse_resource_quantity(cpu_used)
+ if memory_used:
+ total_used["memory"] += parse_resource_quantity(memory_used)
+
+ if quota.spec and quota.spec.hard:
+ cpu_hard = quota.spec.hard.get("requests.cpu") or quota.spec.hard.get("cpu")
+ memory_hard = quota.spec.hard.get("requests.memory") or quota.spec.hard.get("memory")
+
+ if cpu_hard:
+ total_hard["cpu"] += parse_resource_quantity(cpu_hard)
+ if memory_hard:
+ total_hard["memory"] += parse_resource_quantity(memory_hard)
+
+ # Only return quota info if limits are actually defined
+ if total_hard["cpu"] > 0 or total_hard["memory"] > 0:
+ return total_used, total_hard
+
+ return None
+
+
+def check_pod_quota_compliance(client: CoreV1Api, pod: V1Pod, namespace: str) -> None:
+ """
+ Check if creating a pod would exceed namespace resource quotas.
+
+ :param client: Kubernetes CoreV1Api client
+ :param pod: V1Pod object to check
+ :param namespace: Namespace where pod will be created
+ :raises PodResourceQuotaExceededException: If pod would exceed quota
+ """
+ quota_info = get_namespace_quota_status(client, namespace)
+
+ if not quota_info:
+ # No quotas defined or couldn't check, allow pod creation
+ return
+
+ used_resources, hard_limits = quota_info
Review Comment:
This will still fail if pods are created between getting the resource and the check here, maybe a different approach is meeded
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add resource quota awareness to KubernetesPodOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed pull request #63946: feat: Add resource quota awareness to KubernetesPodOperator
URL: https://github.com/apache/airflow/pull/63946
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org