Truncate master and worker logs

locustio · cyberw · Jun 7, 2024 · Jun 3, 2024 · Jun 3, 2024 · Jun 4, 2024
commit 52fd1a3ce14fab627db5d3121f3ce8eda0e2acbb
diff --git a/locust/log.py b/locust/log.py
@@ -2,6 +2,7 @@
 import logging.config
 import re
 import socket
+from collections import deque
 
 HOSTNAME = re.sub(r"\..*", "", socket.gethostname())
 
@@ -13,7 +14,7 @@
 class LogReader(logging.Handler):
     def __init__(self):
         super().__init__()
-        self.logs = []
+        self.logs = deque(maxlen=500)
 
     def emit(self, record):
         self.logs.append(self.format(record))
@@ -76,10 +77,10 @@ def setup_logging(loglevel, logfile=None):
 
 
 def get_logs():
-    log_reader_handler = [handler for handler in logging.getLogger("root").handlers if handler.name == "log_reader"][0]
+    log_reader_handler = [handler for handler in logging.getLogger("root").handlers if handler.name == "log_reader"]
 
     if log_reader_handler:
-        return log_reader_handler.logs
+        return list(log_reader_handler[0].logs)
 
     return []
 

diff --git a/locust/runners.py b/locust/runners.py
@@ -1422,9 +1422,20 @@ def stats_reporter(self) -> NoReturn:
                 logger.error(f"Temporary connection lost to master server: {e}, will retry later.")
             gevent.sleep(WORKER_REPORT_INTERVAL)
 
-    def logs_reporter(self) -> NoReturn:
+    def logs_reporter(self) -> None:
         while True:
-            self._send_logs()
+            current_logs = get_logs()
+
+            if (len(current_logs) - len(self.logs)) > 10:
+                logger.warning(
+                    "The worker attempted to send more than 10 log lines in one interval. Further log sending was disabled for this worker."
+                )
+                self._send_logs(get_logs())
+                break
+            if len(current_logs) > len(self.logs):
+                self._send_logs(current_logs)
+
+            self.logs = current_logs
             gevent.sleep(WORKER_LOG_REPORT_INTERVAL)
 
     def send_message(self, msg_type: str, data: dict[str, Any] | None = None, client_id: str | None = None) -> None:
@@ -1443,12 +1454,8 @@ def _send_stats(self) -> None:
         self.environment.events.report_to_master.fire(client_id=self.client_id, data=data)
         self.client.send(Message("stats", data, self.client_id))
 
-    def _send_logs(self) -> None:
-        current_logs = get_logs()
-
-        if len(current_logs) > len(self.logs):
-            self.logs = current_logs
-            self.send_message("logs", {"worker_id": self.client_id, "logs": current_logs})
+    def _send_logs(self, current_logs) -> None:
+        self.send_message("logs", {"worker_id": self.client_id, "logs": current_logs})
 
     def connect_to_master(self):
         self.retry += 1

diff --git a/locust/test/test_runners.py b/locust/test/test_runners.py
@@ -3992,6 +3992,40 @@ def my_task(self):
             self.assertEqual(worker.client_id, client.outbox[3].data.get("worker_id"))
             worker.quit()
 
+    def test_quit_worker_logs(self):
+        class MyUser(User):
+            wait_time = constant(1)
+
+            @task
+            def my_task(self):
+                pass
+
+        with mock.patch("locust.rpc.rpc.Client", mocked_rpc()) as client:
+            short_time = 0.05
+
+            log_handler = LogReader()
+            log_handler.name = "log_reader"
+            log_handler.setLevel(logging.INFO)
+            logger = logging.getLogger("root")
+            logger.addHandler(log_handler)
+            log_line = "spamming log"
+
+            for _ in range(11):
+                logger.info(log_line)
+
+            worker = self.get_runner(environment=Environment(), user_classes=[MyUser], client=client)
+
+            gevent.sleep(short_time)
+
+            self.assertEqual("logs", client.outbox[3].type)
+            self.assertEqual(
+                "The worker attempted to send more than 10 log lines in one interval. Further log sending was disabled for this worker.",
+                client.outbox[3].data.get("logs", [])[0],
+            )
+            self.assertEqual(worker.client_id, client.outbox[3].data.get("worker_id"))
+            worker.quit()
+            logger.removeHandler(log_handler)
+
 
 class TestMessageSerializing(unittest.TestCase):
     def test_message_serialize(self):