fixed load/users getting distributed to missing worker

locust/dispatch.py

@@ -16,6 +16,8 @@
 if TYPE_CHECKING:
     from locust.runners import WorkerNode
 
+HEARTBEAT_DEAD = -10
+
 
 # To profile line-by-line, uncomment the code below (i.e. `import line_profiler ...`) and
 # place `@profile` on the functions/methods you wish to profile. Then, in the unit test you are
@@ -201,6 +203,13 @@ def remove_worker(self, worker_node: "WorkerNode") -> None:
             return
         self._prepare_rebalance()
 
+    def prepare_rebalance(self) -> None:
+        """
+        This method will remove missing workers after the HEARTBEAT_DEAD interval has passed from getting users during distribution
+        """
+        self._worker_nodes = [w for w in self._worker_nodes if w.heartbeat > HEARTBEAT_DEAD]
+        self._prepare_rebalance()
+
     def _prepare_rebalance(self) -> None:
         """
         When a rebalance is required because of added and/or removed workers, we compute the desired state as if

locust/runners.py

@@ -59,6 +59,7 @@
 CPU_MONITOR_INTERVAL = 5.0
 HEARTBEAT_INTERVAL = 1
 HEARTBEAT_LIVENESS = 3
+HEARTBEAT_DEAD = -10
 FALLBACK_INTERVAL = 5
 
 
@@ -846,6 +847,10 @@ def heartbeat_worker(self):
                 continue
 
             for client in self.clients.all:
+                if client.heartbeat < HEARTBEAT_DEAD and client.state == STATE_MISSING:
+                    if self._users_dispatcher is not None:
+                        logger.info("Worker %s failed to send heartbeat after being marked missing for long time, removing the worker" % str(client.id))
+                        self._users_dispatcher.prepare_rebalance()
                 if client.heartbeat < 0 and client.state != STATE_MISSING:
                     logger.info("Worker %s failed to send heartbeat, setting state to missing." % str(client.id))
                     client.state = STATE_MISSING

locust/test/test_dispatch.py

@@ -4,7 +4,7 @@
 from typing import Dict, List, Tuple
 
 from locust import User
-from locust.dispatch import UsersDispatcher
+from locust.dispatch import HEARTBEAT_DEAD, UsersDispatcher
 from locust.runners import WorkerNode
 from locust.test.util import clear_all_functools_lru_cache
 
@@ -3273,6 +3273,29 @@ class User5(User):
             user_classes=[User1, User2, User3, User4, User5],
         )
 
+    def test_remove_missing_workers_with_dead_heartbeat(self):
+        class User1(User):
+            weight = 1
+
+        class User2(User):
+            weight = 1
+
+        class User3(User):
+            weight = 1
+
+        user_classes = [User1, User2, User3]
+
+        worker_nodes = [WorkerNode(str(i + 1)) for i in range(3)]
+        worker_nodes[1].heartbeat = HEARTBEAT_DEAD - 1
+
+        users_dispatcher = UsersDispatcher(worker_nodes=worker_nodes, user_classes=user_classes)
+
+        users_dispatcher.new_dispatch(target_user_count=9, spawn_rate=3)
+        users_dispatcher._wait_between_dispatch = 0
+        users_dispatcher.prepare_rebalance()
+        list(users_dispatcher)
+        self.assertIs(len(users_dispatcher._worker_nodes), 2, "Expected worker 2 will be removed from worker nodes")
+
     def test_ramp_up_partially_ramp_down_and_rump_up_to_target(self):
         class User1(User):
             fixed_count = 50