amoghrajesh commented on code in PR #68819:
URL: https://github.com/apache/airflow/pull/68819#discussion_r3450479016
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -226,6 +226,21 @@ def execute_async(
pod_template_file = executor_config.get("pod_template_file", None)
else:
pod_template_file = None
+
+ # Serialize any ``pod_override`` ``V1Pod`` to a plain dict before
putting it on the
+ # multiprocessing queue. When the scheduler runs in-cluster, the
kubernetes client attaches an
+ # in-cluster ``Configuration`` to every ``V1Pod`` whose
``refresh_api_key_hook`` is a local
+ # closure
(``InClusterConfigLoader._set_config.<locals>._refresh_api_key``). ``pickle``
cannot
+ # serialize a local closure, so queuing a live ``V1Pod`` raises
``PicklingError`` and crashes the
+ # scheduler. ``run_next`` deserializes the dict back into a ``V1Pod``
worker-side. The same
+ # ``V1Pod`` object is also referenced by the workload's
``executor_config``, so sanitize that copy
+ # too (it is otherwise pickled as part of the workload, even though
the worker rebuilds the pod
+ # override from ``kube_executor_config``).
+ if kube_executor_config is not None:
+ kube_executor_config =
PodGenerator.serialize_pod(kube_executor_config)
+ if executor_config and executor_config.get("pod_override") is not None:
+ executor_config["pod_override"] =
PodGenerator.serialize_pod(executor_config["pod_override"])
Review Comment:
Maybe something like this would be a better fix:
```diff
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
(revision 491c167075187a48dc010862cb2ee9a708e822ef)
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
(date 1782112916157)
@@ -73,6 +73,32 @@
)
+def _strip_local_vars_configuration(obj: Any) -> None:
+ """
+ Null out local_vars_configuration on all kubernetes model objects
recursively.
+
+ kubernetes-python-client v36 changed model constructors to call
+ Configuration.get_default_copy() instead of Configuration(), so every
+ model object created after load_incluster_config() runs captures the
global
+ Configuration, which holds an unpicklable refresh_api_key_hook closure.
+ The field is only used for API calls within the same process; workers
+ reinitialize their own kube config, so nulling it before the
multiprocessing
+ queue put loses nothing functional.
+ """
+ if obj is None or isinstance(obj, (str, int, float, bool)):
+ return
+ if hasattr(obj, "openapi_types"):
+ obj.local_vars_configuration = None
+ for attr in obj.openapi_types:
+ _strip_local_vars_configuration(getattr(obj, attr, None))
+ elif isinstance(obj, dict):
+ for v in obj.values():
+ _strip_local_vars_configuration(v)
+ elif isinstance(obj, (list, tuple)):
+ for item in obj:
+ _strip_local_vars_configuration(item)
+
+
class KubernetesExecutor(BaseExecutor):
"""Executor for Kubernetes."""
@@ -226,6 +252,7 @@
pod_template_file = executor_config.get("pod_template_file",
None)
else:
pod_template_file = None
+ _strip_local_vars_configuration(kube_executor_config)
self.event_buffer[key] = (TaskInstanceState.QUEUED,
self.scheduler_job_id)
self.task_queue.put(KubernetesJob(key, command,
kube_executor_config, pod_template_file))
# We keep a temporary local record that we've handled this so we
don't
```
This keeps `V1Pod` as the type through the queue, does not mutate
`executor_config` in-place, and works identically on v35 where
`local_vars_configuration` was already a plain `Configuration()` with no hook
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -226,6 +226,21 @@ def execute_async(
pod_template_file = executor_config.get("pod_template_file", None)
else:
pod_template_file = None
+
+ # Serialize any ``pod_override`` ``V1Pod`` to a plain dict before
putting it on the
+ # multiprocessing queue. When the scheduler runs in-cluster, the
kubernetes client attaches an
+ # in-cluster ``Configuration`` to every ``V1Pod`` whose
``refresh_api_key_hook`` is a local
+ # closure
(``InClusterConfigLoader._set_config.<locals>._refresh_api_key``). ``pickle``
cannot
+ # serialize a local closure, so queuing a live ``V1Pod`` raises
``PicklingError`` and crashes the
+ # scheduler. ``run_next`` deserializes the dict back into a ``V1Pod``
worker-side. The same
+ # ``V1Pod`` object is also referenced by the workload's
``executor_config``, so sanitize that copy
+ # too (it is otherwise pickled as part of the workload, even though
the worker rebuilds the pod
+ # override from ``kube_executor_config``).
+ if kube_executor_config is not None:
+ kube_executor_config =
PodGenerator.serialize_pod(kube_executor_config)
+ if executor_config and executor_config.get("pod_override") is not None:
+ executor_config["pod_override"] =
PodGenerator.serialize_pod(executor_config["pod_override"])
Review Comment:
The root cause here is that `kubernetes-python-client` v36 changed model
constructors from `Configuration()` to `Configuration.get_default_copy()`,
which causes every `V1Pod` created after `in-cluster`config loads to silently
capture the global config, including its unpicklable `refresh_api_key_hook`
closure defined in
https://github.com/kubernetes-client/python/blob/master/kubernetes/aio/config/incluster_config.py#L100
The serialize/deserialize approach works but treats the symptom at the wrong
layer. `local_vars_configuration` is API client state that has no business
being on a pod template override object, it ended up there as a side effect of
the v36 constructor change, not by intent. Rather than round-tripping through a
dict, nulling it out before the queue put is more direct and correct imo.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
