Patch kubernetes Api Client to work around lock contention issues

Unclear whether the benefits are worth the risk of breaking changes here, but this is a workaround for the signifcant lock contention issues with the k8s client surfaced in #23933 (reply in thread). This appears to be a known issue with the k8s python client - see kubernetes-client/python#2284.

Test Plan: BK

Author: gibsondan

python_modules/libraries/dagster-k8s/dagster_k8s/client.py

@@ -6,11 +6,14 @@
 
 import kubernetes.client
 import kubernetes.client.rest
+import six
 from dagster import (
     DagsterInstance,
     _check as check,
 )
 from dagster._core.storage.dagster_run import DagsterRunStatus
+from kubernetes.client.api_client import ApiClient
+from kubernetes.client.configuration import Configuration
 from kubernetes.client.models import V1Job, V1JobStatus
 
 try:
@@ -91,6 +94,42 @@ class DagsterK8sJobStatusException(Exception):
 ]
 
 
+shared_k8s_model_configuration = Configuration()
+
+
+class PatchedApiClient(ApiClient):
+    # Forked from ApiClient implementation to pass configuration object down into created model
+    # objects, avoiding lock contention issues. See https://github.com/kubernetes-client/python/issues/2284
+    def __deserialize_model(self, data, klass):
+        """Deserializes list or dict to model.
+
+        :param data: dict, list.
+        :param klass: class literal.
+        :return: model object.
+        """
+        if not klass.openapi_types and not hasattr(klass, "get_real_child_model"):
+            return data
+
+        # Below is the only change from the base ApiClient implementation - pass through the
+        # Configuration object to each newly created model so that each one does not have to create
+        # one and acquire a lock
+        kwargs = {"local_vars_configuration": shared_k8s_model_configuration}
+
+        if data is not None and klass.openapi_types is not None and isinstance(data, (list, dict)):
+            for attr, attr_type in six.iteritems(klass.openapi_types):
+                if klass.attribute_map[attr] in data:
+                    value = data[klass.attribute_map[attr]]
+                    kwargs[attr] = self.__deserialize(value, attr_type)
+
+        instance = klass(**kwargs)
+
+        if hasattr(instance, "get_real_child_model"):
+            klass_name = instance.get_real_child_model(data)
+            if klass_name:
+                instance = self.__deserialize(data, klass_name)
+        return instance
+
+
 def k8s_api_retry(
     fn: Callable[..., T],
     max_retries: int,
@@ -209,8 +248,12 @@ def __init__(self, batch_api, core_api, logger, sleeper, timer):
     @staticmethod
     def production_client(batch_api_override=None, core_api_override=None):
         return DagsterKubernetesClient(
-            batch_api=batch_api_override or kubernetes.client.BatchV1Api(),
-            core_api=core_api_override or kubernetes.client.CoreV1Api(),
+            batch_api=(
+                batch_api_override or kubernetes.client.BatchV1Api(api_client=PatchedApiClient())
+            ),
+            core_api=(
+                core_api_override or kubernetes.client.CoreV1Api(api_client=PatchedApiClient())
+            ),
             logger=logging.info,
             sleeper=time.sleep,
             timer=time.time,