We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Event callback should not check the existence of master pod
Currently, there's error log when event callback checks the pod existence like this `Unknown pod name: elasticdl-master-test-mnist`. This is because `elasticdl-master-test-mnist` is the master pod that's created by users instead of being created by `WorkerManager`. We should fix this since it should only check pod existence for worker pod.

Complete log:
```
2019-06-06 17:38:37,952 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:121] Got event ADDED, phase Running for pod: elasticdl-master-test-mnist
2019-06-06 17:38:37,953 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:71] Starting worker: 1
2019-06-06 17:38:37,953 elasticdl.python.elasticdl.master.k8s_worker_manager ERROR    [k8s_worker_manager.py:127] Unknown pod name: elasticdl-master-test-mnist
2019-06-06 17:38:37,954 elasticdl.python.elasticdl.master.k8s_client INFO     [k8s_client.py:105] Creating worker: 1
2019-06-06 17:38:37,955 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:121] Got event ADDED, phase Pending for pod: elasticdl-worker-test-mnist-0
```
Event callback should not check the existence of master pod
Currently, there's error log when event callback checks the pod existence like this `Unknown pod name: elasticdl-master-test-mnist`. This is because `elasticdl-master-test-mnist` is the master pod that's created by users instead of being created by `WorkerManager`. We should fix this since it should only check pod existence for worker pod.

Complete log:
```
2019-06-06 17:38:37,952 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:121] Got event ADDED, phase Running for pod: elasticdl-master-test-mnist
2019-06-06 17:38:37,953 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:71] Starting worker: 1
2019-06-06 17:38:37,953 elasticdl.python.elasticdl.master.k8s_worker_manager ERROR    [k8s_worker_manager.py:127] Unknown pod name: elasticdl-master-test-mnist
2019-06-06 17:38:37,954 elasticdl.python.elasticdl.master.k8s_client INFO     [k8s_client.py:105] Creating worker: 1
2019-06-06 17:38:37,955 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:121] Got event ADDED, phase Pending for pod: elasticdl-worker-test-mnist-0
```

--- 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/elasticdl/master/k8s_worker_manager.py`
Content:
```
1 import itertools
2 import logging
3 import threading
4 
5 from collections import Counter
6 from elasticdl.python.elasticdl.master import k8s_client as k8s
7 
8 
9 class WorkerManager(object):
10     def __init__(
11             self,
12             task_q,
13             command,
14             args,
15             num_worker=1,
16             cpu_request="1000m",
17             cpu_limit="1000m",
18             memory_request="4096Mi",
19             memory_limit="4096Mi",
20             pod_priority=None,
21             mount_path=None,
22             volume_name=None,
23             image_pull_policy=None,
24             restart_policy="OnFailure",
25             **kwargs):
26         self._logger = logging.getLogger(__name__)
27         self._command = command
28         self._args = args
29         self._num_worker = num_worker
30         self._resource_requests = {
31             "cpu": cpu_request,
32             "memory": memory_request
33         }
34         self._resource_limits = {
35             "cpu": cpu_limit,
36             "memory": memory_limit
37         }
38         self._restart_policy = restart_policy
39         self._pod_priority = pod_priority
40         self._mount_path = mount_path
41         self._volume_name = volume_name
42         self._image_pull_policy = image_pull_policy
43         self._task_q = task_q
44         self._next_worker_id = itertools.count().__next__
45 
46         # protects followed variables, which are accessed from event_cb.
47         self._lock = threading.Lock()
48         # worker id to (pod name, phase) mapping
49         # phase: None/Pending/Running/Succeeded/Failed/Unknown
50         #   None: worker was just launched, haven't received event yet.
51         #   Pending: worker pod not started yet
52         #   Running: worker pod is running
53         #   Succeeded: worker pod finishes all tasks and terminates with
54         #       no issue.
55         #   Failed: worker pod is killed for some reason
56         #   Unknown: unknown
57         self._pods_phase = {}
58         # pod name to worker id mapping
59         self._pod_name_to_id = {}
60 
61         self._relaunch_deleted_live_worker = True
62 
63         self._k8s_client = k8s.Client(
64             event_callback=self._event_cb, **kwargs
65         )
66 
67     def set_relaunch_deleted_live_worker(self, val):
68         self._relaunch_deleted_live_worker = bool(val)
69 
70     def _start_worker(self, worker_id):
71         self._logger.info("Starting worker: %d" % worker_id)
72         with self._lock:
73             pod = self._k8s_client.create_worker(
74                 worker_id,
75                 self._resource_requests,
76                 self._resource_limits,
77                 self._pod_priority,
78                 self._mount_path,
79                 self._volume_name,
80                 self._image_pull_policy,
81                 command=self._command,
82                 args=self._args + ["--worker_id", str(worker_id)],
83                 restart_policy=self._restart_policy,
84             )
85             name = pod.metadata.name
86             self._pod_name_to_id[name] = worker_id
87             self._pods_phase[worker_id] = (name, None)
88 
89     def start_workers(self):
90         for i in range(self._num_worker):
91             self._start_worker(self._next_worker_id())
92 
93     def _remove_worker(self, worker_id):
94         with self._lock:
95             if worker_id not in self._pods_phase:
96                 self._logger.error("Unknown worker id: %s" % worker_id)
97                 return
98 
99         # TODO: change _k8s_client to accept pod name instead of worker id.
100         self._k8s_client.delete_worker(worker_id)
101 
102     def stop_relaunch_and_remove_workers(self):
103         with self._lock:
104             self._relaunch_deleted_live_worker = False
105             for worker_id in self._pods_phase:
106                 self._k8s_client.delete_worker(worker_id)
107 
108     def get_counters(self):
109         with self._lock:
110             return Counter([v for _, v in self._pods_phase.values()])
111 
112     def _event_cb(self, event):
113         evt_obj = event.get("object")
114         evt_type = event.get("type")
115         if not evt_obj or not evt_type:
116             self._logger.error("Event doesn't have object or type: %s" % event)
117             return
118 
119         pod_name = evt_obj.metadata.name
120         phase = evt_obj.status.phase
121         self._logger.info("Got event %s, phase %s for pod: %s" % (evt_type, phase, pod_name))
122 
123         relaunch = False
124         with self._lock:
125             worker_id = self._pod_name_to_id.get(pod_name)
126             if worker_id is None:
127                 self._logger.error("Unknown pod name: %s" % pod_name)
128                 return
129 
130             self._pods_phase[worker_id] = (pod_name, phase)
131             if evt_type == "DELETED":
132                 del self._pods_phase[worker_id]
133                 del self._pod_name_to_id[pod_name]
134                 self._task_q.recover_tasks(worker_id)
135 
136                 # If the pod being deleted was not "Succeeded", relaunch a worker.
137                 relaunch = self._relaunch_deleted_live_worker and phase != "Succeeded"
138         if relaunch:
139             self._logger.info("Relaunching worker.")
140             self._start_worker(self._next_worker_id())
141
```

Path: `elasticdl/python/elasticdl/master/k8s_client.py`
Content:
```
1 import logging
2 import os
3 import threading
4 import traceback
5 
6 from kubernetes import client, config, watch
7 
8 
9 class Client(object):
10     def __init__(
11         self, *, worker_image, namespace, job_name, event_callback
12     ):
13         """
14         ElasticDL k8s client.
15 
16         Args:
17             worker_image: Docker image path for ElasticDL workers.
18             namespace: k8s namespace for ElasticDL pods.
19             job_name: ElasticDL job name, should be unique in the namespace.
20                 Used as worker pod name prefix and value for "elasticdl" label.
21             event_callback: If not None, an event watcher will be created and
22                 events passed to the callback.
23         """
24         if os.getenv("KUBERNETES_SERVICE_HOST"):
25             # We are running inside k8s
26             config.load_incluster_config()
27         else:
28             # Use user's kube config
29             config.load_kube_config()
30 
31         self._v1 = client.CoreV1Api()
32         self._logger = logging.getLogger(__name__)
33         self._image = worker_image
34         self._ns = namespace
35         self._job_name = job_name
36         self._event_cb = event_callback
37         if self._event_cb:
38             threading.Thread(
39                 target=self._watch, name="event_watcher", daemon=True
40             ).start()
41 
42     def _watch(self):
43         stream = watch.Watch().stream(
44             self._v1.list_namespaced_pod,
45             self._ns,
46             label_selector="elasticdl_job_name=" + self._job_name,
47         )
48         for event in stream:
49             try:
50                 self._event_cb(event)
51             except Exception:
52                 traceback.print_exc()
53 
54     def get_pod_name(self, worker_id):
55         return "elasticdl-worker-" + self._job_name + "-" + str(worker_id)
56 
57     def _create_worker_pod(self, worker_id, resource_requests, resource_limits, priority,
58                            mount_path, volume_name, image_pull_policy, command, args, restart_policy):
59         # Worker container config
60         container = client.V1Container(
61             name=self.get_pod_name(worker_id),
62             image=self._image,
63             command=command,
64             resources=client.V1ResourceRequirements(
65                 requests=resource_requests,
66                 limits=resource_limits
67             ),
68             image_pull_policy=image_pull_policy,
69             args=args
70         )
71 
72         # Pod
73         spec = client.V1PodSpec(
74             containers=[container],
75             restart_policy=restart_policy,
76         )
77 
78         # Mount data path
79         if mount_path is not None and volume_name is not None:
80             volume = client.V1Volume(
81                 name='data-volume',
82                 persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
83                     claim_name="fileserver-claim", read_only=False))
84             spec.volumes = [volume]
85             container.volume_mounts = [client.V1VolumeMount(name=volume_name, mount_path=mount_path)]
86 
87         if priority is not None:
88             spec.priority_class_name = priority
89 
90         pod = client.V1Pod(
91             spec=spec,
92             metadata=client.V1ObjectMeta(
93                 name=self.get_pod_name(worker_id),
94                 labels={
95                     "app": "elasticdl",
96                     "elasticdl_job_name": self._job_name
97                 },
98             ),
99         )
100         return pod
101 
102     def create_worker(self, worker_id, resource_requests, resource_limits, priority=None,
103                       mount_path=None, volume_name=None, image_pull_policy=None,
104                       command=None, args=None, restart_policy="OnFailure"):
105         self._logger.info("Creating worker: " + str(worker_id))
106         pod = self._create_worker_pod(
107             worker_id, resource_requests, resource_limits, priority,
108             mount_path, volume_name, image_pull_policy, command=command,
109             args=args, restart_policy=restart_policy)
110         return self._v1.create_namespaced_pod(self._ns, pod)
111 
112     def delete_worker(self, worker_id):
113         self._logger.info("Deleting worker: " + str(worker_id))
114         self._v1.delete_namespaced_pod(
115             self.get_pod_name(worker_id),
116             self._ns,
117             body=client.V1DeleteOptions(grace_period_seconds=0),
118         )
119
```
--- 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/elasticdl/master/k8s_client.py b/elasticdl/python/elasticdl/master/k8s_client.py
--- a/elasticdl/python/elasticdl/master/k8s_client.py
+++ b/elasticdl/python/elasticdl/master/k8s_client.py
@@ -5,6 +5,8 @@
 
 from kubernetes import client, config, watch
 
+WORKER_POD_NAME_PREFIX = "elasticdl-worker-"
+
 
 class Client(object):
     def __init__(
@@ -51,14 +53,14 @@
             except Exception:
                 traceback.print_exc()
 
-    def get_pod_name(self, worker_id):
-        return "elasticdl-worker-" + self._job_name + "-" + str(worker_id)
+    def get_worker_pod_name(self, worker_id):
+        return WORKER_POD_NAME_PREFIX + self._job_name + "-" + str(worker_id)
 
     def _create_worker_pod(self, worker_id, resource_requests, resource_limits, priority,
                            mount_path, volume_name, image_pull_policy, command, args, restart_policy):
         # Worker container config
         container = client.V1Container(
-            name=self.get_pod_name(worker_id),
+            name=self.get_worker_pod_name(worker_id),
             image=self._image,
             command=command,
             resources=client.V1ResourceRequirements(
@@ -90,7 +92,7 @@
         pod = client.V1Pod(
             spec=spec,
             metadata=client.V1ObjectMeta(
-                name=self.get_pod_name(worker_id),
+                name=self.get_worker_pod_name(worker_id),
                 labels={
                     "app": "elasticdl",
                     "elasticdl_job_name": self._job_name
@@ -112,7 +114,7 @@
     def delete_worker(self, worker_id):
         self._logger.info("Deleting worker: " + str(worker_id))
         self._v1.delete_namespaced_pod(
-            self.get_pod_name(worker_id),
+            self.get_worker_pod_name(worker_id),
             self._ns,
             body=client.V1DeleteOptions(grace_period_seconds=0),
         )
diff --git a/elasticdl/python/elasticdl/master/k8s_worker_manager.py b/elasticdl/python/elasticdl/master/k8s_worker_manager.py
--- a/elasticdl/python/elasticdl/master/k8s_worker_manager.py
+++ b/elasticdl/python/elasticdl/master/k8s_worker_manager.py
@@ -123,8 +123,8 @@
         relaunch = False
         with self._lock:
             worker_id = self._pod_name_to_id.get(pod_name)
-            if worker_id is None:
-                self._logger.error("Unknown pod name: %s" % pod_name)
+            if worker_id is None and k8s.WORKER_POD_NAME_PREFIX in pod_name:
+                self._logger.error("Unknown worker pod name: %s" % pod_name)
                 return
 
             self._pods_phase[worker_id] = (pod_name, phase)

{"golden_diff": "diff --git a/elasticdl/python/elasticdl/master/k8s_client.py b/elasticdl/python/elasticdl/master/k8s_client.py\n--- a/elasticdl/python/elasticdl/master/k8s_client.py\n+++ b/elasticdl/python/elasticdl/master/k8s_client.py\n@@ -5,6 +5,8 @@\n \n from kubernetes import client, config, watch\n \n+WORKER_POD_NAME_PREFIX = \"elasticdl-worker-\"\n+\n \n class Client(object):\n     def __init__(\n@@ -51,14 +53,14 @@\n             except Exception:\n                 traceback.print_exc()\n \n-    def get_pod_name(self, worker_id):\n-        return \"elasticdl-worker-\" + self._job_name + \"-\" + str(worker_id)\n+    def get_worker_pod_name(self, worker_id):\n+        return WORKER_POD_NAME_PREFIX + self._job_name + \"-\" + str(worker_id)\n \n     def _create_worker_pod(self, worker_id, resource_requests, resource_limits, priority,\n                            mount_path, volume_name, image_pull_policy, command, args, restart_policy):\n         # Worker container config\n         container = client.V1Container(\n-            name=self.get_pod_name(worker_id),\n+            name=self.get_worker_pod_name(worker_id),\n             image=self._image,\n             command=command,\n             resources=client.V1ResourceRequirements(\n@@ -90,7 +92,7 @@\n         pod = client.V1Pod(\n             spec=spec,\n             metadata=client.V1ObjectMeta(\n-                name=self.get_pod_name(worker_id),\n+                name=self.get_worker_pod_name(worker_id),\n                 labels={\n                     \"app\": \"elasticdl\",\n                     \"elasticdl_job_name\": self._job_name\n@@ -112,7 +114,7 @@\n     def delete_worker(self, worker_id):\n         self._logger.info(\"Deleting worker: \" + str(worker_id))\n         self._v1.delete_namespaced_pod(\n-            self.get_pod_name(worker_id),\n+            self.get_worker_pod_name(worker_id),\n             self._ns,\n             body=client.V1DeleteOptions(grace_period_seconds=0),\n         )\ndiff --git a/elasticdl/python/elasticdl/master/k8s_worker_manager.py b/elasticdl/python/elasticdl/master/k8s_worker_manager.py\n--- a/elasticdl/python/elasticdl/master/k8s_worker_manager.py\n+++ b/elasticdl/python/elasticdl/master/k8s_worker_manager.py\n@@ -123,8 +123,8 @@\n         relaunch = False\n         with self._lock:\n             worker_id = self._pod_name_to_id.get(pod_name)\n-            if worker_id is None:\n-                self._logger.error(\"Unknown pod name: %s\" % pod_name)\n+            if worker_id is None and k8s.WORKER_POD_NAME_PREFIX in pod_name:\n+                self._logger.error(\"Unknown worker pod name: %s\" % pod_name)\n                 return\n \n             self._pods_phase[worker_id] = (pod_name, phase)\n", "issue": "Event callback should not check the existence of master pod\nCurrently, there's error log when event callback checks the pod existence like this `Unknown pod name: elasticdl-master-test-mnist`. This is because `elasticdl-master-test-mnist` is the master pod that's created by users instead of being created by `WorkerManager`. We should fix this since it should only check pod existence for worker pod.\r\n\r\nComplete log:\r\n```\r\n2019-06-06 17:38:37,952 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:121] Got event ADDED, phase Running for pod: elasticdl-master-test-mnist\r\n2019-06-06 17:38:37,953 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:71] Starting worker: 1\r\n2019-06-06 17:38:37,953 elasticdl.python.elasticdl.master.k8s_worker_manager ERROR    [k8s_worker_manager.py:127] Unknown pod name: elasticdl-master-test-mnist\r\n2019-06-06 17:38:37,954 elasticdl.python.elasticdl.master.k8s_client INFO     [k8s_client.py:105] Creating worker: 1\r\n2019-06-06 17:38:37,955 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:121] Got event ADDED, phase Pending for pod: elasticdl-worker-test-mnist-0\r\n```\nEvent callback should not check the existence of master pod\nCurrently, there's error log when event callback checks the pod existence like this `Unknown pod name: elasticdl-master-test-mnist`. This is because `elasticdl-master-test-mnist` is the master pod that's created by users instead of being created by `WorkerManager`. We should fix this since it should only check pod existence for worker pod.\r\n\r\nComplete log:\r\n```\r\n2019-06-06 17:38:37,952 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:121] Got event ADDED, phase Running for pod: elasticdl-master-test-mnist\r\n2019-06-06 17:38:37,953 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:71] Starting worker: 1\r\n2019-06-06 17:38:37,953 elasticdl.python.elasticdl.master.k8s_worker_manager ERROR    [k8s_worker_manager.py:127] Unknown pod name: elasticdl-master-test-mnist\r\n2019-06-06 17:38:37,954 elasticdl.python.elasticdl.master.k8s_client INFO     [k8s_client.py:105] Creating worker: 1\r\n2019-06-06 17:38:37,955 elasticdl.python.elasticdl.master.k8s_worker_manager INFO     [k8s_worker_manager.py:121] Got event ADDED, phase Pending for pod: elasticdl-worker-test-mnist-0\r\n```\n", "before_files": [{"content": "import itertools\nimport logging\nimport threading\n\nfrom collections import Counter\nfrom elasticdl.python.elasticdl.master import k8s_client as k8s\n\n\nclass WorkerManager(object):\n    def __init__(\n            self,\n            task_q,\n            command,\n            args,\n            num_worker=1,\n            cpu_request=\"1000m\",\n            cpu_limit=\"1000m\",\n            memory_request=\"4096Mi\",\n            memory_limit=\"4096Mi\",\n            pod_priority=None,\n            mount_path=None,\n            volume_name=None,\n            image_pull_policy=None,\n            restart_policy=\"OnFailure\",\n            **kwargs):\n        self._logger = logging.getLogger(__name__)\n        self._command = command\n        self._args = args\n        self._num_worker = num_worker\n        self._resource_requests = {\n            \"cpu\": cpu_request,\n            \"memory\": memory_request\n        }\n        self._resource_limits = {\n            \"cpu\": cpu_limit,\n            \"memory\": memory_limit\n        }\n        self._restart_policy = restart_policy\n        self._pod_priority = pod_priority\n        self._mount_path = mount_path\n        self._volume_name = volume_name\n        self._image_pull_policy = image_pull_policy\n        self._task_q = task_q\n        self._next_worker_id = itertools.count().__next__\n\n        # protects followed variables, which are accessed from event_cb.\n        self._lock = threading.Lock()\n        # worker id to (pod name, phase) mapping\n        # phase: None/Pending/Running/Succeeded/Failed/Unknown\n        #   None: worker was just launched, haven't received event yet.\n        #   Pending: worker pod not started yet\n        #   Running: worker pod is running\n        #   Succeeded: worker pod finishes all tasks and terminates with\n        #       no issue.\n        #   Failed: worker pod is killed for some reason\n        #   Unknown: unknown\n        self._pods_phase = {}\n        # pod name to worker id mapping\n        self._pod_name_to_id = {}\n\n        self._relaunch_deleted_live_worker = True\n\n        self._k8s_client = k8s.Client(\n            event_callback=self._event_cb, **kwargs\n        )\n\n    def set_relaunch_deleted_live_worker(self, val):\n        self._relaunch_deleted_live_worker = bool(val)\n\n    def _start_worker(self, worker_id):\n        self._logger.info(\"Starting worker: %d\" % worker_id)\n        with self._lock:\n            pod = self._k8s_client.create_worker(\n                worker_id,\n                self._resource_requests,\n                self._resource_limits,\n                self._pod_priority,\n                self._mount_path,\n                self._volume_name,\n                self._image_pull_policy,\n                command=self._command,\n                args=self._args + [\"--worker_id\", str(worker_id)],\n                restart_policy=self._restart_policy,\n            )\n            name = pod.metadata.name\n            self._pod_name_to_id[name] = worker_id\n            self._pods_phase[worker_id] = (name, None)\n\n    def start_workers(self):\n        for i in range(self._num_worker):\n            self._start_worker(self._next_worker_id())\n\n    def _remove_worker(self, worker_id):\n        with self._lock:\n            if worker_id not in self._pods_phase:\n                self._logger.error(\"Unknown worker id: %s\" % worker_id)\n                return\n\n        # TODO: change _k8s_client to accept pod name instead of worker id.\n        self._k8s_client.delete_worker(worker_id)\n\n    def stop_relaunch_and_remove_workers(self):\n        with self._lock:\n            self._relaunch_deleted_live_worker = False\n            for worker_id in self._pods_phase:\n                self._k8s_client.delete_worker(worker_id)\n\n    def get_counters(self):\n        with self._lock:\n            return Counter([v for _, v in self._pods_phase.values()])\n\n    def _event_cb(self, event):\n        evt_obj = event.get(\"object\")\n        evt_type = event.get(\"type\")\n        if not evt_obj or not evt_type:\n            self._logger.error(\"Event doesn't have object or type: %s\" % event)\n            return\n\n        pod_name = evt_obj.metadata.name\n        phase = evt_obj.status.phase\n        self._logger.info(\"Got event %s, phase %s for pod: %s\" % (evt_type, phase, pod_name))\n\n        relaunch = False\n        with self._lock:\n            worker_id = self._pod_name_to_id.get(pod_name)\n            if worker_id is None:\n                self._logger.error(\"Unknown pod name: %s\" % pod_name)\n                return\n\n            self._pods_phase[worker_id] = (pod_name, phase)\n            if evt_type == \"DELETED\":\n                del self._pods_phase[worker_id]\n                del self._pod_name_to_id[pod_name]\n                self._task_q.recover_tasks(worker_id)\n\n                # If the pod being deleted was not \"Succeeded\", relaunch a worker.\n                relaunch = self._relaunch_deleted_live_worker and phase != \"Succeeded\"\n        if relaunch:\n            self._logger.info(\"Relaunching worker.\")\n            self._start_worker(self._next_worker_id())\n", "path": "elasticdl/python/elasticdl/master/k8s_worker_manager.py"}, {"content": "import logging\nimport os\nimport threading\nimport traceback\n\nfrom kubernetes import client, config, watch\n\n\nclass Client(object):\n    def __init__(\n        self, *, worker_image, namespace, job_name, event_callback\n    ):\n        \"\"\"\n        ElasticDL k8s client.\n\n        Args:\n            worker_image: Docker image path for ElasticDL workers.\n            namespace: k8s namespace for ElasticDL pods.\n            job_name: ElasticDL job name, should be unique in the namespace.\n                Used as worker 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._v1 = client.CoreV1Api()\n        self._logger = logging.getLogger(__name__)\n        self._image = worker_image\n        self._ns = namespace\n        self._job_name = job_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\n    def _watch(self):\n        stream = watch.Watch().stream(\n            self._v1.list_namespaced_pod,\n            self._ns,\n            label_selector=\"elasticdl_job_name=\" + 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_pod_name(self, worker_id):\n        return \"elasticdl-worker-\" + self._job_name + \"-\" + str(worker_id)\n\n    def _create_worker_pod(self, worker_id, resource_requests, resource_limits, priority,\n                           mount_path, volume_name, image_pull_policy, command, args, restart_policy):\n        # Worker container config\n        container = client.V1Container(\n            name=self.get_pod_name(worker_id),\n            image=self._image,\n            command=command,\n            resources=client.V1ResourceRequirements(\n                requests=resource_requests,\n                limits=resource_limits\n            ),\n            image_pull_policy=image_pull_policy,\n            args=args\n        )\n\n        # Pod\n        spec = client.V1PodSpec(\n            containers=[container],\n            restart_policy=restart_policy,\n        )\n\n        # Mount data path\n        if mount_path is not None and volume_name is not None:\n            volume = client.V1Volume(\n                name='data-volume',\n                persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(\n                    claim_name=\"fileserver-claim\", read_only=False))\n            spec.volumes = [volume]\n            container.volume_mounts = [client.V1VolumeMount(name=volume_name, mount_path=mount_path)]\n\n        if priority is not None:\n            spec.priority_class_name = priority\n\n        pod = client.V1Pod(\n            spec=spec,\n            metadata=client.V1ObjectMeta(\n                name=self.get_pod_name(worker_id),\n                labels={\n                    \"app\": \"elasticdl\",\n                    \"elasticdl_job_name\": self._job_name\n                },\n            ),\n        )\n        return pod\n\n    def create_worker(self, worker_id, resource_requests, resource_limits, priority=None,\n                      mount_path=None, volume_name=None, image_pull_policy=None,\n                      command=None, args=None, restart_policy=\"OnFailure\"):\n        self._logger.info(\"Creating worker: \" + str(worker_id))\n        pod = self._create_worker_pod(\n            worker_id, resource_requests, resource_limits, priority,\n            mount_path, volume_name, image_pull_policy, command=command,\n            args=args, restart_policy=restart_policy)\n        return self._v1.create_namespaced_pod(self._ns, pod)\n\n    def delete_worker(self, worker_id):\n        self._logger.info(\"Deleting worker: \" + str(worker_id))\n        self._v1.delete_namespaced_pod(\n            self.get_pod_name(worker_id),\n            self._ns,\n            body=client.V1DeleteOptions(grace_period_seconds=0),\n        )\n", "path": "elasticdl/python/elasticdl/master/k8s_client.py"}], "after_files": [{"content": "import itertools\nimport logging\nimport threading\n\nfrom collections import Counter\nfrom elasticdl.python.elasticdl.master import k8s_client as k8s\n\n\nclass WorkerManager(object):\n    def __init__(\n            self,\n            task_q,\n            command,\n            args,\n            num_worker=1,\n            cpu_request=\"1000m\",\n            cpu_limit=\"1000m\",\n            memory_request=\"4096Mi\",\n            memory_limit=\"4096Mi\",\n            pod_priority=None,\n            mount_path=None,\n            volume_name=None,\n            image_pull_policy=None,\n            restart_policy=\"OnFailure\",\n            **kwargs):\n        self._logger = logging.getLogger(__name__)\n        self._command = command\n        self._args = args\n        self._num_worker = num_worker\n        self._resource_requests = {\n            \"cpu\": cpu_request,\n            \"memory\": memory_request\n        }\n        self._resource_limits = {\n            \"cpu\": cpu_limit,\n            \"memory\": memory_limit\n        }\n        self._restart_policy = restart_policy\n        self._pod_priority = pod_priority\n        self._mount_path = mount_path\n        self._volume_name = volume_name\n        self._image_pull_policy = image_pull_policy\n        self._task_q = task_q\n        self._next_worker_id = itertools.count().__next__\n\n        # protects followed variables, which are accessed from event_cb.\n        self._lock = threading.Lock()\n        # worker id to (pod name, phase) mapping\n        # phase: None/Pending/Running/Succeeded/Failed/Unknown\n        #   None: worker was just launched, haven't received event yet.\n        #   Pending: worker pod not started yet\n        #   Running: worker pod is running\n        #   Succeeded: worker pod finishes all tasks and terminates with\n        #       no issue.\n        #   Failed: worker pod is killed for some reason\n        #   Unknown: unknown\n        self._pods_phase = {}\n        # pod name to worker id mapping\n        self._pod_name_to_id = {}\n\n        self._relaunch_deleted_live_worker = True\n\n        self._k8s_client = k8s.Client(\n            event_callback=self._event_cb, **kwargs\n        )\n\n    def set_relaunch_deleted_live_worker(self, val):\n        self._relaunch_deleted_live_worker = bool(val)\n\n    def _start_worker(self, worker_id):\n        self._logger.info(\"Starting worker: %d\" % worker_id)\n        with self._lock:\n            pod = self._k8s_client.create_worker(\n                worker_id,\n                self._resource_requests,\n                self._resource_limits,\n                self._pod_priority,\n                self._mount_path,\n                self._volume_name,\n                self._image_pull_policy,\n                command=self._command,\n                args=self._args + [\"--worker_id\", str(worker_id)],\n                restart_policy=self._restart_policy,\n            )\n            name = pod.metadata.name\n            self._pod_name_to_id[name] = worker_id\n            self._pods_phase[worker_id] = (name, None)\n\n    def start_workers(self):\n        for i in range(self._num_worker):\n            self._start_worker(self._next_worker_id())\n\n    def _remove_worker(self, worker_id):\n        with self._lock:\n            if worker_id not in self._pods_phase:\n                self._logger.error(\"Unknown worker id: %s\" % worker_id)\n                return\n\n        # TODO: change _k8s_client to accept pod name instead of worker id.\n        self._k8s_client.delete_worker(worker_id)\n\n    def stop_relaunch_and_remove_workers(self):\n        with self._lock:\n            self._relaunch_deleted_live_worker = False\n            for worker_id in self._pods_phase:\n                self._k8s_client.delete_worker(worker_id)\n\n    def get_counters(self):\n        with self._lock:\n            return Counter([v for _, v in self._pods_phase.values()])\n\n    def _event_cb(self, event):\n        evt_obj = event.get(\"object\")\n        evt_type = event.get(\"type\")\n        if not evt_obj or not evt_type:\n            self._logger.error(\"Event doesn't have object or type: %s\" % event)\n            return\n\n        pod_name = evt_obj.metadata.name\n        phase = evt_obj.status.phase\n        self._logger.info(\"Got event %s, phase %s for pod: %s\" % (evt_type, phase, pod_name))\n\n        relaunch = False\n        with self._lock:\n            worker_id = self._pod_name_to_id.get(pod_name)\n            if worker_id is None and k8s.WORKER_POD_NAME_PREFIX in pod_name:\n                self._logger.error(\"Unknown worker pod name: %s\" % pod_name)\n                return\n\n            self._pods_phase[worker_id] = (pod_name, phase)\n            if evt_type == \"DELETED\":\n                del self._pods_phase[worker_id]\n                del self._pod_name_to_id[pod_name]\n                self._task_q.recover_tasks(worker_id)\n\n                # If the pod being deleted was not \"Succeeded\", relaunch a worker.\n                relaunch = self._relaunch_deleted_live_worker and phase != \"Succeeded\"\n        if relaunch:\n            self._logger.info(\"Relaunching worker.\")\n            self._start_worker(self._next_worker_id())\n", "path": "elasticdl/python/elasticdl/master/k8s_worker_manager.py"}, {"content": "import logging\nimport os\nimport threading\nimport traceback\n\nfrom kubernetes import client, config, watch\n\nWORKER_POD_NAME_PREFIX = \"elasticdl-worker-\"\n\n\nclass Client(object):\n    def __init__(\n        self, *, worker_image, namespace, job_name, event_callback\n    ):\n        \"\"\"\n        ElasticDL k8s client.\n\n        Args:\n            worker_image: Docker image path for ElasticDL workers.\n            namespace: k8s namespace for ElasticDL pods.\n            job_name: ElasticDL job name, should be unique in the namespace.\n                Used as worker 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._v1 = client.CoreV1Api()\n        self._logger = logging.getLogger(__name__)\n        self._image = worker_image\n        self._ns = namespace\n        self._job_name = job_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\n    def _watch(self):\n        stream = watch.Watch().stream(\n            self._v1.list_namespaced_pod,\n            self._ns,\n            label_selector=\"elasticdl_job_name=\" + 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_worker_pod_name(self, worker_id):\n        return WORKER_POD_NAME_PREFIX + self._job_name + \"-\" + str(worker_id)\n\n    def _create_worker_pod(self, worker_id, resource_requests, resource_limits, priority,\n                           mount_path, volume_name, image_pull_policy, command, args, restart_policy):\n        # Worker container config\n        container = client.V1Container(\n            name=self.get_worker_pod_name(worker_id),\n            image=self._image,\n            command=command,\n            resources=client.V1ResourceRequirements(\n                requests=resource_requests,\n                limits=resource_limits\n            ),\n            image_pull_policy=image_pull_policy,\n            args=args\n        )\n\n        # Pod\n        spec = client.V1PodSpec(\n            containers=[container],\n            restart_policy=restart_policy,\n        )\n\n        # Mount data path\n        if mount_path is not None and volume_name is not None:\n            volume = client.V1Volume(\n                name='data-volume',\n                persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(\n                    claim_name=\"fileserver-claim\", read_only=False))\n            spec.volumes = [volume]\n            container.volume_mounts = [client.V1VolumeMount(name=volume_name, mount_path=mount_path)]\n\n        if priority is not None:\n            spec.priority_class_name = priority\n\n        pod = client.V1Pod(\n            spec=spec,\n            metadata=client.V1ObjectMeta(\n                name=self.get_worker_pod_name(worker_id),\n                labels={\n                    \"app\": \"elasticdl\",\n                    \"elasticdl_job_name\": self._job_name\n                },\n            ),\n        )\n        return pod\n\n    def create_worker(self, worker_id, resource_requests, resource_limits, priority=None,\n                      mount_path=None, volume_name=None, image_pull_policy=None,\n                      command=None, args=None, restart_policy=\"OnFailure\"):\n        self._logger.info(\"Creating worker: \" + str(worker_id))\n        pod = self._create_worker_pod(\n            worker_id, resource_requests, resource_limits, priority,\n            mount_path, volume_name, image_pull_policy, command=command,\n            args=args, restart_policy=restart_policy)\n        return self._v1.create_namespaced_pod(self._ns, pod)\n\n    def delete_worker(self, worker_id):\n        self._logger.info(\"Deleting worker: \" + str(worker_id))\n        self._v1.delete_namespaced_pod(\n            self.get_worker_pod_name(worker_id),\n            self._ns,\n            body=client.V1DeleteOptions(grace_period_seconds=0),\n        )\n", "path": "elasticdl/python/elasticdl/master/k8s_client.py"}]}