Skip to content

Commit

Permalink
Fixing issue #1961 with incorrect "All users spawned" log messages wh…
Browse files Browse the repository at this point in the history 
…en running distributed by increasing wait time.

Now if the timeout of worker reports is expired, log the message about it.
  • Loading branch information
@EzR1d3r
EzR1d3r committed Jan 26, 2022
1 parent 7bff126 commit aa256cf
Show file tree
Hide file tree
Showing 4 changed files with 135 additions and 26 deletions.
15 changes: 9 additions & 6 deletions locust/runners.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -749,32 +749,35 @@ def start(self, user_count: int, spawn_rate: float, wait=False) -> None:
# when the user count is really at the desired value.
timeout = gevent.Timeout(self._wait_for_workers_report_after_ramp_up())
timeout.start()
msg_prefix = "All users spawned"
try:
while self.user_count != self.target_user_count:
gevent.sleep()
gevent.sleep(0.01)
except gevent.Timeout:
pass
msg_prefix = (
"Spawning is complete and report waittime is expired, but not all reports recieved from workers"
)
finally:
timeout.cancel()

self.environment.events.spawning_complete.fire(user_count=sum(self.target_user_classes_count.values()))
self.spawning_completed = True

logger.info("All users spawned: %s" % _format_user_classes_count_for_log(self.reported_user_classes_count))
logger.info("%s: %s" % (msg_prefix, _format_user_classes_count_for_log(self.reported_user_classes_count)))

@functools.lru_cache()
def _wait_for_workers_report_after_ramp_up(self) -> float:
"""
The amount of time to wait after a ramp-up in order for all the workers to report their state
to the master. If not supplied by the user, it is 100ms by default. If the supplied value is a number,
to the master. If not supplied by the user, it is 1000ms by default. If the supplied value is a number,
it is taken as-is. If the supplied value is a pattern like "some_number * WORKER_REPORT_INTERVAL",
the value will be "some_number * WORKER_REPORT_INTERVAL". The most sensible value would be something
like "1.25 * WORKER_REPORT_INTERVAL". However, some users might find it too high, so it is left
to a really small value of 100ms by default.
to a relatively small value of 1000ms by default.
"""
locust_wait_for_workers_report_after_ramp_up = os.getenv("LOCUST_WAIT_FOR_WORKERS_REPORT_AFTER_RAMP_UP")
if locust_wait_for_workers_report_after_ramp_up is None:
return 0.1
return 1.0

match = re.search(
r"^(?P<coeff>(\d+)|(\d+\.\d+))[ ]*\*[ ]*WORKER_REPORT_INTERVAL$",
Expand Down
107 changes: 106 additions & 1 deletion locust/test/test_main.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
from locust.user import HttpUser, User, TaskSet
from .mock_locustfile import mock_locustfile, MOCK_LOCUSTFILE_CONTENT
from .testcases import LocustTestCase
from .util import temporary_file, get_free_tcp_port
from .util import temporary_file, get_free_tcp_port, patch_env


class TestLoadLocustfile(LocustTestCase):
Expand Down Expand Up @@ -879,3 +879,108 @@ def task2(self):
self.assertNotIn("task2", stdout_worker)
self.assertEqual(0, proc.returncode)
self.assertEqual(0, proc_worker.returncode)

def test_distributed(self):
LOCUSTFILE_CONTENT = textwrap.dedent(
"""
from locust import User, task, constant
class User1(User):
wait_time = constant(1)
@task
def t(self):
pass
"""
)
with mock_locustfile(content=LOCUSTFILE_CONTENT) as mocked:
proc = subprocess.Popen(
[
"locust",
"-f",
mocked.file_path,
"--headless",
"--master",
"--expect-workers",
"1",
"-u",
"3",
"-t",
"5s",
],
stderr=STDOUT,
stdout=PIPE,
)
proc_worker = subprocess.Popen(
[
"locust",
"-f",
mocked.file_path,
"--worker",
],
stderr=STDOUT,
stdout=PIPE,
)
stdout = proc.communicate()[0].decode("utf-8")
proc_worker.communicate()

self.assertIn('All users spawned: {"User1": 3} (3 total users)', stdout)
self.assertIn("Shutting down (exit code 0)", stdout)

self.assertEqual(0, proc.returncode)
self.assertEqual(0, proc_worker.returncode)

def test_distributed_report_timeout_expired(self):
LOCUSTFILE_CONTENT = textwrap.dedent(
"""
from locust import User, task, constant
class User1(User):
wait_time = constant(1)
@task
def t(self):
pass
"""
)
with mock_locustfile(content=LOCUSTFILE_CONTENT) as mocked, patch_env(
"LOCUST_WAIT_FOR_WORKERS_REPORT_AFTER_RAMP_UP", "0.01"
) as _:
proc = subprocess.Popen(
[
"locust",
"-f",
mocked.file_path,
"--headless",
"--master",
"--expect-workers",
"1",
"-u",
"3",
"-t",
"5s",
],
stderr=STDOUT,
stdout=PIPE,
)
proc_worker = subprocess.Popen(
[
"locust",
"-f",
mocked.file_path,
"--worker",
],
stderr=STDOUT,
stdout=PIPE,
)
stdout = proc.communicate()[0].decode("utf-8")
proc_worker.communicate()

self.assertIn(
'Spawning is complete and report waittime is expired, but not all reports recieved from workers: {"User1": 2} (2 total users)',
stdout,
)
self.assertIn("Shutting down (exit code 0)", stdout)

self.assertEqual(0, proc.returncode)
self.assertEqual(0, proc_worker.returncode)
26 changes: 7 additions & 19 deletions locust/test/test_runners.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
task,
)
from retry import retry
from .util import patch_env

NETWORK_BROKEN = "network broken"

Expand Down Expand Up @@ -1173,7 +1174,7 @@ def tick(self):
return None

locust_worker_additional_wait_before_ready_after_stop = 5
with mock.patch("locust.runners.WORKER_REPORT_INTERVAL", new=0.3), _patch_env(
with mock.patch("locust.runners.WORKER_REPORT_INTERVAL", new=0.3), patch_env(
"LOCUST_WORKER_ADDITIONAL_WAIT_BEFORE_READY_AFTER_STOP",
str(locust_worker_additional_wait_before_ready_after_stop),
):
Expand Down Expand Up @@ -1430,7 +1431,7 @@ def tick(self):
user_class.weight = random.uniform(1, 20)

locust_worker_additional_wait_before_ready_after_stop = 5
with mock.patch("locust.runners.WORKER_REPORT_INTERVAL", new=0.3), _patch_env(
with mock.patch("locust.runners.WORKER_REPORT_INTERVAL", new=0.3), patch_env(
"LOCUST_WORKER_ADDITIONAL_WAIT_BEFORE_READY_AFTER_STOP",
str(locust_worker_additional_wait_before_ready_after_stop),
):
Expand Down Expand Up @@ -1548,7 +1549,7 @@ def tick(self):
return None

locust_worker_additional_wait_before_ready_after_stop = 2
with mock.patch("locust.runners.WORKER_REPORT_INTERVAL", new=0.3), _patch_env(
with mock.patch("locust.runners.WORKER_REPORT_INTERVAL", new=0.3), patch_env(
"LOCUST_WORKER_ADDITIONAL_WAIT_BEFORE_READY_AFTER_STOP",
str(locust_worker_additional_wait_before_ready_after_stop),
):
Expand Down Expand Up @@ -2608,16 +2609,16 @@ def assert_cache_hits():
master = self.get_runner()

master._wait_for_workers_report_after_ramp_up.cache_clear()
self.assertEqual(master._wait_for_workers_report_after_ramp_up(), 0.1)
self.assertEqual(master._wait_for_workers_report_after_ramp_up(), 1.0)
assert_cache_hits()

master._wait_for_workers_report_after_ramp_up.cache_clear()
with _patch_env("LOCUST_WAIT_FOR_WORKERS_REPORT_AFTER_RAMP_UP", "5.7"):
with patch_env("LOCUST_WAIT_FOR_WORKERS_REPORT_AFTER_RAMP_UP", "5.7"):
self.assertEqual(master._wait_for_workers_report_after_ramp_up(), 5.7)
assert_cache_hits()

master._wait_for_workers_report_after_ramp_up.cache_clear()
with mock.patch("locust.runners.WORKER_REPORT_INTERVAL", new=1.5), _patch_env(
with mock.patch("locust.runners.WORKER_REPORT_INTERVAL", new=1.5), patch_env(
"LOCUST_WAIT_FOR_WORKERS_REPORT_AFTER_RAMP_UP", "5.7 * WORKER_REPORT_INTERVAL"
):
self.assertEqual(master._wait_for_workers_report_after_ramp_up(), 5.7 * 1.5)
Expand All @@ -2626,19 +2627,6 @@ def assert_cache_hits():
master._wait_for_workers_report_after_ramp_up.cache_clear()


@contextmanager
def _patch_env(name: str, value: str):
prev_value = os.getenv(name)
os.environ[name] = value
try:
yield
finally:
if prev_value is None:
del os.environ[name]
else:
os.environ[name] = prev_value


class TestWorkerRunner(LocustTestCase):
def setUp(self):
super().setUp()
Expand Down
13 changes: 13 additions & 0 deletions locust/test/util.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,19 @@ def temporary_file(content, suffix="_locustfile.py"):
os.remove(f.name)


@contextmanager
def patch_env(name: str, value: str):
prev_value = os.getenv(name)
os.environ[name] = value
try:
yield
finally:
if prev_value is None:
del os.environ[name]
else:
os.environ[name] = prev_value


def get_free_tcp_port():
"""
Find an unused TCP port
Expand Down

0 comments on commit aa256cf

Please sign in to comment.