FIX: use len parameter name in prefix structure in ip_route_add_del
[csit.git] / resources / tools / wrk / wrk.py
1 # Copyright (c) 2018 Cisco and/or its affiliates.
2 # Licensed under the Apache License, Version 2.0 (the "License");
3 # you may not use this file except in compliance with the License.
4 # You may obtain a copy of the License at:
5 #
6 #     http://www.apache.org/licenses/LICENSE-2.0
7 #
8 # Unless required by applicable law or agreed to in writing, software
9 # distributed under the License is distributed on an "AS IS" BASIS,
10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11 # See the License for the specific language governing permissions and
12 # limitations under the License.
13
14 """wrk implementation into CSIT framework.
15 """
16
17 import re
18
19 from copy import deepcopy
20 from time import sleep
21
22 from robot.api import logger
23
24 from resources.libraries.python.ssh import SSH
25 from resources.libraries.python.topology import NodeType
26 from resources.libraries.python.CpuUtils import CpuUtils
27 from resources.libraries.python.Constants import Constants
28
29 from resources.tools.wrk.wrk_traffic_profile_parser import WrkTrafficProfile
30 from resources.tools.wrk.wrk_errors import WrkError
31
32
33 REGEX_LATENCY_STATS = \
34     r"Latency\s*" \
35     r"(\d*\.*\d*\S*)\s*" \
36     r"(\d*\.*\d*\S*)\s*" \
37     r"(\d*\.*\d*\S*)\s*" \
38     r"(\d*\.*\d*\%)"
39 REGEX_RPS_STATS = \
40     r"Req/Sec\s*" \
41     r"(\d*\.*\d*\S*)\s*" \
42     r"(\d*\.*\d*\S*)\s*" \
43     r"(\d*\.*\d*\S*)\s*" \
44     r"(\d*\.*\d*\%)"
45 REGEX_RPS = r"Requests/sec:\s*" \
46             r"(\d*\.*\S*)"
47 REGEX_BW = r"Transfer/sec:\s*" \
48            r"(\d*\.*\S*)"
49 REGEX_LATENCY_DIST = \
50     r"Latency Distribution\n" \
51     r"\s*50\%\s*(\d*\.*\d*\D*)\n" \
52     r"\s*75\%\s*(\d*\.*\d*\D*)\n" \
53     r"\s*90\%\s*(\d*\.*\d*\D*)\n" \
54     r"\s*99\%\s*(\d*\.*\d*\D*)\n"
55
56 # Split number and multiplicand, e.g. 14.25k --> 14.25 and k
57 REGEX_NUM = r"(\d*\.*\d*)(\D*)"
58
59
60 def install_wrk(tg_node):
61     """Install wrk on the TG node.
62
63     :param tg_node: Traffic generator node.
64     :type tg_node: dict
65     :raises: RuntimeError if the given node is not a TG node or if the
66     installation fails.
67     """
68
69     if tg_node['type'] != NodeType.TG:
70         raise RuntimeError('Node type is not a TG.')
71
72     ssh = SSH()
73     ssh.connect(tg_node)
74
75     ret, _, _ = ssh.exec_command(
76         "sudo -E "
77         "sh -c '{0}/resources/tools/wrk/wrk_utils.sh install false'".
78         format(Constants.REMOTE_FW_DIR), timeout=1800)
79     if int(ret) != 0:
80         raise RuntimeError('Installation of wrk on TG node failed.')
81
82
83 def destroy_wrk(tg_node):
84     """Destroy wrk on the TG node.
85
86     :param tg_node: Traffic generator node.
87     :type tg_node: dict
88     :raises: RuntimeError if the given node is not a TG node or the removal of
89     wrk failed.
90     """
91
92     if tg_node['type'] != NodeType.TG:
93         raise RuntimeError('Node type is not a TG.')
94
95     ssh = SSH()
96     ssh.connect(tg_node)
97
98     ret, _, _ = ssh.exec_command(
99         "sudo -E "
100         "sh -c '{0}/resources/tools/wrk/wrk_utils.sh destroy'".
101         format(Constants.REMOTE_FW_DIR), timeout=1800)
102     if int(ret) != 0:
103         raise RuntimeError('Removal of wrk from the TG node failed.')
104
105
106 def run_wrk(tg_node, profile_name, tg_numa, test_type, warm_up=False):
107     """Send the traffic as defined in the profile.
108
109     :param tg_node: Traffic generator node.
110     :param profile_name: The name of wrk traffic profile.
111     :param tg_numa: Numa node on which wrk will run.
112     :param test_type: The type of the tests: cps, rps, bw
113     :param warm_up: If True, warm-up traffic is generated before test traffic.
114     :type profile_name: str
115     :type tg_node: dict
116     :type tg_numa: int
117     :type test_type: str
118     :type warm_up: bool
119     :returns: Message with measured data.
120     :rtype: str
121     :raises: RuntimeError if node type is not a TG.
122     """
123
124     if tg_node['type'] != NodeType.TG:
125         raise RuntimeError('Node type is not a TG.')
126
127     # Parse and validate the profile
128     profile_path = ("resources/traffic_profiles/wrk/{0}.yaml".
129                     format(profile_name))
130     profile = WrkTrafficProfile(profile_path).traffic_profile
131
132     cores = CpuUtils.cpu_list_per_node(tg_node, tg_numa)
133     first_cpu = cores[profile["first-cpu"]]
134
135     if len(profile["urls"]) == 1 and profile["cpus"] == 1:
136         params = [
137             "traffic_1_url_1_core",
138             str(first_cpu),
139             str(profile["nr-of-threads"]),
140             str(profile["nr-of-connections"]),
141             "{0}s".format(profile["duration"]),
142             "'{0}'".format(profile["header"]),
143             str(profile["timeout"]),
144             str(profile["script"]),
145             str(profile["latency"]),
146             "'{0}'".format(" ".join(profile["urls"]))
147         ]
148         if warm_up:
149             warm_up_params = deepcopy(params)
150             warm_up_params[4] = "10s"
151     elif len(profile["urls"]) == profile["cpus"]:
152         params = [
153             "traffic_n_urls_n_cores",
154             str(first_cpu),
155             str(profile["nr-of-threads"]),
156             str(profile["nr-of-connections"]),
157             "{0}s".format(profile["duration"]),
158             "'{0}'".format(profile["header"]),
159             str(profile["timeout"]),
160             str(profile["script"]),
161             str(profile["latency"]),
162             "'{0}'".format(" ".join(profile["urls"]))
163         ]
164         if warm_up:
165             warm_up_params = deepcopy(params)
166             warm_up_params[4] = "10s"
167     else:
168         params = [
169             "traffic_n_urls_m_cores",
170             str(first_cpu),
171             str(profile["cpus"] / len(profile["urls"])),
172             str(profile["nr-of-threads"]),
173             str(profile["nr-of-connections"]),
174             "{0}s".format(profile["duration"]),
175             "'{0}'".format(profile["header"]),
176             str(profile["timeout"]),
177             str(profile["script"]),
178             str(profile["latency"]),
179             "'{0}'".format(" ".join(profile["urls"]))
180         ]
181         if warm_up:
182             warm_up_params = deepcopy(params)
183             warm_up_params[5] = "10s"
184
185     args = " ".join(params)
186
187     ssh = SSH()
188     ssh.connect(tg_node)
189
190     if warm_up:
191         warm_up_args = " ".join(warm_up_params)
192         ret, _, _ = ssh.exec_command(
193             "{0}/resources/tools/wrk/wrk_utils.sh {1}".
194             format(Constants.REMOTE_FW_DIR, warm_up_args), timeout=1800)
195         if int(ret) != 0:
196             raise RuntimeError('wrk runtime error.')
197         sleep(60)
198
199     ret, stdout, _ = ssh.exec_command(
200         "{0}/resources/tools/wrk/wrk_utils.sh {1}".
201         format(Constants.REMOTE_FW_DIR, args), timeout=1800)
202     if int(ret) != 0:
203         raise RuntimeError('wrk runtime error.')
204
205     stats = _parse_wrk_output(stdout)
206
207     log_msg = "\nMeasured values:\n"
208     if test_type == "cps":
209         log_msg += "Connections/sec: Avg / Stdev / Max  / +/- Stdev\n"
210         for item in stats["rps-stats-lst"]:
211             log_msg += "{0} / {1} / {2} / {3}\n".format(*item)
212         log_msg += "Total cps: {0}cps\n".format(stats["rps-sum"])
213     elif test_type == "rps":
214         log_msg += "Requests/sec: Avg / Stdev / Max  / +/- Stdev\n"
215         for item in stats["rps-stats-lst"]:
216             log_msg += "{0} / {1} / {2} / {3}\n".format(*item)
217         log_msg += "Total rps: {0}rps\n".format(stats["rps-sum"])
218     elif test_type == "bw":
219         log_msg += "Transfer/sec: {0}Bps".format(stats["bw-sum"])
220
221     logger.info(log_msg)
222
223     return log_msg
224
225
226 def _parse_wrk_output(msg):
227     """Parse the wrk stdout with the results.
228
229     :param msg: stdout of wrk.
230     :type msg: str
231     :returns: Parsed results.
232     :rtype: dict
233     :raises: WrkError if the message does not include the results.
234     """
235
236     if "Thread Stats" not in msg:
237         raise WrkError("The output of wrk does not include the results.")
238
239     msg_lst = msg.splitlines(False)
240
241     stats = {
242         "latency-dist-lst": list(),
243         "latency-stats-lst": list(),
244         "rps-stats-lst": list(),
245         "rps-lst": list(),
246         "bw-lst": list(),
247         "rps-sum": 0,
248         "bw-sum": None
249     }
250
251     for line in msg_lst:
252         if "Latency Distribution" in line:
253             # Latency distribution - 50%, 75%, 90%, 99%
254             pass
255         elif "Latency" in line:
256             # Latency statistics - Avg, Stdev, Max, +/- Stdev
257             pass
258         elif "Req/Sec" in line:
259             # rps statistics - Avg, Stdev, Max, +/- Stdev
260             stats["rps-stats-lst"].append((
261                 _evaluate_number(re.search(REGEX_RPS_STATS, line).group(1)),
262                 _evaluate_number(re.search(REGEX_RPS_STATS, line).group(2)),
263                 _evaluate_number(re.search(REGEX_RPS_STATS, line).group(3)),
264                 _evaluate_number(re.search(REGEX_RPS_STATS, line).group(4))))
265         elif "Requests/sec:" in line:
266             # rps (cps)
267             stats["rps-lst"].append(
268                 _evaluate_number(re.search(REGEX_RPS, line).group(1)))
269         elif "Transfer/sec:" in line:
270             # BW
271             stats["bw-lst"].append(
272                 _evaluate_number(re.search(REGEX_BW, line).group(1)))
273
274     for item in stats["rps-stats-lst"]:
275         stats["rps-sum"] += item[0]
276     stats["bw-sum"] = sum(stats["bw-lst"])
277
278     return stats
279
280
281 def _evaluate_number(num):
282     """Evaluate the numeric value of the number with multiplicands, e.g.:
283     12.25k --> 12250
284
285     :param num: Number to evaluate.
286     :type num: str
287     :returns: Evaluated number.
288     :rtype: float
289     :raises: WrkError if it is not possible to evaluate the given number.
290     """
291
292     val = re.search(REGEX_NUM, num)
293     try:
294         val_num = float(val.group(1))
295     except ValueError:
296         raise WrkError("The output of wrk does not include the results "
297                        "or the format of results has changed.")
298     val_mul = val.group(2).lower()
299     if val_mul:
300         if "k" in val_mul:
301             val_num *= 1000
302         elif "m" in val_mul:
303             val_num *= 1000000
304         elif "g" in val_mul:
305             val_num *= 1000000000
306         elif "b" in val_mul:
307             pass
308         elif "%" in val_mul:
309             pass
310         elif "" in val_mul:
311             pass
312         else:
313             raise WrkError("The multiplicand {0} is not defined.".
314                            format(val_mul))
315     return val_num