You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "1fanwang (via GitHub)" <gi...@apache.org> on 2026/06/12 19:09:26 UTC
[PR] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
1fanwang opened a new pull request, #68480:
URL: https://github.com/apache/airflow/pull/68480
## Why
`KubernetesExecutor` creates worker pods one at a time within each scheduler loop. In
`sync()` the dequeued tasks are handed to `run_next()`, which builds the pod and calls
`create_namespaced_pod` **synchronously**, blocking on the API response before the next
pod is created. (`run_pod_async` refers only to the watcher tracking the pod afterwards —
the create call itself blocks.)
When per-create latency is high — API round-trip plus mutating admission webhooks, common
in larger clusters — this serializes the scheduler loop and caps how many pods a loop can
launch. Raising `worker_pods_creation_batch_size` does not help, because the dequeued batch
is still created sequentially.
## What changed
Opt-in concurrent creation, off by default:
- `[kubernetes_executor] async_pod_creation` (bool, default `False`) — when enabled, the
pods dequeued in a scheduler loop are built synchronously (pod-mutation hook and
reconciliation unchanged), then their create calls are issued concurrently using the
asynchronous Kubernetes client (`kubernetes_asyncio`, already a provider dependency),
bounded by a semaphore.
- `[kubernetes_executor] pod_creation_max_concurrency` (int, default `0` → falls back to
`worker_pods_creation_batch_size`) — caps simultaneous create requests so the burst
respects API-server priority-and-fairness / rate limits.
- The sequential path stays the default; its create behaviour is unchanged. The async
`ApiException` is converted to the sync client's `ApiException`, so the existing retry /
exceeded-quota / 429-backoff handling is reused untouched by both paths.
- New `pod_creation_batch_duration` timer (and `pod_creation_batch_size` gauge), emitted by
both paths, measure per-loop batch creation time.
## E2E validation
Live `kind` cluster, full Airflow deployed via the official Helm chart with
`KubernetesExecutor`, a fan-out DAG, and a mutating admission webhook injecting 200ms
per-create latency on worker pods (modelling real API + admission latency, since kind
admits in sub-ms). Batch wall-time measured from the webhook's admission timestamps,
independent of Airflow's own metrics.
Per scheduler-loop batch (what `pod_creation_batch_duration` measures):
| metric | serial (`async_pod_creation=False`) | async (`=True`) |
|---|---|---|
| in-loop batch create time | ~3.3s | ~0.11s |
| concurrent admissions (<50ms apart) | 0 / 29 | 24 / 29 |
End-to-end for a 30-task fan-out: **7.95s → 3.06s** (~2.5x). The end-to-end factor is
bounded by how many tasks the scheduler queues per loop; the per-loop create itself is
~30x faster, and the speedup scales with batch size per loop (the large single-DAG burst
case).
## Tests
- The existing sequential create / publish-retry / 403-quota / 409 / 410 / 429 / 500 cases
pass unchanged — no behaviour change on the default path.
- New: concurrent creation creates every pod; the semaphore bound is respected; the
async→sync `ApiException` conversion drives the existing quota-requeue and
429-`create_pods_after` backoff; a build/create failure for one pod isolates to that task
while the rest of the batch still create; config parsing.
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on code in PR #68480:
URL: https://github.com/apache/airflow/pull/68480#discussion_r3438001139
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py:
##########
@@ -776,6 +776,323 @@ def test_run_next_pod_reconciliation_error(
finally:
kubernetes_executor.end()
+ @pytest.mark.skipif(
+ AirflowKubernetesScheduler is None, reason="kubernetes python package is not installed"
+ )
+ @mock.patch("airflow.providers.cncf.kubernetes.kube_client.get_kube_client")
+ def test_kube_config_parses_async_pod_creation_options(self, mock_get_kube_client):
+ """The two new config keys are parsed onto KubeConfig."""
+ with conf_vars(
+ {
+ ("kubernetes_executor", "async_pod_creation"): "True",
+ ("kubernetes_executor", "pod_creation_max_concurrency"): "7",
+ }
+ ):
+ executor = KubernetesExecutor()
+ assert executor.kube_config.async_pod_creation is True
+ assert executor.kube_config.pod_creation_max_concurrency == 7
Review Comment:
Agreed, removed
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on code in PR #68480:
URL: https://github.com/apache/airflow/pull/68480#discussion_r3438015729
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -606,10 +641,93 @@ def run_next(self, next_job: KubernetesJob) -> None:
)
self.log.debug("Kubernetes running for command %s", command)
self.log.debug("Kubernetes launching image %s", pod.spec.containers[0].image)
+ return pod
- # the watcher will monitor pods, so we do not block.
- self.run_pod_async(pod, **self.kube_config.kube_client_request_args)
- self.log.debug("Kubernetes Job created!")
+ def run_next_batch(self, next_jobs: list[KubernetesJob]) -> list[tuple[KubernetesJob, Exception | None]]:
+ """
+ Build and create a batch of worker pods, parallelizing the create API calls.
+
+ Pod request objects are built synchronously (pod-mutation hook, reconciliation),
+ then the create calls are issued concurrently against the Kubernetes API using the
+ asynchronous client, bounded by ``pod_creation_max_concurrency``. Returns one
+ ``(job, exception)`` pair per job (``exception`` is ``None`` on success). Build and
+ create failures are returned rather than raised so the caller can apply the same
+ per-task handling as the sequential path.
+ """
+ built: list[tuple[KubernetesJob, k8s.V1Pod | None, Exception | None]] = []
+ for job in next_jobs:
+ try:
+ built.append((job, self._build_pod_request(job), None))
+ except Exception as e:
+ built.append((job, None, e))
+
+ to_create = [(job, pod) for job, pod, build_err in built if build_err is None and pod is not None]
+ create_errors: dict[int, Exception | None] = {}
+ if to_create:
+ create_errors = self._run_pods_async(to_create)
+
+ return [
+ (job, build_err if build_err is not None else create_errors.get(id(job)))
+ for job, _, build_err in built
+ ]
+
+ def _run_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Create the given pods concurrently on a dedicated event loop; return per-job error (or None)."""
+ if self._async_loop is None:
+ self._async_loop = asyncio.new_event_loop()
+ return self._async_loop.run_until_complete(self._create_pods_async(jobs_and_pods))
+
+ async def _create_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Issue create_namespaced_pod calls concurrently, bounded by a semaphore."""
+ if self._async_pod_client is None:
+ self._async_pod_client = await get_async_kube_client()
+ api = self._async_pod_client
+ semaphore = asyncio.Semaphore(self.pod_creation_max_concurrency)
+ request_kwargs: dict[str, Any] = self.kube_config.kube_client_request_args or {}
+
+ async def _create(pod: k8s.V1Pod) -> None:
Review Comment:
Thanks for the idea!
Actually I checked a bit further - The executor uses a different retry model than KPO — on a failed create it re-queues the task (task_publish_retries + create_pods_after) and retries next scheduler loop, rather than in-line. Both paths share that via _handle_pod_publish_error, so sync and async fail/retry identically. Wrapping the create in @generic_api_retry would stack an in-line tenacity retry on top, and since the concurrent batch runs via run_until_complete, its Retry-After sleeps would block the scheduler loop.
While checking this I can see a real gap that exists today - neither path retries 502/503/504 today, which generic_api_retry's transient set covers — happy to broaden both paths in a follow-up.
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #68480:
URL: https://github.com/apache/airflow/pull/68480
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on PR #68480:
URL: https://github.com/apache/airflow/pull/68480#issuecomment-4727399743
Please resolve conflicts
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #68480:
URL: https://github.com/apache/airflow/pull/68480#issuecomment-4798269155
@1fanwang — this PR has 1 unresolved review thread(s) that still look like they need your attention. Once you've addressed them (push changes and/or reply in-thread), please resolve the threads or reply to confirm, and give the reviewer a nudge for another look. Thanks!
See the [PR quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
<sub>Automated first-pass triage note drafted by an AI-assisted tool — may get things wrong; once addressed, a real Apache Airflow maintainer takes the next look. ([why automated](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md))</sub>
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @potiuk before posting
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #68480:
URL: https://github.com/apache/airflow/pull/68480#discussion_r3409284075
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -606,10 +641,93 @@ def run_next(self, next_job: KubernetesJob) -> None:
)
self.log.debug("Kubernetes running for command %s", command)
self.log.debug("Kubernetes launching image %s", pod.spec.containers[0].image)
+ return pod
- # the watcher will monitor pods, so we do not block.
- self.run_pod_async(pod, **self.kube_config.kube_client_request_args)
- self.log.debug("Kubernetes Job created!")
+ def run_next_batch(self, next_jobs: list[KubernetesJob]) -> list[tuple[KubernetesJob, Exception | None]]:
+ """
+ Build and create a batch of worker pods, parallelizing the create API calls.
+
+ Pod request objects are built synchronously (pod-mutation hook, reconciliation),
+ then the create calls are issued concurrently against the Kubernetes API using the
+ asynchronous client, bounded by ``pod_creation_max_concurrency``. Returns one
+ ``(job, exception)`` pair per job (``exception`` is ``None`` on success). Build and
+ create failures are returned rather than raised so the caller can apply the same
+ per-task handling as the sequential path.
+ """
+ built: list[tuple[KubernetesJob, k8s.V1Pod | None, Exception | None]] = []
+ for job in next_jobs:
+ try:
+ built.append((job, self._build_pod_request(job), None))
+ except Exception as e:
+ built.append((job, None, e))
+
+ to_create = [(job, pod) for job, pod, build_err in built if build_err is None and pod is not None]
+ create_errors: dict[int, Exception | None] = {}
+ if to_create:
+ create_errors = self._run_pods_async(to_create)
+
+ return [
+ (job, build_err if build_err is not None else create_errors.get(id(job)))
+ for job, _, build_err in built
+ ]
+
+ def _run_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Create the given pods concurrently on a dedicated event loop; return per-job error (or None)."""
+ if self._async_loop is None:
+ self._async_loop = asyncio.new_event_loop()
+ return self._async_loop.run_until_complete(self._create_pods_async(jobs_and_pods))
+
+ async def _create_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Issue create_namespaced_pod calls concurrently, bounded by a semaphore."""
+ if self._async_pod_client is None:
+ self._async_pod_client = await get_async_kube_client()
+ api = self._async_pod_client
+ semaphore = asyncio.Semaphore(self.pod_creation_max_concurrency)
+ request_kwargs: dict[str, Any] = self.kube_config.kube_client_request_args or {}
+
+ async def _create(pod: k8s.V1Pod) -> None:
Review Comment:
For improving and hardening under load a lot changes were made in KPO via `@generic_api_retry`, would it make sense to apply the decorator here as well?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -311,92 +311,170 @@ def sync(self) -> None:
last_resource_version[ns] or resource_instance.resource_version[ns]
)
- from kubernetes.client.rest import ApiException
-
if self.create_pods_after and self.create_pods_after > datetime.now():
self.log.warning("Skipping pod creation due to kubernetes rate limit")
return
self.create_pods_after = None
+ if self.kube_config.async_pod_creation:
+ self._create_pods_concurrently()
+ else:
+ self._create_pods_sequentially()
+
+ def _create_pods_sequentially(self) -> None:
+ """Dequeue a batch and create worker pods one at a time (default behavior)."""
+ from kubernetes.client.rest import ApiException
+
+ if TYPE_CHECKING:
+ assert self.kube_scheduler
+ created = 0
+ start = time.monotonic()
with contextlib.suppress(Empty):
for _ in range(self.kube_config.worker_pods_creation_batch_size):
task = self.task_queue.get_nowait()
-
+ created += 1
try:
- key = task.key
self.kube_scheduler.run_next(task)
Review Comment:
I am not a K8sExecutor expert but this deletion looks "strange" or not proper. Are you sure you did not delete by accident? Looks like all the syn Pod creation code is gone.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py:
##########
@@ -776,6 +776,323 @@ def test_run_next_pod_reconciliation_error(
finally:
kubernetes_executor.end()
+ @pytest.mark.skipif(
+ AirflowKubernetesScheduler is None, reason="kubernetes python package is not installed"
+ )
+ @mock.patch("airflow.providers.cncf.kubernetes.kube_client.get_kube_client")
+ def test_kube_config_parses_async_pod_creation_options(self, mock_get_kube_client):
+ """The two new config keys are parsed onto KubeConfig."""
+ with conf_vars(
+ {
+ ("kubernetes_executor", "async_pod_creation"): "True",
+ ("kubernetes_executor", "pod_creation_max_concurrency"): "7",
+ }
+ ):
+ executor = KubernetesExecutor()
+ assert executor.kube_config.async_pod_creation is True
+ assert executor.kube_config.pod_creation_max_concurrency == 7
Review Comment:
I think the value is very low having a test that checks if config properties are read. This is tested implicitly by other tests below.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py:
##########
@@ -154,3 +155,48 @@ def get_kube_client(
api_client = client.ApiClient(configuration=configuration)
return client.CoreV1Api(api_client)
+
+
+async def get_async_kube_client(
Review Comment:
In providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py there is already a API client created that uses also some explicit timeout wrapper. Can you please leverage this codebase w/o creating a new utility?
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on code in PR #68480:
URL: https://github.com/apache/airflow/pull/68480#discussion_r3438003815
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py:
##########
@@ -154,3 +155,48 @@ def get_kube_client(
api_client = client.ApiClient(configuration=configuration)
return client.CoreV1Api(api_client)
+
+
+async def get_async_kube_client(
Review Comment:
Thanks for the pointer! Done in 4718057
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on PR #68480:
URL: https://github.com/apache/airflow/pull/68480#issuecomment-4858997725
@1fanwang Could you please resolve conflicts?
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on code in PR #68480:
URL: https://github.com/apache/airflow/pull/68480#discussion_r3477097801
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -606,10 +641,93 @@ def run_next(self, next_job: KubernetesJob) -> None:
)
self.log.debug("Kubernetes running for command %s", command)
self.log.debug("Kubernetes launching image %s", pod.spec.containers[0].image)
+ return pod
- # the watcher will monitor pods, so we do not block.
- self.run_pod_async(pod, **self.kube_config.kube_client_request_args)
- self.log.debug("Kubernetes Job created!")
+ def run_next_batch(self, next_jobs: list[KubernetesJob]) -> list[tuple[KubernetesJob, Exception | None]]:
+ """
+ Build and create a batch of worker pods, parallelizing the create API calls.
+
+ Pod request objects are built synchronously (pod-mutation hook, reconciliation),
+ then the create calls are issued concurrently against the Kubernetes API using the
+ asynchronous client, bounded by ``pod_creation_max_concurrency``. Returns one
+ ``(job, exception)`` pair per job (``exception`` is ``None`` on success). Build and
+ create failures are returned rather than raised so the caller can apply the same
+ per-task handling as the sequential path.
+ """
+ built: list[tuple[KubernetesJob, k8s.V1Pod | None, Exception | None]] = []
+ for job in next_jobs:
+ try:
+ built.append((job, self._build_pod_request(job), None))
+ except Exception as e:
+ built.append((job, None, e))
+
+ to_create = [(job, pod) for job, pod, build_err in built if build_err is None and pod is not None]
+ create_errors: dict[int, Exception | None] = {}
+ if to_create:
+ create_errors = self._run_pods_async(to_create)
+
+ return [
+ (job, build_err if build_err is not None else create_errors.get(id(job)))
+ for job, _, build_err in built
+ ]
+
+ def _run_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Create the given pods concurrently on a dedicated event loop; return per-job error (or None)."""
+ if self._async_loop is None:
+ self._async_loop = asyncio.new_event_loop()
+ return self._async_loop.run_until_complete(self._create_pods_async(jobs_and_pods))
+
+ async def _create_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Issue create_namespaced_pod calls concurrently, bounded by a semaphore."""
+ if self._async_pod_client is None:
+ self._async_pod_client = await get_async_kube_client()
+ api = self._async_pod_client
+ semaphore = asyncio.Semaphore(self.pod_creation_max_concurrency)
+ request_kwargs: dict[str, Any] = self.kube_config.kube_client_request_args or {}
+
+ async def _create(pod: k8s.V1Pod) -> None:
Review Comment:
Quick follow up on this note: https://github.com/apache/airflow/pull/68998
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on code in PR #68480:
URL: https://github.com/apache/airflow/pull/68480#discussion_r3438015729
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -606,10 +641,93 @@ def run_next(self, next_job: KubernetesJob) -> None:
)
self.log.debug("Kubernetes running for command %s", command)
self.log.debug("Kubernetes launching image %s", pod.spec.containers[0].image)
+ return pod
- # the watcher will monitor pods, so we do not block.
- self.run_pod_async(pod, **self.kube_config.kube_client_request_args)
- self.log.debug("Kubernetes Job created!")
+ def run_next_batch(self, next_jobs: list[KubernetesJob]) -> list[tuple[KubernetesJob, Exception | None]]:
+ """
+ Build and create a batch of worker pods, parallelizing the create API calls.
+
+ Pod request objects are built synchronously (pod-mutation hook, reconciliation),
+ then the create calls are issued concurrently against the Kubernetes API using the
+ asynchronous client, bounded by ``pod_creation_max_concurrency``. Returns one
+ ``(job, exception)`` pair per job (``exception`` is ``None`` on success). Build and
+ create failures are returned rather than raised so the caller can apply the same
+ per-task handling as the sequential path.
+ """
+ built: list[tuple[KubernetesJob, k8s.V1Pod | None, Exception | None]] = []
+ for job in next_jobs:
+ try:
+ built.append((job, self._build_pod_request(job), None))
+ except Exception as e:
+ built.append((job, None, e))
+
+ to_create = [(job, pod) for job, pod, build_err in built if build_err is None and pod is not None]
+ create_errors: dict[int, Exception | None] = {}
+ if to_create:
+ create_errors = self._run_pods_async(to_create)
+
+ return [
+ (job, build_err if build_err is not None else create_errors.get(id(job)))
+ for job, _, build_err in built
+ ]
+
+ def _run_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Create the given pods concurrently on a dedicated event loop; return per-job error (or None)."""
+ if self._async_loop is None:
+ self._async_loop = asyncio.new_event_loop()
+ return self._async_loop.run_until_complete(self._create_pods_async(jobs_and_pods))
+
+ async def _create_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Issue create_namespaced_pod calls concurrently, bounded by a semaphore."""
+ if self._async_pod_client is None:
+ self._async_pod_client = await get_async_kube_client()
+ api = self._async_pod_client
+ semaphore = asyncio.Semaphore(self.pod_creation_max_concurrency)
+ request_kwargs: dict[str, Any] = self.kube_config.kube_client_request_args or {}
+
+ async def _create(pod: k8s.V1Pod) -> None:
Review Comment:
Thanks for the idea!
TLDR of my analysis is `@generic_api_retry` does not fully work here, and I even tried doing some refactoring on it to make it more flexible/generic however I think that is better scoped as a follow up seperately.
Actually I checked a bit further - The executor uses a different retry model than KPO — on a failed create it re-queues the task (task_publish_retries + create_pods_after) and retries next scheduler loop, rather than in-line. Both paths share that via _handle_pod_publish_error, so sync and async fail/retry identically. Wrapping the create in @generic_api_retry would stack an in-line tenacity retry on top, and since the concurrent batch runs via run_until_complete, its Retry-After sleeps would block the scheduler loop.
While checking this I can see a real gap that exists today - neither path retries 502/503/504 today, which generic_api_retry's transient set covers — happy to broaden both paths in a follow-up.
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 commented on PR #68480:
URL: https://github.com/apache/airflow/pull/68480#issuecomment-5049621794
@1fanwang could you please resolve conflicts again?
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on code in PR #68480:
URL: https://github.com/apache/airflow/pull/68480#discussion_r3437935121
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -311,92 +311,170 @@ def sync(self) -> None:
last_resource_version[ns] or resource_instance.resource_version[ns]
)
- from kubernetes.client.rest import ApiException
-
if self.create_pods_after and self.create_pods_after > datetime.now():
self.log.warning("Skipping pod creation due to kubernetes rate limit")
return
self.create_pods_after = None
+ if self.kube_config.async_pod_creation:
+ self._create_pods_concurrently()
+ else:
+ self._create_pods_sequentially()
+
+ def _create_pods_sequentially(self) -> None:
+ """Dequeue a batch and create worker pods one at a time (default behavior)."""
+ from kubernetes.client.rest import ApiException
+
+ if TYPE_CHECKING:
+ assert self.kube_scheduler
+ created = 0
+ start = time.monotonic()
with contextlib.suppress(Empty):
for _ in range(self.kube_config.worker_pods_creation_batch_size):
task = self.task_queue.get_nowait()
-
+ created += 1
try:
- key = task.key
self.kube_scheduler.run_next(task)
Review Comment:
ah lemme clarify - it's moved not deleted
The old inline loop is now _create_pods_sequentially() and stays the default: sync() calls it whenever async_pod_creation is off. The block that looks removed is now _handle_pod_publish_error(), shared by both paths.
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on PR #68480:
URL: https://github.com/apache/airflow/pull/68480#issuecomment-4803446127
> @1fanwang — this PR has 1 unresolved review thread(s) that still look like they need your attention. Once you've addressed them (push changes and/or reply in-thread), please resolve the threads or reply to confirm, and give the reviewer a nudge for another look. Thanks!
>
> See the [PR quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
>
> Automated first-pass triage note drafted by an AI-assisted tool — may get things wrong; once addressed, a real Apache Airflow maintainer takes the next look. ([why automated](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md))
>
> Drafted-by: Claude Code (Opus 4.8); reviewed by @potiuk before posting
thanks for the ping, resolved
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on code in PR #68480:
URL: https://github.com/apache/airflow/pull/68480#discussion_r3438015729
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -606,10 +641,93 @@ def run_next(self, next_job: KubernetesJob) -> None:
)
self.log.debug("Kubernetes running for command %s", command)
self.log.debug("Kubernetes launching image %s", pod.spec.containers[0].image)
+ return pod
- # the watcher will monitor pods, so we do not block.
- self.run_pod_async(pod, **self.kube_config.kube_client_request_args)
- self.log.debug("Kubernetes Job created!")
+ def run_next_batch(self, next_jobs: list[KubernetesJob]) -> list[tuple[KubernetesJob, Exception | None]]:
+ """
+ Build and create a batch of worker pods, parallelizing the create API calls.
+
+ Pod request objects are built synchronously (pod-mutation hook, reconciliation),
+ then the create calls are issued concurrently against the Kubernetes API using the
+ asynchronous client, bounded by ``pod_creation_max_concurrency``. Returns one
+ ``(job, exception)`` pair per job (``exception`` is ``None`` on success). Build and
+ create failures are returned rather than raised so the caller can apply the same
+ per-task handling as the sequential path.
+ """
+ built: list[tuple[KubernetesJob, k8s.V1Pod | None, Exception | None]] = []
+ for job in next_jobs:
+ try:
+ built.append((job, self._build_pod_request(job), None))
+ except Exception as e:
+ built.append((job, None, e))
+
+ to_create = [(job, pod) for job, pod, build_err in built if build_err is None and pod is not None]
+ create_errors: dict[int, Exception | None] = {}
+ if to_create:
+ create_errors = self._run_pods_async(to_create)
+
+ return [
+ (job, build_err if build_err is not None else create_errors.get(id(job)))
+ for job, _, build_err in built
+ ]
+
+ def _run_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Create the given pods concurrently on a dedicated event loop; return per-job error (or None)."""
+ if self._async_loop is None:
+ self._async_loop = asyncio.new_event_loop()
+ return self._async_loop.run_until_complete(self._create_pods_async(jobs_and_pods))
+
+ async def _create_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Issue create_namespaced_pod calls concurrently, bounded by a semaphore."""
+ if self._async_pod_client is None:
+ self._async_pod_client = await get_async_kube_client()
+ api = self._async_pod_client
+ semaphore = asyncio.Semaphore(self.pod_creation_max_concurrency)
+ request_kwargs: dict[str, Any] = self.kube_config.kube_client_request_args or {}
+
+ async def _create(pod: k8s.V1Pod) -> None:
Review Comment:
Thanks for the idea!
TLDR of my analysis is `@generic_api_retry` does not fully work here, and I even tried doing some refactoring on it to make it more flexible/generic however I think that is better scoped as a follow up seperately.
The executor uses a different retry model than KPO — on a failed create it re-queues the task (task_publish_retries + create_pods_after) and retries next scheduler loop, rather than in-line. Both paths share that via _handle_pod_publish_error, so sync and async fail/retry identically. Wrapping the create in @generic_api_retry would stack an in-line tenacity retry on top, and since the concurrent batch runs via run_until_complete, its Retry-After sleeps would block the scheduler loop.
While checking this I can see a real gap that exists today - neither path retries 502/503/504 today, which generic_api_retry's transient set covers — happy to broaden both paths in a follow-up.
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on code in PR #68480:
URL: https://github.com/apache/airflow/pull/68480#discussion_r3477097801
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -606,10 +641,93 @@ def run_next(self, next_job: KubernetesJob) -> None:
)
self.log.debug("Kubernetes running for command %s", command)
self.log.debug("Kubernetes launching image %s", pod.spec.containers[0].image)
+ return pod
- # the watcher will monitor pods, so we do not block.
- self.run_pod_async(pod, **self.kube_config.kube_client_request_args)
- self.log.debug("Kubernetes Job created!")
+ def run_next_batch(self, next_jobs: list[KubernetesJob]) -> list[tuple[KubernetesJob, Exception | None]]:
+ """
+ Build and create a batch of worker pods, parallelizing the create API calls.
+
+ Pod request objects are built synchronously (pod-mutation hook, reconciliation),
+ then the create calls are issued concurrently against the Kubernetes API using the
+ asynchronous client, bounded by ``pod_creation_max_concurrency``. Returns one
+ ``(job, exception)`` pair per job (``exception`` is ``None`` on success). Build and
+ create failures are returned rather than raised so the caller can apply the same
+ per-task handling as the sequential path.
+ """
+ built: list[tuple[KubernetesJob, k8s.V1Pod | None, Exception | None]] = []
+ for job in next_jobs:
+ try:
+ built.append((job, self._build_pod_request(job), None))
+ except Exception as e:
+ built.append((job, None, e))
+
+ to_create = [(job, pod) for job, pod, build_err in built if build_err is None and pod is not None]
+ create_errors: dict[int, Exception | None] = {}
+ if to_create:
+ create_errors = self._run_pods_async(to_create)
+
+ return [
+ (job, build_err if build_err is not None else create_errors.get(id(job)))
+ for job, _, build_err in built
+ ]
+
+ def _run_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Create the given pods concurrently on a dedicated event loop; return per-job error (or None)."""
+ if self._async_loop is None:
+ self._async_loop = asyncio.new_event_loop()
+ return self._async_loop.run_until_complete(self._create_pods_async(jobs_and_pods))
+
+ async def _create_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Issue create_namespaced_pod calls concurrently, bounded by a semaphore."""
+ if self._async_pod_client is None:
+ self._async_pod_client = await get_async_kube_client()
+ api = self._async_pod_client
+ semaphore = asyncio.Semaphore(self.pod_creation_max_concurrency)
+ request_kwargs: dict[str, Any] = self.kube_config.kube_client_request_args or {}
+
+ async def _create(pod: k8s.V1Pod) -> None:
Review Comment:
Quick follow up on this note for the sync path as well: https://github.com/apache/airflow/pull/68998
--
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] Add opt-in concurrent pod creation to KubernetesExecutor [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on code in PR #68480:
URL: https://github.com/apache/airflow/pull/68480#discussion_r3438015729
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py:
##########
@@ -606,10 +641,93 @@ def run_next(self, next_job: KubernetesJob) -> None:
)
self.log.debug("Kubernetes running for command %s", command)
self.log.debug("Kubernetes launching image %s", pod.spec.containers[0].image)
+ return pod
- # the watcher will monitor pods, so we do not block.
- self.run_pod_async(pod, **self.kube_config.kube_client_request_args)
- self.log.debug("Kubernetes Job created!")
+ def run_next_batch(self, next_jobs: list[KubernetesJob]) -> list[tuple[KubernetesJob, Exception | None]]:
+ """
+ Build and create a batch of worker pods, parallelizing the create API calls.
+
+ Pod request objects are built synchronously (pod-mutation hook, reconciliation),
+ then the create calls are issued concurrently against the Kubernetes API using the
+ asynchronous client, bounded by ``pod_creation_max_concurrency``. Returns one
+ ``(job, exception)`` pair per job (``exception`` is ``None`` on success). Build and
+ create failures are returned rather than raised so the caller can apply the same
+ per-task handling as the sequential path.
+ """
+ built: list[tuple[KubernetesJob, k8s.V1Pod | None, Exception | None]] = []
+ for job in next_jobs:
+ try:
+ built.append((job, self._build_pod_request(job), None))
+ except Exception as e:
+ built.append((job, None, e))
+
+ to_create = [(job, pod) for job, pod, build_err in built if build_err is None and pod is not None]
+ create_errors: dict[int, Exception | None] = {}
+ if to_create:
+ create_errors = self._run_pods_async(to_create)
+
+ return [
+ (job, build_err if build_err is not None else create_errors.get(id(job)))
+ for job, _, build_err in built
+ ]
+
+ def _run_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Create the given pods concurrently on a dedicated event loop; return per-job error (or None)."""
+ if self._async_loop is None:
+ self._async_loop = asyncio.new_event_loop()
+ return self._async_loop.run_until_complete(self._create_pods_async(jobs_and_pods))
+
+ async def _create_pods_async(
+ self, jobs_and_pods: list[tuple[KubernetesJob, k8s.V1Pod]]
+ ) -> dict[int, Exception | None]:
+ """Issue create_namespaced_pod calls concurrently, bounded by a semaphore."""
+ if self._async_pod_client is None:
+ self._async_pod_client = await get_async_kube_client()
+ api = self._async_pod_client
+ semaphore = asyncio.Semaphore(self.pod_creation_max_concurrency)
+ request_kwargs: dict[str, Any] = self.kube_config.kube_client_request_args or {}
+
+ async def _create(pod: k8s.V1Pod) -> None:
Review Comment:
`@generic_api_retry` doesn't fit as-is: the executor re-queues on the next scheduler loop (`task_publish_retries` / `create_pods_after`) rather than retrying in-line, and both paths share that via `_handle_pod_publish_error`. Stacking the decorator's in-line `tenacity` backoff on top would block the scheduler loop — and the whole `run_until_complete` batch on the concurrent path.
So I broadened `_handle_pod_publish_error` instead (in this PR): both paths now re-queue what `generic_api_retry` treats as transient but the executor dropped — `502/503/504` (503's `Retry-After` honored like `429`) and connection errors on the sync (`urllib3`) and async (`aiohttp`) clients, sharing the transient set with `generic_api_retry`. Making the decorator itself more generic stays a separate follow-up.
--
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