Fix the bug if the host of a new worker is the same as an old worker (#2496)

* Fix the bug if the host of a new worker is the same as an old worker

* Do not call on_pod_deleted if the status is from FAILED to DELETED

Author: workingloong

elasticdl/python/master/pod_manager.py

@@ -586,7 +586,10 @@ def _event_cb(self, event):
                 ):
                     self._worker_info[pod_id].inc_relaunch_count()
                     should_relaunch = True
-        elif matched_pod_state_flow.to_status == PodStatus.DELETED:
+        elif (
+            matched_pod_state_flow.from_status != PodStatus.FAILED
+            and matched_pod_state_flow.to_status == PodStatus.DELETED
+        ):
             [
                 callback.on_pod_deleted(pod_info, cluster_context)
                 for callback in self._pod_event_callbacks

elasticdl/python/master/rendezvous_server.py

@@ -14,6 +14,8 @@
 import time
 from threading import Lock
 
+from elasticai_api.util.log_utils import default_logger as logger
+
 try:
     from horovod.runner.common.util.hosts import (
         get_host_assignments,
@@ -78,6 +80,11 @@ def start(self):
         self._rendezvous_port = self._rendezvous_server.start()
 
     def _init_rendezvous_server(self):
+        logger.info(
+            "Initialize rendezvous server with hosts {}".format(
+                self._next_rendezvous_hosts
+            )
+        )
         self._cur_rendezvous_hosts = self._next_rendezvous_hosts
         self._next_rendezvous_hosts = None
         host_alloc_plan = self._get_host_plan()
@@ -128,7 +135,12 @@ def get_rendezvous_id(self):
 
     def add_worker(self, worker_host):
         with self._lock:
-            if worker_host and worker_host not in self._cur_rendezvous_hosts:
+            logger.info(
+                "Add worker host {} into rendenzvous and cur hosts {}.".format(
+                    worker_host, self._cur_rendezvous_hosts
+                )
+            )
+            if worker_host:
                 if self._next_rendezvous_hosts is None:
                     self._next_rendezvous_hosts = copy.deepcopy(
                         self._cur_rendezvous_hosts
@@ -137,6 +149,9 @@ def add_worker(self, worker_host):
 
     def remove_worker(self, worker_host):
         with self._lock:
+            logger.info(
+                "Remove worker host {} from rendenzvous.".format(worker_host)
+            )
             if worker_host in self._cur_rendezvous_hosts:
                 if self._next_rendezvous_hosts is None:
                     self._next_rendezvous_hosts = copy.deepcopy(