/
locust-parameter-variation.py
176 lines (123 loc) · 5.55 KB
/
locust-parameter-variation.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
#!/usr/bin/env python
import argparse
import csv
import glob
import platform
import os
import logging
import time
from common.Common import call_locust_with, call_locust_and_distribute_work
input_args = argparse.Namespace()
plt = platform.system()
if plt != "Windows":
import readline
def complete_python(text, state):
# replace ~ with the user's home dir. See https://docs.python.org/2/library/os.path.html
if '~' in text:
text = os.path.expanduser('~')
# autocomplete directories with having a trailing slash
if os.path.isdir(text):
text += '/'
return [x for x in glob.glob(text + '*.py')][state]
#if plt != "Windows":
# readline.set_completer_delims(' \t\n;')
# readline.parse_and_bind("tab: complete")
# readline.set_completer(complete_python)
#locust_script = input('Path to the Locust script: ')
#if plt != "Windows":
# readline.set_completer(None)
# url = input('URL of the software to test: ')
url = "http://localhost:1337"
fh = logging.FileHandler('locust-parameter-variation.log')
fh.setLevel(logging.DEBUG)
logging.basicConfig(format="%(asctime)s %(message)s",
level=os.environ.get("LOGLEVEL", "INFO"),
handlers=[fh])
avg_time_allowed_in_s = 10
max_time_allowed_in_s = 30
average_response_time = {}
min_response_time = {}
max_response_time = {}
def read_measurements_from_locust_csv_and_append_to_dictonaries(path, num_clients):
logger = logging.getLogger('readMeasurementsFromCsvAndAppendToDictonaries')
with open(path, newline='') as csvfile:
reader = csv.DictReader(csvfile)
avg = 0
min = 0
max = 0
for row in reader:
v = float(row['Average Response Time'])
avg = v if avg < v else avg
v = float(row['Min Response Time'])
min = v if min < v else min
v = float(row['Max Response Time'])
max = v if max < v else max
logger.info("Avg: {}, Min: {}, Max: {}".format(avg, min, max))
average_response_time[num_clients] = float(avg)
min_response_time[num_clients] = float(min)
max_response_time[num_clients] = float(max)
def config_complies_with_real_time_requirements(num_clients):
logger = logging.getLogger('config_complies_with_real_time_requirements')
# return True, because we have no measurements yet
if len(average_response_time) == 0 or len(max_response_time) == 0:
return True
if average_response_time[num_clients] == 0 or max_response_time[num_clients] == 0:
logger.error("Something went wrong: average or max response time was 0")
return False
average_response_time_s = average_response_time[num_clients] / 1000
max_response_time_s = max_response_time[num_clients] / 1000
logger.info(f"Clients: {num_clients}: avg: {average_response_time_s}s, max: {max_response_time_s}s")
exceeds_average_response_time = average_response_time_s > avg_time_allowed_in_s
exceeds_max_response_time = max_response_time_s > max_time_allowed_in_s
is_compliant = not (exceeds_average_response_time or exceeds_max_response_time)
logger.info(f"--> {is_compliant}")
return is_compliant
def parameter_variation_loop(multiplier: int = 5000):
logger = logging.getLogger('parameter_variation_loop')
num_clients = 1
x = 1
y = 0
z = 0
logger.info(f"Starting performance test.")
is_first_run = True
while config_complies_with_real_time_requirements(num_clients):
if not is_first_run:
logger.info("Sleeping for 1 min ...")
time.sleep(60)
is_first_run = False
# start with multiplier clients, then increase linearly (2*multiplier, ... x*multiplier)
# until the number of clients exceeds the threshold.
# After that, keep increasing with one tenth of the multiplier (x*multiplier + y*multiplier/10)
num_clients = max(x * multiplier + y * int(multiplier / 10) + z * 10, 1)
if num_clients >= 1500:
z += 1
elif num_clients >= 1200:
y += 1
else:
x += 1
call_locust_and_distribute_work(locust_script, url, num_clients, runtime_in_min=10, use_load_test_shape=False)
# call_locust_with(locust_script, url, num_clients, runtime_in_min=10, omit_csv_files=True)
read_measurements_from_locust_csv_and_append_to_dictonaries(f"loadtest_{num_clients}_clients_stats.csv", num_clients)
logger.info(f"Finished performance test. System failed at {num_clients}")
def read_cli_args():
parser = argparse.ArgumentParser(description='Locust Wrapper.')
parser.add_argument('locust_script', help='Path to the locust script to execute')
parser.add_argument('-p', '--parametervariation', action='store_true', help='run the test and variate parameters')
parser.add_argument('-m', '--multiplier',
type=int,
help='start and linearly increase number of clients by the given multiplier',
default=200)
parser.add_argument('-u', '--url', help='URL of the System under Test')
global input_args
input_args = parser.parse_args()
print("Args: " + str(input_args))
if __name__ == "__main__":
read_cli_args()
locust_script = input_args.locust_script
if input_args.url:
url = input_args.url
if input_args.parametervariation:
parameter_variation_loop(input_args.multiplier)
else:
call_locust_with(locust_script, url, clients=1)
read_measurements_from_locust_csv_and_append_to_dictonaries("loadtest_1_clients_stats.csv", 1)