We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Need more informative error message when failing to load kubeconfig 
Currently if kubeconfig is not properly setup, we'll receive an error like the following which is not very informative:
```
Traceback (most recent call last):
  File "/miniconda/envs/sqlflow-dev/bin/elasticdl", line 11, in <module>
    load_entry_point('elasticdl==0.0.1', 'console_scripts', 'elasticdl')()
  File "/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/elasticdl-0.0.1-py3.6.egg/elasticdl/python/elasticdl/client.py", line 46, in main
    args.func(args)
  File "/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/elasticdl-0.0.1-py3.6.egg/elasticdl/python/elasticdl/api.py", line 43, in train
    _submit_job(image_name, args, container_args)
  File "/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/elasticdl-0.0.1-py3.6.egg/elasticdl/python/elasticdl/api.py", line 132, in _submit_job
    cluster_spec=client_args.cluster_spec,
  File "/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/elasticdl-0.0.1-py3.6.egg/elasticdl/python/common/k8s_client.py", line 51, in __init__
    config.load_kube_config()
  File "/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py", line 645, in load_kube_config
    persist_config=persist_config)
  File "/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py", line 613, in _get_kube_config_loader_for_yaml_file
    **kwargs)
  File "/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py", line 153, in __init__
    self.set_active_context(active_context)
  File "/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py", line 173, in set_active_context
    context_name = self._config['current-context']
  File "/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py", line 495, in __getitem__
    v = self.safe_get(key)
  File "/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py", line 491, in safe_get
    key in self.value):
TypeError: argument of type 'NoneType' is not iterable
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `elasticdl/python/common/k8s_client.py`
Content:
```
1 import os
2 import threading
3 import traceback
4 
5 from kubernetes import client, config, watch
6 from kubernetes.client import (
7     V1EnvVar,
8     V1EnvVarSource,
9     V1ObjectFieldSelector,
10     V1PersistentVolumeClaimVolumeSource,
11 )
12 
13 from elasticdl.python.common.k8s_resource import parse as parse_resource
14 from elasticdl.python.common.k8s_volume import parse as parse_volume
15 from elasticdl.python.common.log_utils import default_logger as logger
16 from elasticdl.python.common.model_utils import load_module
17 
18 ELASTICDL_APP_NAME = "elasticdl"
19 ELASTICDL_JOB_KEY = "elasticdl-job-name"
20 ELASTICDL_REPLICA_TYPE_KEY = "elasticdl-replica-type"
21 ELASTICDL_REPLICA_INDEX_KEY = "elasticdl-replica-index"
22 
23 
24 class Client(object):
25     def __init__(
26         self,
27         *,
28         image_name,
29         namespace,
30         job_name,
31         event_callback=None,
32         cluster_spec=""
33     ):
34         """
35         ElasticDL k8s client.
36 
37         Args:
38             image_name: Docker image path for ElasticDL pod.
39             namespace: The name of the Kubernetes namespace where ElasticDL
40                 pods will be created.
41             job_name: ElasticDL job name, should be unique in the namespace.
42                 Used as pod name prefix and value for "elasticdl" label.
43             event_callback: If not None, an event watcher will be created and
44                 events passed to the callback.
45         """
46         if os.getenv("KUBERNETES_SERVICE_HOST"):
47             # We are running inside k8s
48             config.load_incluster_config()
49         else:
50             # Use user's kube config
51             config.load_kube_config()
52 
53         self.client = client.CoreV1Api()
54         self.namespace = namespace
55         self.job_name = job_name
56         self._image_name = image_name
57         self._event_cb = event_callback
58         if self._event_cb:
59             threading.Thread(
60                 target=self._watch, name="event_watcher", daemon=True
61             ).start()
62         self.cluster = None
63         if cluster_spec:
64             cluster_spec_module = load_module(cluster_spec)
65             self.cluster = cluster_spec_module.cluster
66 
67     def _watch(self):
68         stream = watch.Watch().stream(
69             self.client.list_namespaced_pod,
70             self.namespace,
71             label_selector=ELASTICDL_JOB_KEY + "=" + self.job_name,
72         )
73         for event in stream:
74             try:
75                 self._event_cb(event)
76             except Exception:
77                 traceback.print_exc()
78 
79     def get_master_pod_name(self):
80         return "elasticdl-%s-master" % self.job_name
81 
82     def get_worker_pod_name(self, worker_id):
83         return "elasticdl-%s-worker-%s" % (self.job_name, str(worker_id))
84 
85     def get_embedding_service_pod_name(self, embedding_service_id):
86         return "elasticdl-%s-embedding-service-%s" % (
87             self.job_name,
88             str(embedding_service_id),
89         )
90 
91     def patch_labels_to_pod(self, pod_name, labels_dict):
92         body = {"metadata": {"labels": labels_dict}}
93         try:
94             return self.client.patch_namespaced_pod(
95                 name=pod_name, namespace=self.namespace, body=body
96             )
97         except client.api_client.ApiException as e:
98             logger.warning("Exception when patching labels to pod: %s\n" % e)
99             return None
100 
101     def get_master_pod(self):
102         try:
103             return self.client.read_namespaced_pod(
104                 name=self.get_master_pod_name(), namespace=self.namespace
105             )
106         except client.api_client.ApiException as e:
107             logger.warning("Exception when reading master pod: %s\n" % e)
108             return None
109 
110     def get_worker_pod(self, worker_id):
111         try:
112             return self.client.read_namespaced_pod(
113                 name=self.get_worker_pod_name(worker_id),
114                 namespace=self.namespace,
115             )
116         except client.api_client.ApiException as e:
117             logger.warning("Exception when reading worker pod: %s\n" % e)
118             return None
119 
120     def get_embedding_service_pod(self, embedding_service_id):
121         try:
122             return self.client.read_namespaced_pod(
123                 name=self.get_embedding_service_pod_name(embedding_service_id),
124                 namespace=self.namespace,
125             )
126         except client.api_client.ApiException as e:
127             logger.warning(
128                 "Exception when reading embedding service pod: %s\n" % e
129             )
130             return None
131 
132     @staticmethod
133     def create_owner_reference(owner_pod):
134         owner_ref = (
135             [
136                 client.V1OwnerReference(
137                     api_version="v1",
138                     block_owner_deletion=True,
139                     kind="Pod",
140                     name=owner_pod.metadata.name,
141                     uid=owner_pod.metadata.uid,
142                 )
143             ]
144             if owner_pod
145             else None
146         )
147         return owner_ref
148 
149     def _create_pod(self, **kargs):
150         # Container
151         pod_resource_requests = kargs["resource_requests"]
152         pod_resource_limits = kargs["resource_limits"]
153         pod_resource_limits = (
154             pod_resource_limits
155             if pod_resource_limits
156             else pod_resource_requests
157         )
158         container = client.V1Container(
159             name=kargs["pod_name"],
160             image=kargs["image_name"],
161             command=kargs["command"],
162             resources=client.V1ResourceRequirements(
163                 requests=parse_resource(pod_resource_requests),
164                 limits=parse_resource(pod_resource_limits),
165             ),
166             args=kargs["container_args"],
167             image_pull_policy=kargs["image_pull_policy"],
168             env=kargs["env"],
169         )
170 
171         # Pod
172         spec = client.V1PodSpec(
173             containers=[container],
174             restart_policy=kargs["restart_policy"],
175             priority_class_name=kargs["pod_priority"],
176         )
177 
178         # Mount data path
179         if kargs["volume"]:
180             volume_dict = parse_volume(kargs["volume"])
181             volume_name = kargs["pod_name"] + "-volume"
182             volume = client.V1Volume(
183                 name=volume_name,
184                 persistent_volume_claim=V1PersistentVolumeClaimVolumeSource(
185                     claim_name=volume_dict["claim_name"], read_only=False
186                 ),
187             )
188             spec.volumes = [volume]
189             container.volume_mounts = [
190                 client.V1VolumeMount(
191                     name=volume_name, mount_path=volume_dict["mount_path"]
192                 )
193             ]
194 
195         pod = client.V1Pod(
196             spec=spec,
197             metadata=client.V1ObjectMeta(
198                 name=kargs["pod_name"],
199                 labels={
200                     "app": ELASTICDL_APP_NAME,
201                     ELASTICDL_JOB_KEY: kargs["job_name"],
202                 },
203                 owner_references=self.create_owner_reference(
204                     kargs["owner_pod"]
205                 ),
206                 namespace=self.namespace,
207             ),
208         )
209         if self.cluster:
210             pod = self.cluster.with_pod(pod)
211 
212         return pod
213 
214     def create_master(self, **kargs):
215         env = [
216             V1EnvVar(
217                 name="MY_POD_IP",
218                 value_from=V1EnvVarSource(
219                     field_ref=V1ObjectFieldSelector(field_path="status.podIP")
220                 ),
221             )
222         ]
223         if "envs" in kargs:
224             for key in kargs["envs"]:
225                 env.append(V1EnvVar(name=key, value=kargs["envs"][key]))
226 
227         pod = self._create_pod(
228             pod_name=self.get_master_pod_name(),
229             job_name=self.job_name,
230             image_name=self._image_name,
231             command=["python"],
232             resource_requests=kargs["resource_requests"],
233             resource_limits=kargs["resource_limits"],
234             container_args=kargs["args"],
235             pod_priority=kargs["pod_priority"],
236             image_pull_policy=kargs["image_pull_policy"],
237             restart_policy=kargs["restart_policy"],
238             volume=kargs["volume"],
239             owner_pod=None,
240             env=env,
241         )
242         # Add replica type and index
243         pod.metadata.labels[ELASTICDL_REPLICA_TYPE_KEY] = "master"
244         pod.metadata.labels[ELASTICDL_REPLICA_INDEX_KEY] = "0"
245         self.client.create_namespaced_pod(self.namespace, pod)
246         logger.info("Master launched.")
247 
248     def _create_worker_pod(self, pod_name, type_key, **kargs):
249         # Find that master pod that will be used as the owner reference
250         # for this worker pod.
251         master_pod = self.get_master_pod()
252         env = kargs["envs"] if "envs" in kargs else None
253         pod = self._create_pod(
254             pod_name=pod_name,
255             job_name=self.job_name,
256             image_name=self._image_name,
257             command=kargs["command"],
258             resource_requests=kargs["resource_requests"],
259             resource_limits=kargs["resource_limits"],
260             container_args=kargs["args"],
261             pod_priority=kargs["pod_priority"],
262             image_pull_policy=kargs["image_pull_policy"],
263             restart_policy=kargs["restart_policy"],
264             volume=kargs["volume"],
265             owner_pod=master_pod,
266             env=env,
267         )
268         # Add replica type and index
269         pod.metadata.labels[ELASTICDL_REPLICA_TYPE_KEY] = type_key
270         pod.metadata.labels[ELASTICDL_REPLICA_INDEX_KEY] = str(
271             kargs["worker_id"]
272         )
273         return self.client.create_namespaced_pod(self.namespace, pod)
274 
275     def create_worker(self, **kargs):
276         pod_name = self.get_worker_pod_name(kargs["worker_id"])
277         return self._create_worker_pod(pod_name, "worker", **kargs)
278 
279     def create_embedding_service(self, **kargs):
280         pod_name = self.get_embedding_service_pod_name(kargs["worker_id"])
281         return self._create_worker_pod(pod_name, "embedding_service", **kargs)
282 
283     def delete_master(self):
284         logger.info("pod name is %s" % self.get_master_pod_name())
285         self.client.delete_namespaced_pod(
286             self.get_master_pod_name(),
287             self.namespace,
288             body=client.V1DeleteOptions(grace_period_seconds=0),
289         )
290 
291     def delete_worker(self, worker_id):
292         self.client.delete_namespaced_pod(
293             self.get_worker_pod_name(worker_id),
294             self.namespace,
295             body=client.V1DeleteOptions(grace_period_seconds=0),
296         )
297 
298     def delete_embedding_service(self, embedding_service_id):
299         self.client.delete_namespaced_pod(
300             self.get_embedding_service_pod_name(embedding_service_id),
301             self.namespace,
302             body=client.V1DeleteOptions(grace_period_seconds=0),
303         )
304
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/elasticdl/python/common/k8s_client.py b/elasticdl/python/common/k8s_client.py
--- a/elasticdl/python/common/k8s_client.py
+++ b/elasticdl/python/common/k8s_client.py
@@ -43,12 +43,18 @@
             event_callback: If not None, an event watcher will be created and
                 events passed to the callback.
         """
-        if os.getenv("KUBERNETES_SERVICE_HOST"):
-            # We are running inside k8s
-            config.load_incluster_config()
-        else:
-            # Use user's kube config
-            config.load_kube_config()
+        try:
+            if os.getenv("KUBERNETES_SERVICE_HOST"):
+                # We are running inside a k8s cluster
+                config.load_incluster_config()
+            else:
+                # Use user's kube config
+                config.load_kube_config()
+        except Exception as ex:
+            traceback.print_exc()
+            raise Exception(
+                "Failed to load configuration for Kubernetes:\n%s" % str(ex)
+            )
 
         self.client = client.CoreV1Api()
         self.namespace = namespace

{"golden_diff": "diff --git a/elasticdl/python/common/k8s_client.py b/elasticdl/python/common/k8s_client.py\n--- a/elasticdl/python/common/k8s_client.py\n+++ b/elasticdl/python/common/k8s_client.py\n@@ -43,12 +43,18 @@\n             event_callback: If not None, an event watcher will be created and\n                 events passed to the callback.\n         \"\"\"\n-        if os.getenv(\"KUBERNETES_SERVICE_HOST\"):\n-            # We are running inside k8s\n-            config.load_incluster_config()\n-        else:\n-            # Use user's kube config\n-            config.load_kube_config()\n+        try:\n+            if os.getenv(\"KUBERNETES_SERVICE_HOST\"):\n+                # We are running inside a k8s cluster\n+                config.load_incluster_config()\n+            else:\n+                # Use user's kube config\n+                config.load_kube_config()\n+        except Exception as ex:\n+            traceback.print_exc()\n+            raise Exception(\n+                \"Failed to load configuration for Kubernetes:\\n%s\" % str(ex)\n+            )\n \n         self.client = client.CoreV1Api()\n         self.namespace = namespace\n", "issue": "Need more informative error message when failing to load kubeconfig \nCurrently if kubeconfig is not properly setup, we'll receive an error like the following which is not very informative:\r\n```\r\nTraceback (most recent call last):\r\n  File \"/miniconda/envs/sqlflow-dev/bin/elasticdl\", line 11, in <module>\r\n    load_entry_point('elasticdl==0.0.1', 'console_scripts', 'elasticdl')()\r\n  File \"/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/elasticdl-0.0.1-py3.6.egg/elasticdl/python/elasticdl/client.py\", line 46, in main\r\n    args.func(args)\r\n  File \"/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/elasticdl-0.0.1-py3.6.egg/elasticdl/python/elasticdl/api.py\", line 43, in train\r\n    _submit_job(image_name, args, container_args)\r\n  File \"/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/elasticdl-0.0.1-py3.6.egg/elasticdl/python/elasticdl/api.py\", line 132, in _submit_job\r\n    cluster_spec=client_args.cluster_spec,\r\n  File \"/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/elasticdl-0.0.1-py3.6.egg/elasticdl/python/common/k8s_client.py\", line 51, in __init__\r\n    config.load_kube_config()\r\n  File \"/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py\", line 645, in load_kube_config\r\n    persist_config=persist_config)\r\n  File \"/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py\", line 613, in _get_kube_config_loader_for_yaml_file\r\n    **kwargs)\r\n  File \"/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py\", line 153, in __init__\r\n    self.set_active_context(active_context)\r\n  File \"/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py\", line 173, in set_active_context\r\n    context_name = self._config['current-context']\r\n  File \"/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py\", line 495, in __getitem__\r\n    v = self.safe_get(key)\r\n  File \"/miniconda/envs/sqlflow-dev/lib/python3.6/site-packages/kubernetes/config/kube_config.py\", line 491, in safe_get\r\n    key in self.value):\r\nTypeError: argument of type 'NoneType' is not iterable\r\n```\n", "before_files": [{"content": "import os\nimport threading\nimport traceback\n\nfrom kubernetes import client, config, watch\nfrom kubernetes.client import (\n    V1EnvVar,\n    V1EnvVarSource,\n    V1ObjectFieldSelector,\n    V1PersistentVolumeClaimVolumeSource,\n)\n\nfrom elasticdl.python.common.k8s_resource import parse as parse_resource\nfrom elasticdl.python.common.k8s_volume import parse as parse_volume\nfrom elasticdl.python.common.log_utils import default_logger as logger\nfrom elasticdl.python.common.model_utils import load_module\n\nELASTICDL_APP_NAME = \"elasticdl\"\nELASTICDL_JOB_KEY = \"elasticdl-job-name\"\nELASTICDL_REPLICA_TYPE_KEY = \"elasticdl-replica-type\"\nELASTICDL_REPLICA_INDEX_KEY = \"elasticdl-replica-index\"\n\n\nclass Client(object):\n    def __init__(\n        self,\n        *,\n        image_name,\n        namespace,\n        job_name,\n        event_callback=None,\n        cluster_spec=\"\"\n    ):\n        \"\"\"\n        ElasticDL k8s client.\n\n        Args:\n            image_name: Docker image path for ElasticDL pod.\n            namespace: The name of the Kubernetes namespace where ElasticDL\n                pods will be created.\n            job_name: ElasticDL job name, should be unique in the namespace.\n                Used as pod name prefix and value for \"elasticdl\" label.\n            event_callback: If not None, an event watcher will be created and\n                events passed to the callback.\n        \"\"\"\n        if os.getenv(\"KUBERNETES_SERVICE_HOST\"):\n            # We are running inside k8s\n            config.load_incluster_config()\n        else:\n            # Use user's kube config\n            config.load_kube_config()\n\n        self.client = client.CoreV1Api()\n        self.namespace = namespace\n        self.job_name = job_name\n        self._image_name = image_name\n        self._event_cb = event_callback\n        if self._event_cb:\n            threading.Thread(\n                target=self._watch, name=\"event_watcher\", daemon=True\n            ).start()\n        self.cluster = None\n        if cluster_spec:\n            cluster_spec_module = load_module(cluster_spec)\n            self.cluster = cluster_spec_module.cluster\n\n    def _watch(self):\n        stream = watch.Watch().stream(\n            self.client.list_namespaced_pod,\n            self.namespace,\n            label_selector=ELASTICDL_JOB_KEY + \"=\" + self.job_name,\n        )\n        for event in stream:\n            try:\n                self._event_cb(event)\n            except Exception:\n                traceback.print_exc()\n\n    def get_master_pod_name(self):\n        return \"elasticdl-%s-master\" % self.job_name\n\n    def get_worker_pod_name(self, worker_id):\n        return \"elasticdl-%s-worker-%s\" % (self.job_name, str(worker_id))\n\n    def get_embedding_service_pod_name(self, embedding_service_id):\n        return \"elasticdl-%s-embedding-service-%s\" % (\n            self.job_name,\n            str(embedding_service_id),\n        )\n\n    def patch_labels_to_pod(self, pod_name, labels_dict):\n        body = {\"metadata\": {\"labels\": labels_dict}}\n        try:\n            return self.client.patch_namespaced_pod(\n                name=pod_name, namespace=self.namespace, body=body\n            )\n        except client.api_client.ApiException as e:\n            logger.warning(\"Exception when patching labels to pod: %s\\n\" % e)\n            return None\n\n    def get_master_pod(self):\n        try:\n            return self.client.read_namespaced_pod(\n                name=self.get_master_pod_name(), namespace=self.namespace\n            )\n        except client.api_client.ApiException as e:\n            logger.warning(\"Exception when reading master pod: %s\\n\" % e)\n            return None\n\n    def get_worker_pod(self, worker_id):\n        try:\n            return self.client.read_namespaced_pod(\n                name=self.get_worker_pod_name(worker_id),\n                namespace=self.namespace,\n            )\n        except client.api_client.ApiException as e:\n            logger.warning(\"Exception when reading worker pod: %s\\n\" % e)\n            return None\n\n    def get_embedding_service_pod(self, embedding_service_id):\n        try:\n            return self.client.read_namespaced_pod(\n                name=self.get_embedding_service_pod_name(embedding_service_id),\n                namespace=self.namespace,\n            )\n        except client.api_client.ApiException as e:\n            logger.warning(\n                \"Exception when reading embedding service pod: %s\\n\" % e\n            )\n            return None\n\n    @staticmethod\n    def create_owner_reference(owner_pod):\n        owner_ref = (\n            [\n                client.V1OwnerReference(\n                    api_version=\"v1\",\n                    block_owner_deletion=True,\n                    kind=\"Pod\",\n                    name=owner_pod.metadata.name,\n                    uid=owner_pod.metadata.uid,\n                )\n            ]\n            if owner_pod\n            else None\n        )\n        return owner_ref\n\n    def _create_pod(self, **kargs):\n        # Container\n        pod_resource_requests = kargs[\"resource_requests\"]\n        pod_resource_limits = kargs[\"resource_limits\"]\n        pod_resource_limits = (\n            pod_resource_limits\n            if pod_resource_limits\n            else pod_resource_requests\n        )\n        container = client.V1Container(\n            name=kargs[\"pod_name\"],\n            image=kargs[\"image_name\"],\n            command=kargs[\"command\"],\n            resources=client.V1ResourceRequirements(\n                requests=parse_resource(pod_resource_requests),\n                limits=parse_resource(pod_resource_limits),\n            ),\n            args=kargs[\"container_args\"],\n            image_pull_policy=kargs[\"image_pull_policy\"],\n            env=kargs[\"env\"],\n        )\n\n        # Pod\n        spec = client.V1PodSpec(\n            containers=[container],\n            restart_policy=kargs[\"restart_policy\"],\n            priority_class_name=kargs[\"pod_priority\"],\n        )\n\n        # Mount data path\n        if kargs[\"volume\"]:\n            volume_dict = parse_volume(kargs[\"volume\"])\n            volume_name = kargs[\"pod_name\"] + \"-volume\"\n            volume = client.V1Volume(\n                name=volume_name,\n                persistent_volume_claim=V1PersistentVolumeClaimVolumeSource(\n                    claim_name=volume_dict[\"claim_name\"], read_only=False\n                ),\n            )\n            spec.volumes = [volume]\n            container.volume_mounts = [\n                client.V1VolumeMount(\n                    name=volume_name, mount_path=volume_dict[\"mount_path\"]\n                )\n            ]\n\n        pod = client.V1Pod(\n            spec=spec,\n            metadata=client.V1ObjectMeta(\n                name=kargs[\"pod_name\"],\n                labels={\n                    \"app\": ELASTICDL_APP_NAME,\n                    ELASTICDL_JOB_KEY: kargs[\"job_name\"],\n                },\n                owner_references=self.create_owner_reference(\n                    kargs[\"owner_pod\"]\n                ),\n                namespace=self.namespace,\n            ),\n        )\n        if self.cluster:\n            pod = self.cluster.with_pod(pod)\n\n        return pod\n\n    def create_master(self, **kargs):\n        env = [\n            V1EnvVar(\n                name=\"MY_POD_IP\",\n                value_from=V1EnvVarSource(\n                    field_ref=V1ObjectFieldSelector(field_path=\"status.podIP\")\n                ),\n            )\n        ]\n        if \"envs\" in kargs:\n            for key in kargs[\"envs\"]:\n                env.append(V1EnvVar(name=key, value=kargs[\"envs\"][key]))\n\n        pod = self._create_pod(\n            pod_name=self.get_master_pod_name(),\n            job_name=self.job_name,\n            image_name=self._image_name,\n            command=[\"python\"],\n            resource_requests=kargs[\"resource_requests\"],\n            resource_limits=kargs[\"resource_limits\"],\n            container_args=kargs[\"args\"],\n            pod_priority=kargs[\"pod_priority\"],\n            image_pull_policy=kargs[\"image_pull_policy\"],\n            restart_policy=kargs[\"restart_policy\"],\n            volume=kargs[\"volume\"],\n            owner_pod=None,\n            env=env,\n        )\n        # Add replica type and index\n        pod.metadata.labels[ELASTICDL_REPLICA_TYPE_KEY] = \"master\"\n        pod.metadata.labels[ELASTICDL_REPLICA_INDEX_KEY] = \"0\"\n        self.client.create_namespaced_pod(self.namespace, pod)\n        logger.info(\"Master launched.\")\n\n    def _create_worker_pod(self, pod_name, type_key, **kargs):\n        # Find that master pod that will be used as the owner reference\n        # for this worker pod.\n        master_pod = self.get_master_pod()\n        env = kargs[\"envs\"] if \"envs\" in kargs else None\n        pod = self._create_pod(\n            pod_name=pod_name,\n            job_name=self.job_name,\n            image_name=self._image_name,\n            command=kargs[\"command\"],\n            resource_requests=kargs[\"resource_requests\"],\n            resource_limits=kargs[\"resource_limits\"],\n            container_args=kargs[\"args\"],\n            pod_priority=kargs[\"pod_priority\"],\n            image_pull_policy=kargs[\"image_pull_policy\"],\n            restart_policy=kargs[\"restart_policy\"],\n            volume=kargs[\"volume\"],\n            owner_pod=master_pod,\n            env=env,\n        )\n        # Add replica type and index\n        pod.metadata.labels[ELASTICDL_REPLICA_TYPE_KEY] = type_key\n        pod.metadata.labels[ELASTICDL_REPLICA_INDEX_KEY] = str(\n            kargs[\"worker_id\"]\n        )\n        return self.client.create_namespaced_pod(self.namespace, pod)\n\n    def create_worker(self, **kargs):\n        pod_name = self.get_worker_pod_name(kargs[\"worker_id\"])\n        return self._create_worker_pod(pod_name, \"worker\", **kargs)\n\n    def create_embedding_service(self, **kargs):\n        pod_name = self.get_embedding_service_pod_name(kargs[\"worker_id\"])\n        return self._create_worker_pod(pod_name, \"embedding_service\", **kargs)\n\n    def delete_master(self):\n        logger.info(\"pod name is %s\" % self.get_master_pod_name())\n        self.client.delete_namespaced_pod(\n            self.get_master_pod_name(),\n            self.namespace,\n            body=client.V1DeleteOptions(grace_period_seconds=0),\n        )\n\n    def delete_worker(self, worker_id):\n        self.client.delete_namespaced_pod(\n            self.get_worker_pod_name(worker_id),\n            self.namespace,\n            body=client.V1DeleteOptions(grace_period_seconds=0),\n        )\n\n    def delete_embedding_service(self, embedding_service_id):\n        self.client.delete_namespaced_pod(\n            self.get_embedding_service_pod_name(embedding_service_id),\n            self.namespace,\n            body=client.V1DeleteOptions(grace_period_seconds=0),\n        )\n", "path": "elasticdl/python/common/k8s_client.py"}], "after_files": [{"content": "import os\nimport threading\nimport traceback\n\nfrom kubernetes import client, config, watch\nfrom kubernetes.client import (\n    V1EnvVar,\n    V1EnvVarSource,\n    V1ObjectFieldSelector,\n    V1PersistentVolumeClaimVolumeSource,\n)\n\nfrom elasticdl.python.common.k8s_resource import parse as parse_resource\nfrom elasticdl.python.common.k8s_volume import parse as parse_volume\nfrom elasticdl.python.common.log_utils import default_logger as logger\nfrom elasticdl.python.common.model_utils import load_module\n\nELASTICDL_APP_NAME = \"elasticdl\"\nELASTICDL_JOB_KEY = \"elasticdl-job-name\"\nELASTICDL_REPLICA_TYPE_KEY = \"elasticdl-replica-type\"\nELASTICDL_REPLICA_INDEX_KEY = \"elasticdl-replica-index\"\n\n\nclass Client(object):\n    def __init__(\n        self,\n        *,\n        image_name,\n        namespace,\n        job_name,\n        event_callback=None,\n        cluster_spec=\"\"\n    ):\n        \"\"\"\n        ElasticDL k8s client.\n\n        Args:\n            image_name: Docker image path for ElasticDL pod.\n            namespace: The name of the Kubernetes namespace where ElasticDL\n                pods will be created.\n            job_name: ElasticDL job name, should be unique in the namespace.\n                Used as pod name prefix and value for \"elasticdl\" label.\n            event_callback: If not None, an event watcher will be created and\n                events passed to the callback.\n        \"\"\"\n        try:\n            if os.getenv(\"KUBERNETES_SERVICE_HOST\"):\n                # We are running inside a k8s cluster\n                config.load_incluster_config()\n            else:\n                # Use user's kube config\n                config.load_kube_config()\n        except Exception as ex:\n            traceback.print_exc()\n            raise Exception(\n                \"Failed to load configuration for Kubernetes:\\n%s\" % str(ex)\n            )\n\n        self.client = client.CoreV1Api()\n        self.namespace = namespace\n        self.job_name = job_name\n        self._image_name = image_name\n        self._event_cb = event_callback\n        if self._event_cb:\n            threading.Thread(\n                target=self._watch, name=\"event_watcher\", daemon=True\n            ).start()\n        self.cluster = None\n        if cluster_spec:\n            cluster_spec_module = load_module(cluster_spec)\n            self.cluster = cluster_spec_module.cluster\n\n    def _watch(self):\n        stream = watch.Watch().stream(\n            self.client.list_namespaced_pod,\n            self.namespace,\n            label_selector=ELASTICDL_JOB_KEY + \"=\" + self.job_name,\n        )\n        for event in stream:\n            try:\n                self._event_cb(event)\n            except Exception:\n                traceback.print_exc()\n\n    def get_master_pod_name(self):\n        return \"elasticdl-%s-master\" % self.job_name\n\n    def get_worker_pod_name(self, worker_id):\n        return \"elasticdl-%s-worker-%s\" % (self.job_name, str(worker_id))\n\n    def get_embedding_service_pod_name(self, embedding_service_id):\n        return \"elasticdl-%s-embedding-service-%s\" % (\n            self.job_name,\n            str(embedding_service_id),\n        )\n\n    def patch_labels_to_pod(self, pod_name, labels_dict):\n        body = {\"metadata\": {\"labels\": labels_dict}}\n        try:\n            return self.client.patch_namespaced_pod(\n                name=pod_name, namespace=self.namespace, body=body\n            )\n        except client.api_client.ApiException as e:\n            logger.warning(\"Exception when patching labels to pod: %s\\n\" % e)\n            return None\n\n    def get_master_pod(self):\n        try:\n            return self.client.read_namespaced_pod(\n                name=self.get_master_pod_name(), namespace=self.namespace\n            )\n        except client.api_client.ApiException as e:\n            logger.warning(\"Exception when reading master pod: %s\\n\" % e)\n            return None\n\n    def get_worker_pod(self, worker_id):\n        try:\n            return self.client.read_namespaced_pod(\n                name=self.get_worker_pod_name(worker_id),\n                namespace=self.namespace,\n            )\n        except client.api_client.ApiException as e:\n            logger.warning(\"Exception when reading worker pod: %s\\n\" % e)\n            return None\n\n    def get_embedding_service_pod(self, embedding_service_id):\n        try:\n            return self.client.read_namespaced_pod(\n                name=self.get_embedding_service_pod_name(embedding_service_id),\n                namespace=self.namespace,\n            )\n        except client.api_client.ApiException as e:\n            logger.warning(\n                \"Exception when reading embedding service pod: %s\\n\" % e\n            )\n            return None\n\n    @staticmethod\n    def create_owner_reference(owner_pod):\n        owner_ref = (\n            [\n                client.V1OwnerReference(\n                    api_version=\"v1\",\n                    block_owner_deletion=True,\n                    kind=\"Pod\",\n                    name=owner_pod.metadata.name,\n                    uid=owner_pod.metadata.uid,\n                )\n            ]\n            if owner_pod\n            else None\n        )\n        return owner_ref\n\n    def _create_pod(self, **kargs):\n        # Container\n        pod_resource_requests = kargs[\"resource_requests\"]\n        pod_resource_limits = kargs[\"resource_limits\"]\n        pod_resource_limits = (\n            pod_resource_limits\n            if pod_resource_limits\n            else pod_resource_requests\n        )\n        container = client.V1Container(\n            name=kargs[\"pod_name\"],\n            image=kargs[\"image_name\"],\n            command=kargs[\"command\"],\n            resources=client.V1ResourceRequirements(\n                requests=parse_resource(pod_resource_requests),\n                limits=parse_resource(pod_resource_limits),\n            ),\n            args=kargs[\"container_args\"],\n            image_pull_policy=kargs[\"image_pull_policy\"],\n            env=kargs[\"env\"],\n        )\n\n        # Pod\n        spec = client.V1PodSpec(\n            containers=[container],\n            restart_policy=kargs[\"restart_policy\"],\n            priority_class_name=kargs[\"pod_priority\"],\n        )\n\n        # Mount data path\n        if kargs[\"volume\"]:\n            volume_dict = parse_volume(kargs[\"volume\"])\n            volume_name = kargs[\"pod_name\"] + \"-volume\"\n            volume = client.V1Volume(\n                name=volume_name,\n                persistent_volume_claim=V1PersistentVolumeClaimVolumeSource(\n                    claim_name=volume_dict[\"claim_name\"], read_only=False\n                ),\n            )\n            spec.volumes = [volume]\n            container.volume_mounts = [\n                client.V1VolumeMount(\n                    name=volume_name, mount_path=volume_dict[\"mount_path\"]\n                )\n            ]\n\n        pod = client.V1Pod(\n            spec=spec,\n            metadata=client.V1ObjectMeta(\n                name=kargs[\"pod_name\"],\n                labels={\n                    \"app\": ELASTICDL_APP_NAME,\n                    ELASTICDL_JOB_KEY: kargs[\"job_name\"],\n                },\n                owner_references=self.create_owner_reference(\n                    kargs[\"owner_pod\"]\n                ),\n                namespace=self.namespace,\n            ),\n        )\n        if self.cluster:\n            pod = self.cluster.with_pod(pod)\n\n        return pod\n\n    def create_master(self, **kargs):\n        env = [\n            V1EnvVar(\n                name=\"MY_POD_IP\",\n                value_from=V1EnvVarSource(\n                    field_ref=V1ObjectFieldSelector(field_path=\"status.podIP\")\n                ),\n            )\n        ]\n        if \"envs\" in kargs:\n            for key in kargs[\"envs\"]:\n                env.append(V1EnvVar(name=key, value=kargs[\"envs\"][key]))\n\n        pod = self._create_pod(\n            pod_name=self.get_master_pod_name(),\n            job_name=self.job_name,\n            image_name=self._image_name,\n            command=[\"python\"],\n            resource_requests=kargs[\"resource_requests\"],\n            resource_limits=kargs[\"resource_limits\"],\n            container_args=kargs[\"args\"],\n            pod_priority=kargs[\"pod_priority\"],\n            image_pull_policy=kargs[\"image_pull_policy\"],\n            restart_policy=kargs[\"restart_policy\"],\n            volume=kargs[\"volume\"],\n            owner_pod=None,\n            env=env,\n        )\n        # Add replica type and index\n        pod.metadata.labels[ELASTICDL_REPLICA_TYPE_KEY] = \"master\"\n        pod.metadata.labels[ELASTICDL_REPLICA_INDEX_KEY] = \"0\"\n        self.client.create_namespaced_pod(self.namespace, pod)\n        logger.info(\"Master launched.\")\n\n    def _create_worker_pod(self, pod_name, type_key, **kargs):\n        # Find that master pod that will be used as the owner reference\n        # for this worker pod.\n        master_pod = self.get_master_pod()\n        env = kargs[\"envs\"] if \"envs\" in kargs else None\n        pod = self._create_pod(\n            pod_name=pod_name,\n            job_name=self.job_name,\n            image_name=self._image_name,\n            command=kargs[\"command\"],\n            resource_requests=kargs[\"resource_requests\"],\n            resource_limits=kargs[\"resource_limits\"],\n            container_args=kargs[\"args\"],\n            pod_priority=kargs[\"pod_priority\"],\n            image_pull_policy=kargs[\"image_pull_policy\"],\n            restart_policy=kargs[\"restart_policy\"],\n            volume=kargs[\"volume\"],\n            owner_pod=master_pod,\n            env=env,\n        )\n        # Add replica type and index\n        pod.metadata.labels[ELASTICDL_REPLICA_TYPE_KEY] = type_key\n        pod.metadata.labels[ELASTICDL_REPLICA_INDEX_KEY] = str(\n            kargs[\"worker_id\"]\n        )\n        return self.client.create_namespaced_pod(self.namespace, pod)\n\n    def create_worker(self, **kargs):\n        pod_name = self.get_worker_pod_name(kargs[\"worker_id\"])\n        return self._create_worker_pod(pod_name, \"worker\", **kargs)\n\n    def create_embedding_service(self, **kargs):\n        pod_name = self.get_embedding_service_pod_name(kargs[\"worker_id\"])\n        return self._create_worker_pod(pod_name, \"embedding_service\", **kargs)\n\n    def delete_master(self):\n        logger.info(\"pod name is %s\" % self.get_master_pod_name())\n        self.client.delete_namespaced_pod(\n            self.get_master_pod_name(),\n            self.namespace,\n            body=client.V1DeleteOptions(grace_period_seconds=0),\n        )\n\n    def delete_worker(self, worker_id):\n        self.client.delete_namespaced_pod(\n            self.get_worker_pod_name(worker_id),\n            self.namespace,\n            body=client.V1DeleteOptions(grace_period_seconds=0),\n        )\n\n    def delete_embedding_service(self, embedding_service_id):\n        self.client.delete_namespaced_pod(\n            self.get_embedding_service_pod_name(embedding_service_id),\n            self.namespace,\n            body=client.V1DeleteOptions(grace_period_seconds=0),\n        )\n", "path": "elasticdl/python/common/k8s_client.py"}]}