resources/tools/wrk/wrk.py

   1 # Copyright (c) 2019 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 check_wrk(tg_node):
  61     """Check if wrk is installed 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         command is not availble.
  67     """
  68
  69     if tg_node[u"type"] != NodeType.TG:
  70         raise RuntimeError(u"Node type is not a TG.")
  71
  72     ssh = SSH()
  73     ssh.connect(tg_node)
  74
  75     ret, _, _ = ssh.exec_command(
  76         f"sudo -E sh -c '{Constants.REMOTE_FW_DIR}/resources/tools/"
  77         f"wrk/wrk_utils.sh installed'"
  78     )
  79     if int(ret) != 0:
  80         raise RuntimeError(u"WRK is not installed on TG node.")
  81
  82
  83 def run_wrk(tg_node, profile_name, tg_numa, test_type, warm_up=False):
  84     """Send the traffic as defined in the profile.
  85
  86     :param tg_node: Traffic generator node.
  87     :param profile_name: The name of wrk traffic profile.
  88     :param tg_numa: Numa node on which wrk will run.
  89     :param test_type: The type of the tests: cps, rps, bw
  90     :param warm_up: If True, warm-up traffic is generated before test traffic.
  91     :type profile_name: str
  92     :type tg_node: dict
  93     :type tg_numa: int
  94     :type test_type: str
  95     :type warm_up: bool
  96     :returns: Message with measured data.
  97     :rtype: str
  98     :raises: RuntimeError if node type is not a TG.
  99     """
 100
 101     if tg_node[u"type"] != NodeType.TG:
 102         raise RuntimeError(u"Node type is not a TG.")
 103
 104     # Parse and validate the profile
 105     profile_path = f"resources/traffic_profiles/wrk/{profile_name}.yaml"
 106     profile = WrkTrafficProfile(profile_path).traffic_profile
 107
 108     cores = CpuUtils.cpu_list_per_node(tg_node, tg_numa)
 109     first_cpu = cores[profile[u"first-cpu"]]
 110
 111     if len(profile[u"urls"]) == 1 and profile[u"cpus"] == 1:
 112         params = [
 113             u"traffic_1_url_1_core",
 114             str(first_cpu),
 115             str(profile[u"nr-of-threads"]),
 116             str(profile[u"nr-of-connections"]),
 117             f"{profile[u'duration']}s",
 118             f"'{profile[u'header']}'",
 119             str(profile[u"timeout"]),
 120             str(profile[u"script"]),
 121             str(profile[u"latency"]),
 122             f"'{u' '.join(profile[u'urls'])}'"
 123         ]
 124         if warm_up:
 125             warm_up_params = deepcopy(params)
 126             warm_up_params[4] = u"10s"
 127     elif len(profile[u"urls"]) == profile[u"cpus"]:
 128         params = [
 129             u"traffic_n_urls_n_cores",
 130             str(first_cpu),
 131             str(profile[u"nr-of-threads"]),
 132             str(profile[u"nr-of-connections"]),
 133             f"{profile[u'duration']}s",
 134             f"'{profile[u'header']}'",
 135             str(profile[u"timeout"]),
 136             str(profile[u"script"]),
 137             str(profile[u"latency"]),
 138             f"'{u' '.join(profile[u'urls'])}'"
 139         ]
 140         if warm_up:
 141             warm_up_params = deepcopy(params)
 142             warm_up_params[4] = u"10s"
 143     else:
 144         params = [
 145             u"traffic_n_urls_m_cores",
 146             str(first_cpu),
 147             str(profile[u"cpus"] // len(profile[u"urls"])),
 148             str(profile[u"nr-of-threads"]),
 149             str(profile[u"nr-of-connections"]),
 150             f"{profile[u'duration']}s",
 151             f"'{profile[u'header']}'",
 152             str(profile[u"timeout"]),
 153             str(profile[u"script"]),
 154             str(profile[u"latency"]),
 155             f"'{u' '.join(profile[u'urls'])}'"
 156         ]
 157         if warm_up:
 158             warm_up_params = deepcopy(params)
 159             warm_up_params[5] = u"10s"
 160
 161     args = u" ".join(params)
 162
 163     ssh = SSH()
 164     ssh.connect(tg_node)
 165
 166     if warm_up:
 167         warm_up_args = u" ".join(warm_up_params)
 168         ret, _, _ = ssh.exec_command(
 169             f"{Constants.REMOTE_FW_DIR}/resources/tools/wrk/wrk_utils.sh "
 170             f"{warm_up_args}", timeout=1800
 171         )
 172         if int(ret) != 0:
 173             raise RuntimeError(u"wrk runtime error.")
 174         sleep(60)
 175
 176     ret, stdout, _ = ssh.exec_command(
 177         f"{Constants.REMOTE_FW_DIR}/resources/tools/wrk/wrk_utils.sh {args}",
 178         timeout=1800
 179     )
 180     if int(ret) != 0:
 181         raise RuntimeError('wrk runtime error.')
 182
 183     stats = _parse_wrk_output(stdout)
 184
 185     log_msg = u"\nMeasured values:\n"
 186     if test_type == u"cps":
 187         log_msg += u"Connections/sec: Avg / Stdev / Max  / +/- Stdev\n"
 188         for item in stats[u"rps-stats-lst"]:
 189             log_msg += f"{0} / {1} / {2} / {3}\n".format(*item)
 190         log_msg += f"Total cps: {stats[u'rps-sum']}cps\n"
 191     elif test_type == u"rps":
 192         log_msg += u"Requests/sec: Avg / Stdev / Max  / +/- Stdev\n"
 193         for item in stats[u"rps-stats-lst"]:
 194             log_msg += f"{0} / {1} / {2} / {3}\n".format(*item)
 195         log_msg += f"Total rps: {stats[u'rps-sum']}rps\n"
 196     elif test_type == u"bw":
 197         log_msg += f"Transfer/sec: {stats[u'bw-sum']}Bps"
 198
 199     logger.info(log_msg)
 200
 201     return log_msg
 202
 203
 204 def _parse_wrk_output(msg):
 205     """Parse the wrk stdout with the results.
 206
 207     :param msg: stdout of wrk.
 208     :type msg: str
 209     :returns: Parsed results.
 210     :rtype: dict
 211     :raises: WrkError if the message does not include the results.
 212     """
 213
 214     if u"Thread Stats" not in msg:
 215         raise WrkError(u"The output of wrk does not include the results.")
 216
 217     msg_lst = msg.splitlines(False)
 218
 219     stats = {
 220         u"latency-dist-lst": list(),
 221         u"latency-stats-lst": list(),
 222         u"rps-stats-lst": list(),
 223         u"rps-lst": list(),
 224         u"bw-lst": list(),
 225         u"rps-sum": 0,
 226         u"bw-sum": None
 227     }
 228
 229     for line in msg_lst:
 230         if u"Latency Distribution" in line:
 231             # Latency distribution - 50%, 75%, 90%, 99%
 232             pass
 233         elif u"Latency" in line:
 234             # Latency statistics - Avg, Stdev, Max, +/- Stdev
 235             pass
 236         elif u"Req/Sec" in line:
 237             # rps statistics - Avg, Stdev, Max, +/- Stdev
 238             stats[u"rps-stats-lst"].append(
 239                 (
 240                     _evaluate_number(re.search(REGEX_RPS_STATS, line).group(1)),
 241                     _evaluate_number(re.search(REGEX_RPS_STATS, line).group(2)),
 242                     _evaluate_number(re.search(REGEX_RPS_STATS, line).group(3)),
 243                     _evaluate_number(re.search(REGEX_RPS_STATS, line).group(4))
 244                 )
 245             )
 246         elif u"Requests/sec:" in line:
 247             # rps (cps)
 248             stats[u"rps-lst"].append(
 249                 _evaluate_number(re.search(REGEX_RPS, line).group(1))
 250             )
 251         elif u"Transfer/sec:" in line:
 252             # BW
 253             stats[u"bw-lst"].append(
 254                 _evaluate_number(re.search(REGEX_BW, line).group(1))
 255             )
 256
 257     for item in stats[u"rps-stats-lst"]:
 258         stats[u"rps-sum"] += item[0]
 259     stats[u"bw-sum"] = sum(stats[u"bw-lst"])
 260
 261     return stats
 262
 263
 264 def _evaluate_number(num):
 265     """Evaluate the numeric value of the number with multiplicands, e.g.:
 266     12.25k --> 12250
 267
 268     :param num: Number to evaluate.
 269     :type num: str
 270     :returns: Evaluated number.
 271     :rtype: float
 272     :raises: WrkError if it is not possible to evaluate the given number.
 273     """
 274
 275     val = re.search(REGEX_NUM, num)
 276     try:
 277         val_num = float(val.group(1))
 278     except ValueError:
 279         raise WrkError(
 280             u"The output of wrk does not include the results or the format "
 281             u"of results has changed."
 282         )
 283     val_mul = val.group(2).lower()
 284     if val_mul:
 285         if u"k" in val_mul:
 286             val_num *= 1000
 287         elif u"m" in val_mul:
 288             val_num *= 1000000
 289         elif u"g" in val_mul:
 290             val_num *= 1000000000
 291         elif u"b" in val_mul:
 292             pass
 293         elif u"%" in val_mul:
 294             pass
 295         elif u"" in val_mul:
 296             pass
 297         else:
 298             raise WrkError(f"The multiplicand {val_mul} is not defined.")
 299     return val_num