X-Git-Url: https://gerrit.fd.io/r/gitweb?p=csit.git;a=blobdiff_plain;f=resources%2Ftools%2Fpresentation%2Finput_data_parser.py;h=8747f938a90fbabd3b8a9e1694ce24ab6d87dd46;hp=d108d09e84774a5f17a9c6c1d6882651323aa741;hb=7bd576415bbc25c00d9f5dd1b4cd19482432718f;hpb=c763cfcb064e4f4acf6b8309b08d3800b9bd5331 diff --git a/resources/tools/presentation/input_data_parser.py b/resources/tools/presentation/input_data_parser.py index d108d09e84..8747f938a9 100644 --- a/resources/tools/presentation/input_data_parser.py +++ b/resources/tools/presentation/input_data_parser.py @@ -239,6 +239,15 @@ class ExecutionChecker(ResultVisitor): ) REGEX_MRR_MSG_INFO = re.compile(r'.*\[(.*)\]') + REGEX_VSAP_MSG_INFO = re.compile( + r'Transfer Rate: (\d*.\d*).*\n' + r'Latency: (\d*.\d*).*\n' + r'Completed requests: (\d*).*\n' + r'Failed requests: (\d*).*\n' + r'Total data transferred: (\d*).*\n' + r'Connection [cr]ps rate:\s*(\d*.\d*)' + ) + # Needed for CPS and PPS tests REGEX_NDRPDR_LAT_BASE = re.compile( r'LATENCY.*\[\'(.*)\', \'(.*)\'\]\s\n.*\n.*\n' @@ -278,15 +287,17 @@ class ExecutionChecker(ResultVisitor): ) REGEX_TC_TAG = re.compile(r'\d+[tT]\d+[cC]') - REGEX_TC_NAME_OLD = re.compile(r'-\d+[tT]\d+[cC]-') - REGEX_TC_NAME_NEW = re.compile(r'-\d+[cC]-') REGEX_TC_NUMBER = re.compile(r'tc\d{2}-') REGEX_TC_PAPI_CLI = re.compile(r'.*\((\d+.\d+.\d+.\d+.) - (.*)\)') - def __init__(self, metadata, mapping, ignore): + REGEX_SH_RUN_HOST = re.compile( + r'hostname=\"(\d{1,3}.\d{1,3}.\d{1,3}.\d{1,3})\",hook=\"(.*)\"' + ) + + def __init__(self, metadata, mapping, ignore, for_output): """Initialisation. :param metadata: Key-value pairs to be included in "metadata" part of @@ -294,9 +305,11 @@ class ExecutionChecker(ResultVisitor): :param mapping: Mapping of the old names of test cases to the new (actual) one. :param ignore: List of TCs to be ignored. + :param for_output: Output to be generated from downloaded data. :type metadata: dict :type mapping: dict :type ignore: list + :type for_output: str """ # Type of message to parse out from the test messages @@ -317,6 +330,8 @@ class ExecutionChecker(ResultVisitor): # Ignore list self._ignore = ignore + self._for_output = for_output + # Number of PAPI History messages found: # 0 - no message # 1 - PAPI History of DUT1 @@ -324,6 +339,8 @@ class ExecutionChecker(ResultVisitor): self._conf_history_lookup_nr = 0 self._sh_run_counter = 0 + self._telemetry_kw_counter = 0 + self._telemetry_msg_counter = 0 # Test ID of currently processed test- the lowercase full path to the # test @@ -343,12 +360,12 @@ class ExecutionChecker(ResultVisitor): # Dictionary defining the methods used to parse different types of # messages self.parse_msg = { - u"timestamp": self._get_timestamp, u"vpp-version": self._get_vpp_version, u"dpdk-version": self._get_dpdk_version, u"teardown-papi-history": self._get_papi_history, u"test-show-runtime": self._get_show_run, - u"testbed": self._get_testbed + u"testbed": self._get_testbed, + u"test-telemetry": self._get_telemetry } @property @@ -594,18 +611,6 @@ class ExecutionChecker(ResultVisitor): finally: self._msg_type = None - def _get_timestamp(self, msg): - """Called when extraction of timestamp is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - self._timestamp = msg.timestamp[:14] - self._data[u"metadata"][u"generated"] = self._timestamp - self._msg_type = None - def _get_papi_history(self, msg): """Called when extraction of PAPI command history is required. @@ -660,53 +665,90 @@ class ExecutionChecker(ResultVisitor): except (AttributeError, IndexError): sock = u"" - runtime = loads(str(msg.message).replace(u' ', u'').replace(u'\n', u''). - replace(u"'", u'"').replace(u'b"', u'"'). - replace(u'u"', u'"').split(u":", 1)[1]) - - try: - threads_nr = len(runtime[0][u"clocks"]) - except (IndexError, KeyError): - return - dut = u"dut{nr}".format( nr=len(self._data[u'tests'][self._test_id][u'show-run'].keys()) + 1) - oper = { - u"host": host, - u"socket": sock, - u"runtime": runtime, - u"threads": OrderedDict({idx: list() for idx in range(threads_nr)}) - } + self._data[u'tests'][self._test_id][u'show-run'][dut] = \ + copy.copy( + { + u"host": host, + u"socket": sock, + u"runtime": str(msg.message).replace(u' ', u''). + replace(u'\n', u'').replace(u"'", u'"'). + replace(u'b"', u'"').replace(u'u"', u'"'). + split(u":", 1)[1] + } + ) - for item in runtime: - for idx in range(threads_nr): - if item[u"vectors"][idx] > 0: - clocks = item[u"clocks"][idx] / item[u"vectors"][idx] - elif item[u"calls"][idx] > 0: - clocks = item[u"clocks"][idx] / item[u"calls"][idx] - elif item[u"suspends"][idx] > 0: - clocks = item[u"clocks"][idx] / item[u"suspends"][idx] - else: - clocks = 0.0 + def _get_telemetry(self, msg): + """Called when extraction of VPP telemetry data is required. - if item[u"calls"][idx] > 0: - vectors_call = item[u"vectors"][idx] / item[u"calls"][idx] - else: - vectors_call = 0.0 + :param msg: Message to process. + :type msg: Message + :returns: Nothing. + """ - if int(item[u"calls"][idx]) + int(item[u"vectors"][idx]) + \ - int(item[u"suspends"][idx]): - oper[u"threads"][idx].append([ - item[u"name"], - item[u"calls"][idx], - item[u"vectors"][idx], - item[u"suspends"][idx], - clocks, - vectors_call - ]) + if self._telemetry_kw_counter > 1: + return + if not msg.message.count(u"# TYPE vpp_runtime_calls"): + return - self._data[u'tests'][self._test_id][u'show-run'][dut] = copy.copy(oper) + if u"telemetry-show-run" not in \ + self._data[u"tests"][self._test_id].keys(): + self._data[u"tests"][self._test_id][u"telemetry-show-run"] = dict() + + self._telemetry_msg_counter += 1 + groups = re.search(self.REGEX_SH_RUN_HOST, msg.message) + if not groups: + return + try: + host = groups.group(1) + except (AttributeError, IndexError): + host = u"" + try: + sock = groups.group(2) + except (AttributeError, IndexError): + sock = u"" + runtime = { + u"source_type": u"node", + u"source_id": host, + u"msg_type": u"metric", + u"log_level": u"INFO", + u"timestamp": msg.timestamp, + u"msg": u"show_runtime", + u"host": host, + u"socket": sock, + u"data": list() + } + for line in msg.message.splitlines(): + if not line.startswith(u"vpp_runtime_"): + continue + try: + params, value, timestamp = line.rsplit(u" ", maxsplit=2) + cut = params.index(u"{") + name = params[:cut].split(u"_", maxsplit=2)[-1] + labels = eval( + u"dict" + params[cut:].replace('{', '(').replace('}', ')') + ) + labels[u"graph_node"] = labels.pop(u"name") + runtime[u"data"].append( + { + u"name": name, + u"value": value, + u"timestamp": timestamp, + u"labels": labels + } + ) + except (TypeError, ValueError, IndexError): + continue + self._data[u'tests'][self._test_id][u'telemetry-show-run']\ + [f"dut{self._telemetry_msg_counter}"] = copy.copy( + { + u"host": host, + u"socket": sock, + u"runtime": runtime + } + ) def _get_ndrpdr_throughput(self, msg): """Get NDR_LOWER, NDR_UPPER, PDR_LOWER and PDR_UPPER from the test @@ -926,6 +968,39 @@ class ExecutionChecker(ResultVisitor): return result, status + def _get_vsap_data(self, msg, tags): + """Get data from the vsap test message. + + :param msg: The test message to be parsed. + :param tags: Test tags. + :type msg: str + :type tags: list + :returns: Parsed data as a JSON dict and the status (PASS/FAIL). + :rtype: tuple(dict, str) + """ + result = dict() + status = u"FAIL" + + groups = re.search(self.REGEX_VSAP_MSG_INFO, msg) + if groups is not None: + try: + result[u"transfer-rate"] = float(groups.group(1)) * 1e3 + result[u"latency"] = float(groups.group(2)) + result[u"completed-requests"] = int(groups.group(3)) + result[u"failed-requests"] = int(groups.group(4)) + result[u"bytes-transferred"] = int(groups.group(5)) + if u"TCP_CPS"in tags: + result[u"cps"] = float(groups.group(6)) + elif u"TCP_RPS" in tags: + result[u"rps"] = float(groups.group(6)) + else: + return result, status + status = u"PASS" + except (IndexError, ValueError): + pass + + return result, status + def visit_suite(self, suite): """Implements traversing through the suite and its direct children. @@ -990,6 +1065,8 @@ class ExecutionChecker(ResultVisitor): """ self._sh_run_counter = 0 + self._telemetry_kw_counter = 0 + self._telemetry_msg_counter = 0 longname_orig = test.longname.lower() @@ -1044,38 +1121,36 @@ class ExecutionChecker(ResultVisitor): else: test_result[u"msg"] = test.message - if u"PERFTEST" in tags: + if u"PERFTEST" in tags and u"TREX" not in tags: # Replace info about cores (e.g. -1c-) with the info about threads # and cores (e.g. -1t1c-) in the long test case names and in the # test case names if necessary. - groups = re.search(self.REGEX_TC_NAME_OLD, self._test_id) - if not groups: - tag_count = 0 - tag_tc = str() - for tag in test_result[u"tags"]: - groups = re.search(self.REGEX_TC_TAG, tag) - if groups: - tag_count += 1 - tag_tc = tag - - if tag_count == 1: - self._test_id = re.sub( - self.REGEX_TC_NAME_NEW, f"-{tag_tc.lower()}-", - self._test_id, count=1 - ) - test_result[u"name"] = re.sub( - self.REGEX_TC_NAME_NEW, f"-{tag_tc.lower()}-", - test_result["name"], count=1 - ) - else: - test_result[u"status"] = u"FAIL" - self._data[u"tests"][self._test_id] = test_result - logging.debug( - f"The test {self._test_id} has no or more than one " - f"multi-threading tags.\n" - f"Tags: {test_result[u'tags']}" - ) - return + tag_count = 0 + tag_tc = str() + for tag in test_result[u"tags"]: + groups = re.search(self.REGEX_TC_TAG, tag) + if groups: + tag_count += 1 + tag_tc = tag + + if tag_count == 1: + self._test_id = re.sub( + self.REGEX_TC_NAME_NEW, f"-{tag_tc.lower()}-", + self._test_id, count=1 + ) + test_result[u"name"] = re.sub( + self.REGEX_TC_NAME_NEW, f"-{tag_tc.lower()}-", + test_result["name"], count=1 + ) + else: + test_result[u"status"] = u"FAIL" + self._data[u"tests"][self._test_id] = test_result + logging.debug( + f"The test {self._test_id} has no or more than one " + f"multi-threading tags.\n" + f"Tags: {test_result[u'tags']}" + ) + return if u"DEVICETEST" in tags: test_result[u"type"] = u"DEVICETEST" @@ -1124,6 +1199,10 @@ class ExecutionChecker(ResultVisitor): if test.status == u"PASS": test_result[u"result"], test_result[u"status"] = \ self._get_hoststack_data(test.message, tags) + elif u"LDP_NGINX" in tags: + test_result[u"type"] = u"LDP_NGINX" + test_result[u"result"], test_result[u"status"] = \ + self._get_vsap_data(test.message, tags) # elif u"TCP" in tags: # This might be not used # test_result[u"type"] = u"TCP" # if test.status == u"PASS": @@ -1211,9 +1290,13 @@ class ExecutionChecker(ResultVisitor): :type test_kw: Keyword :returns: Nothing. """ - if test_kw.name.count(u"Show Runtime On All Duts") or \ - test_kw.name.count(u"Show Runtime Counters On All Duts") or \ - test_kw.name.count(u"Vpp Show Runtime On All Duts"): + if self._for_output == u"trending": + return + + if test_kw.name.count(u"Run Telemetry On All Duts"): + self._msg_type = u"test-telemetry" + self._telemetry_kw_counter += 1 + elif test_kw.name.count(u"Show Runtime On All Duts"): self._msg_type = u"test-show-runtime" self._sh_run_counter += 1 else: @@ -1255,9 +1338,6 @@ class ExecutionChecker(ResultVisitor): elif setup_kw.name.count(u"Install Dpdk Framework On All Duts") and \ not self._version: self._msg_type = u"dpdk-version" - elif setup_kw.name.count(u"Set Global Variable") \ - and not self._timestamp: - self._msg_type = u"timestamp" elif setup_kw.name.count(u"Setup Framework") and not self._testbed: self._msg_type = u"testbed" else: @@ -1352,16 +1432,20 @@ class InputData: (as described in ExecutionChecker documentation) """ - def __init__(self, spec): + def __init__(self, spec, for_output): """Initialization. :param spec: Specification. + :param for_output: Output to be generated from downloaded data. :type spec: Specification + :type for_output: str """ # Specification: self._cfg = spec + self._for_output = for_output + # Data store: self._input_data = pd.Series() @@ -1436,10 +1520,19 @@ class InputData: ) return None checker = ExecutionChecker( - metadata, self._cfg.mapping, self._cfg.ignore + metadata, self._cfg.mapping, self._cfg.ignore, self._for_output ) result.visit(checker) + checker.data[u"metadata"][u"tests_total"] = \ + result.statistics.total.all.total + checker.data[u"metadata"][u"tests_passed"] = \ + result.statistics.total.all.passed + checker.data[u"metadata"][u"tests_failed"] = \ + result.statistics.total.all.failed + checker.data[u"metadata"][u"elapsedtime"] = result.suite.elapsedtime + checker.data[u"metadata"][u"generated"] = result.suite.endtime[:14] + return checker.data def _download_and_parse_build(self, job, build, repeat, pid=10000): @@ -1975,15 +2068,6 @@ class InputData: """Print all operational data to console. """ - tbl_hdr = ( - u"Name", - u"Nr of Vectors", - u"Nr of Packets", - u"Suspends", - u"Cycles per Packet", - u"Average Vector Size" - ) - for job in self._input_data.values: for build in job.values: for test_id, test_data in build[u"tests"].items(): @@ -1991,12 +2075,60 @@ class InputData: if test_data.get(u"show-run", None) is None: continue for dut_name, data in test_data[u"show-run"].items(): - if data.get(u"threads", None) is None: + if data.get(u"runtime", None) is None: continue + runtime = loads(data[u"runtime"]) + try: + threads_nr = len(runtime[0][u"clocks"]) + except (IndexError, KeyError): + continue + threads = OrderedDict( + {idx: list() for idx in range(threads_nr)}) + for item in runtime: + for idx in range(threads_nr): + if item[u"vectors"][idx] > 0: + clocks = item[u"clocks"][idx] / \ + item[u"vectors"][idx] + elif item[u"calls"][idx] > 0: + clocks = item[u"clocks"][idx] / \ + item[u"calls"][idx] + elif item[u"suspends"][idx] > 0: + clocks = item[u"clocks"][idx] / \ + item[u"suspends"][idx] + else: + clocks = 0.0 + + if item[u"calls"][idx] > 0: + vectors_call = item[u"vectors"][idx] / \ + item[u"calls"][idx] + else: + vectors_call = 0.0 + + if int(item[u"calls"][idx]) + int( + item[u"vectors"][idx]) + \ + int(item[u"suspends"][idx]): + threads[idx].append([ + item[u"name"], + item[u"calls"][idx], + item[u"vectors"][idx], + item[u"suspends"][idx], + clocks, + vectors_call + ]) + print(f"Host IP: {data.get(u'host', '')}, " f"Socket: {data.get(u'socket', '')}") - for thread_nr, thread in data[u"threads"].items(): - txt_table = prettytable.PrettyTable(tbl_hdr) + for thread_nr, thread in threads.items(): + txt_table = prettytable.PrettyTable( + ( + u"Name", + u"Nr of Vectors", + u"Nr of Packets", + u"Suspends", + u"Cycles per Packet", + u"Average Vector Size" + ) + ) avg = 0.0 for row in thread: txt_table.add_row(row)