amoghrajesh commented on code in PR #68819:
URL: https://github.com/apache/airflow/pull/68819#discussion_r3450813871
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -226,6 +258,15 @@ def execute_async(
pod_template_file = executor_config.get("pod_template_file", None)
else:
pod_template_file = None
+
+ # Reset ``local_vars_configuration`` on the ``pod_override`` to a
fresh ``Configuration()`` before
+ # putting it on the multiprocessing queue. kubernetes-python-client
v36 makes model constructors
+ # capture the global in-cluster ``Configuration`` (with an unpicklable
``refresh_api_key_hook``
+ # closure), so queuing the pod raises ``PicklingError`` and crashes
the scheduler.
+ # ``kube_executor_config`` is the same ``V1Pod`` object referenced by
the workload's
+ # ``executor_config``, so this also makes the workload picklable.
No-op on v35.
+ _reset_local_vars_configuration(kube_executor_config)
Review Comment:
Restates what doc-string states, adds no value
```suggestion
```
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -73,6 +74,37 @@
)
+def _reset_local_vars_configuration(obj: Any) -> None:
+ """
+ Reset ``local_vars_configuration`` to a fresh ``Configuration`` on every
kubernetes model object.
+
+ 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 -- pickling such a pod
onto the multiprocessing
+ queue raises ``PicklingError`` and crashes the scheduler.
+
+ Replace it with a fresh ``Configuration()`` (exactly what v35 model
constructors used): it carries
+ no in-cluster auth hook so it is picklable, while still providing
``client_side_validation`` so the
+ worker-side ``PodGenerator.reconcile_pods`` setters keep working. Setting
it to ``None`` instead
+ would break reconcile -- model setters dereference
+ ``self.local_vars_configuration.client_side_validation``. The field is
only API-client state used
+ within the same process; workers reinitialize their own kube config, so
nothing is lost. No-op on v35.
Review Comment:
```suggestion
Reset local_vars_configuration to a fresh Configuration() on every
kubernetes model object.
kubernetes-python-client v36.0.0 changed model constructors to call
``Configuration.get_default_copy()``
instead of ``Configuration()``, so objects created after incluster
config was loaded capture the global
config whose `refresh_api_key_hook` is an unpicklable local closure
function. We choose to fallback to the pre-v36 behaviour which is picklable.
```
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py:
##########
@@ -915,6 +915,132 @@ def test_pod_template_file_override_in_executor_config(
finally:
executor.end()
+ @staticmethod
+ def _install_unpicklable_incluster_hook(pod):
+ """Simulate the in-cluster kubernetes v36 default config: an
unpicklable local-closure hook
+ on every nested model object's ``Configuration``
(``InClusterConfigLoader._set_config.<locals>.
+ _refresh_api_key``)."""
+ from kubernetes.client import Configuration
Review Comment:
Top level import pls.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py:
##########
@@ -915,6 +915,132 @@ def test_pod_template_file_override_in_executor_config(
finally:
executor.end()
+ @staticmethod
+ def _install_unpicklable_incluster_hook(pod):
+ """Simulate the in-cluster kubernetes v36 default config: an
unpicklable local-closure hook
+ on every nested model object's ``Configuration``
(``InClusterConfigLoader._set_config.<locals>.
+ _refresh_api_key``)."""
+ from kubernetes.client import Configuration
+
+ def _make_hook():
+ def _refresh_api_key(config):
+ return None
+
+ return _refresh_api_key
+
+ for obj in (
+ pod,
+ pod.metadata,
+ pod.spec,
+ pod.spec.containers[0],
+ pod.spec.containers[0].resources,
+ ):
+ cfg = Configuration()
+ cfg.refresh_api_key_hook = _make_hook()
+ obj.local_vars_configuration = cfg
+
+ def test_reset_local_vars_configuration_strips_hook_recursively(self):
+ """``_reset_local_vars_configuration`` must replace every nested
config with a hook-free one."""
+ import pickle
Review Comment:
Top level import pls.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py:
##########
@@ -915,6 +915,132 @@ def test_pod_template_file_override_in_executor_config(
finally:
executor.end()
+ @staticmethod
+ def _install_unpicklable_incluster_hook(pod):
+ """Simulate the in-cluster kubernetes v36 default config: an
unpicklable local-closure hook
+ on every nested model object's ``Configuration``
(``InClusterConfigLoader._set_config.<locals>.
+ _refresh_api_key``)."""
+ from kubernetes.client import Configuration
+
+ def _make_hook():
+ def _refresh_api_key(config):
+ return None
+
+ return _refresh_api_key
+
+ for obj in (
+ pod,
+ pod.metadata,
+ pod.spec,
+ pod.spec.containers[0],
+ pod.spec.containers[0].resources,
+ ):
+ cfg = Configuration()
+ cfg.refresh_api_key_hook = _make_hook()
+ obj.local_vars_configuration = cfg
+
+ def test_reset_local_vars_configuration_strips_hook_recursively(self):
+ """``_reset_local_vars_configuration`` must replace every nested
config with a hook-free one."""
+ import pickle
+
+ from airflow.providers.cncf.kubernetes.executors.kubernetes_executor
import (
+ _reset_local_vars_configuration,
+ )
Review Comment:
Same here
--
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]
