Lee-W commented on code in PR #68819:
URL: https://github.com/apache/airflow/pull/68819#discussion_r3451109259
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -73,6 +74,30 @@
)
+def _reset_local_vars_configuration(obj: Any) -> None:
Review Comment:
```suggestion
def _reset_local_vars_configuration(config: Any) -> None:
```
IMO, this is more readable. I don't quite understand what `obj` is in the
original context.
Also, should we type it? looks like we know the potential type.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py:
##########
@@ -915,6 +917,123 @@ 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``)."""
+
+ 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."""
+ # Build a pod with the unpicklable in-cluster hook on every nested
config; confirm it is
+ # unpicklable first, then that the reset makes it picklable.
+ pod = k8s.V1Pod(
+ metadata=k8s.V1ObjectMeta(),
+ spec=k8s.V1PodSpec(
+ containers=[k8s.V1Container(name="base",
resources=k8s.V1ResourceRequirements())]
+ ),
+ )
+ self._install_unpicklable_incluster_hook(pod)
+ with pytest.raises((pickle.PicklingError, AttributeError, TypeError)):
Review Comment:
Would like to check when will `AttributeError`, `TypeError` happen in
different versions?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -73,6 +74,30 @@
)
+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.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 fall back to the
+ pre-v36 behaviour which is picklable. A fresh ``Configuration()`` keeps
``client_side_validation`` so
+ the worker-side ``reconcile_pods`` setters still work; ``None`` would
break them.
+ """
+ if obj is None or isinstance(obj, (str, int, float, bool)):
+ return
+ if hasattr(obj, "openapi_types"):
+ obj.local_vars_configuration = Configuration()
+ for attr in obj.openapi_types:
+ _reset_local_vars_configuration(getattr(obj, attr, None))
+ elif isinstance(obj, dict):
+ for v in obj.values():
+ _reset_local_vars_configuration(v)
+ elif isinstance(obj, (list, tuple)):
+ for item in obj:
+ _reset_local_vars_configuration(item)
Review Comment:
What should happen if we go into the `else` block? Is it expected to do
nothing? or not expcted to happen
--
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]
