diff options
Diffstat (limited to 'resources/tools/presentation/input_data_parser.py')
| -rw-r--r-- | resources/tools/presentation/input_data_parser.py | 2162 |
1 files changed, 0 insertions, 2162 deletions
diff --git a/resources/tools/presentation/input_data_parser.py b/resources/tools/presentation/input_data_parser.py deleted file mode 100644 index 7499935ee8..0000000000 --- a/resources/tools/presentation/input_data_parser.py +++ /dev/null @@ -1,2162 +0,0 @@ -# Copyright (c) 2022 Cisco and/or its affiliates. -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at: -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -"""Data pre-processing - -- extract data from output.xml files generated by Jenkins jobs and store in - pandas' Series, -- provide access to the data. -- filter the data using tags, -""" - -import re -import copy -import resource -import logging - -from collections import OrderedDict -from os import remove, walk, listdir -from os.path import isfile, isdir, join -from datetime import datetime as dt -from datetime import timedelta -from json import loads -from json.decoder import JSONDecodeError - -import hdrh.histogram -import hdrh.codec -import prettytable -import pandas as pd - -from robot.api import ExecutionResult, ResultVisitor -from robot import errors - -from resources.libraries.python import jumpavg -from input_data_files import download_and_unzip_data_file -from pal_errors import PresentationError - - -# Separator used in file names -SEPARATOR = u"__" - - -class ExecutionChecker(ResultVisitor): - """Class to traverse through the test suite structure. - - The functionality implemented in this class generates a json structure: - - Performance tests: - - { - "metadata": { - "generated": "Timestamp", - "version": "SUT version", - "job": "Jenkins job name", - "build": "Information about the build" - }, - "suites": { - "Suite long name 1": { - "name": Suite name, - "doc": "Suite 1 documentation", - "parent": "Suite 1 parent", - "level": "Level of the suite in the suite hierarchy" - } - "Suite long name N": { - "name": Suite name, - "doc": "Suite N documentation", - "parent": "Suite 2 parent", - "level": "Level of the suite in the suite hierarchy" - } - } - "tests": { - # NDRPDR tests: - "ID": { - "name": "Test name", - "parent": "Name of the parent of the test", - "doc": "Test documentation", - "msg": "Test message", - "conf-history": "DUT1 and DUT2 VAT History", - "show-run": "Show Run", - "tags": ["tag 1", "tag 2", "tag n"], - "type": "NDRPDR", - "status": "PASS" | "FAIL", - "throughput": { - "NDR": { - "LOWER": float, - "UPPER": float - }, - "PDR": { - "LOWER": float, - "UPPER": float - } - }, - "latency": { - "NDR": { - "direction1": { - "min": float, - "avg": float, - "max": float, - "hdrh": str - }, - "direction2": { - "min": float, - "avg": float, - "max": float, - "hdrh": str - } - }, - "PDR": { - "direction1": { - "min": float, - "avg": float, - "max": float, - "hdrh": str - }, - "direction2": { - "min": float, - "avg": float, - "max": float, - "hdrh": str - } - } - } - } - - # TCP tests: - "ID": { - "name": "Test name", - "parent": "Name of the parent of the test", - "doc": "Test documentation", - "msg": "Test message", - "tags": ["tag 1", "tag 2", "tag n"], - "type": "TCP", - "status": "PASS" | "FAIL", - "result": int - } - - # MRR, BMRR tests: - "ID": { - "name": "Test name", - "parent": "Name of the parent of the test", - "doc": "Test documentation", - "msg": "Test message", - "tags": ["tag 1", "tag 2", "tag n"], - "type": "MRR" | "BMRR", - "status": "PASS" | "FAIL", - "result": { - "receive-rate": float, - # Average of a list, computed using AvgStdevStats. - # In CSIT-1180, replace with List[float]. - } - } - - "ID" { - # next test - } - } - } - - - Functional tests: - - { - "metadata": { # Optional - "version": "VPP version", - "job": "Jenkins job name", - "build": "Information about the build" - }, - "suites": { - "Suite name 1": { - "doc": "Suite 1 documentation", - "parent": "Suite 1 parent", - "level": "Level of the suite in the suite hierarchy" - } - "Suite name N": { - "doc": "Suite N documentation", - "parent": "Suite 2 parent", - "level": "Level of the suite in the suite hierarchy" - } - } - "tests": { - "ID": { - "name": "Test name", - "parent": "Name of the parent of the test", - "doc": "Test documentation" - "msg": "Test message" - "tags": ["tag 1", "tag 2", "tag n"], - "conf-history": "DUT1 and DUT2 VAT History" - "show-run": "Show Run" - "status": "PASS" | "FAIL" - }, - "ID" { - # next test - } - } - } - - .. note:: ID is the lowercase full path to the test. - """ - - REGEX_PLR_RATE = re.compile( - r'PLRsearch lower bound::?\s(\d+.\d+).*\n' - r'PLRsearch upper bound::?\s(\d+.\d+)' - ) - REGEX_NDRPDR_RATE = re.compile( - r'NDR_LOWER:\s(\d+.\d+).*\n.*\n' - r'NDR_UPPER:\s(\d+.\d+).*\n' - r'PDR_LOWER:\s(\d+.\d+).*\n.*\n' - r'PDR_UPPER:\s(\d+.\d+)' - ) - REGEX_NDRPDR_GBPS = re.compile( - r'NDR_LOWER:.*,\s(\d+.\d+).*\n.*\n' - r'NDR_UPPER:.*,\s(\d+.\d+).*\n' - r'PDR_LOWER:.*,\s(\d+.\d+).*\n.*\n' - r'PDR_UPPER:.*,\s(\d+.\d+)' - ) - REGEX_PERF_MSG_INFO = re.compile( - r'NDR_LOWER:\s(\d+.\d+)\s.*\s(\d+.\d+)\s.*\n.*\n.*\n' - r'PDR_LOWER:\s(\d+.\d+)\s.*\s(\d+.\d+)\s.*\n.*\n.*\n' - r'Latency at 90% PDR:.*\[\'(.*)\', \'(.*)\'\].*\n' - r'Latency at 50% PDR:.*\[\'(.*)\', \'(.*)\'\].*\n' - r'Latency at 10% PDR:.*\[\'(.*)\', \'(.*)\'\].*\n' - ) - REGEX_CPS_MSG_INFO = re.compile( - r'NDR_LOWER:\s(\d+.\d+)\s.*\s.*\n.*\n.*\n' - r'PDR_LOWER:\s(\d+.\d+)\s.*\s.*\n.*\n.*' - ) - REGEX_PPS_MSG_INFO = re.compile( - r'NDR_LOWER:\s(\d+.\d+)\s.*\s(\d+.\d+)\s.*\n.*\n.*\n' - r'PDR_LOWER:\s(\d+.\d+)\s.*\s(\d+.\d+)\s.*\n.*\n.*' - ) - 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' - r'LATENCY.*\[\'(.*)\', \'(.*)\'\]' - ) - REGEX_NDRPDR_LAT = re.compile( - r'LATENCY.*\[\'(.*)\', \'(.*)\'\]\s\n.*\n.*\n' - r'LATENCY.*\[\'(.*)\', \'(.*)\'\]\s\n.*\n' - r'Latency.*\[\'(.*)\', \'(.*)\'\]\s\n' - r'Latency.*\[\'(.*)\', \'(.*)\'\]\s\n' - r'Latency.*\[\'(.*)\', \'(.*)\'\]\s\n' - r'Latency.*\[\'(.*)\', \'(.*)\'\]' - ) - - REGEX_VERSION_VPP = re.compile( - r"(return STDOUT Version:\s*|" - r"VPP Version:\s*|VPP version:\s*)(.*)" - ) - REGEX_VERSION_DPDK = re.compile( - r"(DPDK version:\s*|DPDK Version:\s*)(.*)" - ) - REGEX_TCP = re.compile( - r'Total\s(rps|cps|throughput):\s(\d*).*$' - ) - REGEX_MRR = re.compile( - r'MaxReceivedRate_Results\s\[pkts/(\d*)sec\]:\s' - r'tx\s(\d*),\srx\s(\d*)' - ) - REGEX_BMRR = re.compile( - r'.*trial results.*: \[(.*)\]' - ) - REGEX_RECONF_LOSS = re.compile( - r'Packets lost due to reconfig: (\d*)' - ) - REGEX_RECONF_TIME = re.compile( - r'Implied time lost: (\d*.[\de-]*)' - ) - REGEX_TC_TAG = 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+.) - (.*)\)') - - 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, process_oper): - """Initialisation. - - :param metadata: Key-value pairs to be included in "metadata" part of - JSON structure. - :param mapping: Mapping of the old names of test cases to the new - (actual) one. - :param ignore: List of TCs to be ignored. - :param process_oper: If True, operational data (show run, telemetry) is - processed. - :type metadata: dict - :type mapping: dict - :type ignore: list - :type process_oper: bool - """ - - # Type of message to parse out from the test messages - self._msg_type = None - - # VPP version - self._version = None - - # Timestamp - self._timestamp = None - - # Testbed. The testbed is identified by TG node IP address. - self._testbed = None - - # Mapping of TCs long names - self._mapping = mapping - - # Ignore list - self._ignore = ignore - - self._process_oper = process_oper - - # Number of PAPI History messages found: - # 0 - no message - # 1 - PAPI History of DUT1 - # 2 - PAPI History of DUT2 - 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 - self._test_id = None - - # The main data structure - self._data = { - u"metadata": OrderedDict(), - u"suites": OrderedDict(), - u"tests": OrderedDict() - } - - # Save the provided metadata - for key, val in metadata.items(): - self._data[u"metadata"][key] = val - - # Dictionary defining the methods used to parse different types of - # messages - self.parse_msg = { - 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"test-telemetry": self._get_telemetry - } - - @property - def data(self): - """Getter - Data parsed from the XML file. - - :returns: Data parsed from the XML file. - :rtype: dict - """ - return self._data - - def _get_data_from_mrr_test_msg(self, msg): - """Get info from message of MRR performance tests. - - :param msg: Message to be processed. - :type msg: str - :returns: Processed message or original message if a problem occurs. - :rtype: str - """ - - groups = re.search(self.REGEX_MRR_MSG_INFO, msg) - if not groups or groups.lastindex != 1: - return u"Test Failed." - - try: - data = groups.group(1).split(u", ") - except (AttributeError, IndexError, ValueError, KeyError): - return u"Test Failed." - - out_str = u"[" - try: - for item in data: - out_str += f"{(float(item) / 1e6):.2f}, " - return out_str[:-2] + u"]" - except (AttributeError, IndexError, ValueError, KeyError): - return u"Test Failed." - - def _get_data_from_cps_test_msg(self, msg): - """Get info from message of NDRPDR CPS tests. - - :param msg: Message to be processed. - :type msg: str - :returns: Processed message or "Test Failed." if a problem occurs. - :rtype: str - """ - - groups = re.search(self.REGEX_CPS_MSG_INFO, msg) - if not groups or groups.lastindex != 2: - return u"Test Failed." - - try: - return ( - f"1. {(float(groups.group(1)) / 1e6):5.2f}\n" - f"2. {(float(groups.group(2)) / 1e6):5.2f}" - ) - except (AttributeError, IndexError, ValueError, KeyError): - return u"Test Failed." - - def _get_data_from_pps_test_msg(self, msg): - """Get info from message of NDRPDR PPS tests. - - :param msg: Message to be processed. - :type msg: str - :returns: Processed message or "Test Failed." if a problem occurs. - :rtype: str - """ - - groups = re.search(self.REGEX_PPS_MSG_INFO, msg) - if not groups or groups.lastindex != 4: - return u"Test Failed." - - try: - return ( - f"1. {(float(groups.group(1)) / 1e6):5.2f} " - f"{float(groups.group(2)):5.2f}\n" - f"2. {(float(groups.group(3)) / 1e6):5.2f} " - f"{float(groups.group(4)):5.2f}" - ) - except (AttributeError, IndexError, ValueError, KeyError): - return u"Test Failed." - - def _get_data_from_perf_test_msg(self, msg): - """Get info from message of NDRPDR performance tests. - - :param msg: Message to be processed. - :type msg: str - :returns: Processed message or "Test Failed." if a problem occurs. - :rtype: str - """ - - groups = re.search(self.REGEX_PERF_MSG_INFO, msg) - if not groups or groups.lastindex != 10: - return u"Test Failed." - - try: - data = { - u"ndr_low": float(groups.group(1)), - u"ndr_low_b": float(groups.group(2)), - u"pdr_low": float(groups.group(3)), - u"pdr_low_b": float(groups.group(4)), - u"pdr_lat_90_1": groups.group(5), - u"pdr_lat_90_2": groups.group(6), - u"pdr_lat_50_1": groups.group(7), - u"pdr_lat_50_2": groups.group(8), - u"pdr_lat_10_1": groups.group(9), - u"pdr_lat_10_2": groups.group(10), - } - except (AttributeError, IndexError, ValueError, KeyError): - return u"Test Failed." - - def _process_lat(in_str_1, in_str_2): - """Extract P50, P90 and P99 latencies or min, avg, max values from - latency string. - - :param in_str_1: Latency string for one direction produced by robot - framework. - :param in_str_2: Latency string for second direction produced by - robot framework. - :type in_str_1: str - :type in_str_2: str - :returns: Processed latency string or None if a problem occurs. - :rtype: tuple - """ - in_list_1 = in_str_1.split('/', 3) - in_list_2 = in_str_2.split('/', 3) - - if len(in_list_1) != 4 and len(in_list_2) != 4: - return None - - in_list_1[3] += u"=" * (len(in_list_1[3]) % 4) - try: - hdr_lat_1 = hdrh.histogram.HdrHistogram.decode(in_list_1[3]) - except hdrh.codec.HdrLengthException: - hdr_lat_1 = None - - in_list_2[3] += u"=" * (len(in_list_2[3]) % 4) - try: - hdr_lat_2 = hdrh.histogram.HdrHistogram.decode(in_list_2[3]) - except hdrh.codec.HdrLengthException: - hdr_lat_2 = None - - if hdr_lat_1 and hdr_lat_2: - hdr_lat = ( - hdr_lat_1.get_value_at_percentile(50.0), - hdr_lat_1.get_value_at_percentile(90.0), - hdr_lat_1.get_value_at_percentile(99.0), - hdr_lat_2.get_value_at_percentile(50.0), - hdr_lat_2.get_value_at_percentile(90.0), - hdr_lat_2.get_value_at_percentile(99.0) - ) - if all(hdr_lat): - return hdr_lat - - hdr_lat = ( - int(in_list_1[0]), int(in_list_1[1]), int(in_list_1[2]), - int(in_list_2[0]), int(in_list_2[1]), int(in_list_2[2]) - ) - for item in hdr_lat: - if item in (-1, 4294967295, 0): - return None - return hdr_lat - - try: - out_msg = ( - f"1. {(data[u'ndr_low'] / 1e6):5.2f} " - f"{data[u'ndr_low_b']:5.2f}" - f"\n2. {(data[u'pdr_low'] / 1e6):5.2f} " - f"{data[u'pdr_low_b']:5.2f}" - ) - latency = ( - _process_lat(data[u'pdr_lat_10_1'], data[u'pdr_lat_10_2']), - _process_lat(data[u'pdr_lat_50_1'], data[u'pdr_lat_50_2']), - _process_lat(data[u'pdr_lat_90_1'], data[u'pdr_lat_90_2']) - ) - if all(latency): - max_len = len(str(max((max(item) for item in latency)))) - max_len = 4 if max_len < 4 else max_len - - for idx, lat in enumerate(latency): - if not idx: - out_msg += u"\n" - out_msg += ( - f"\n{idx + 3}. " - f"{lat[0]:{max_len}d} " - f"{lat[1]:{max_len}d} " - f"{lat[2]:{max_len}d} " - f"{lat[3]:{max_len}d} " - f"{lat[4]:{max_len}d} " - f"{lat[5]:{max_len}d} " - ) - - return out_msg - - except (AttributeError, IndexError, ValueError, KeyError): - return u"Test Failed." - - def _get_testbed(self, msg): - """Called when extraction of testbed IP is required. - The testbed is identified by TG node IP address. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - if msg.message.count(u"Setup of TG node") or \ - msg.message.count(u"Setup of node TG host"): - reg_tg_ip = re.compile( - r'.*TG .* (\d{1,3}.\d{1,3}.\d{1,3}.\d{1,3}).*') - try: - self._testbed = str(re.search(reg_tg_ip, msg.message).group(1)) - except (KeyError, ValueError, IndexError, AttributeError): - pass - finally: - self._data[u"metadata"][u"testbed"] = self._testbed - self._msg_type = None - - def _get_vpp_version(self, msg): - """Called when extraction of VPP version is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - if msg.message.count(u"return STDOUT Version:") or \ - msg.message.count(u"VPP Version:") or \ - msg.message.count(u"VPP version:"): - self._version = str( - re.search(self.REGEX_VERSION_VPP, msg.message).group(2) - ) - self._data[u"metadata"][u"version"] = self._version - self._msg_type = None - - def _get_dpdk_version(self, msg): - """Called when extraction of DPDK version is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - if msg.message.count(u"DPDK Version:"): - try: - self._version = str(re.search( - self.REGEX_VERSION_DPDK, msg.message).group(2)) - self._data[u"metadata"][u"version"] = self._version - except IndexError: - pass - finally: - self._msg_type = None - - def _get_papi_history(self, msg): - """Called when extraction of PAPI command history is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - if msg.message.count(u"PAPI command history:"): - self._conf_history_lookup_nr += 1 - if self._conf_history_lookup_nr == 1: - self._data[u"tests"][self._test_id][u"conf-history"] = str() - else: - self._msg_type = None - text = re.sub( - r"\d{1,3}.\d{1,3}.\d{1,3}.\d{1,3} PAPI command history:", - u"", - msg.message, - count=1 - ).replace(u'"', u"'") - self._data[u"tests"][self._test_id][u"conf-history"] += ( - f"**DUT{str(self._conf_history_lookup_nr)}:** {text}" - ) - - def _get_show_run(self, msg): - """Called when extraction of VPP operational data (output of CLI command - Show Runtime) is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - if not msg.message.count(u"stats runtime"): - return - - # Temporary solution - if self._sh_run_counter > 1: - return - - if u"show-run" not in self._data[u"tests"][self._test_id].keys(): - self._data[u"tests"][self._test_id][u"show-run"] = dict() - - groups = re.search(self.REGEX_TC_PAPI_CLI, 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"" - - dut = u"dut{nr}".format( - nr=len(self._data[u'tests'][self._test_id][u'show-run'].keys()) + 1) - - 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] - } - ) - - def _get_telemetry(self, msg): - """Called when extraction of VPP telemetry data is required. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - if self._telemetry_kw_counter > 1: - return - if not msg.message.count(u"# TYPE vpp_runtime_calls"): - return - - 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 - message. - - :param msg: The test message to be parsed. - :type msg: str - :returns: Parsed data as a dict and the status (PASS/FAIL). - :rtype: tuple(dict, str) - """ - - throughput = { - u"NDR": {u"LOWER": -1.0, u"UPPER": -1.0}, - u"PDR": {u"LOWER": -1.0, u"UPPER": -1.0} - } - status = u"FAIL" - groups = re.search(self.REGEX_NDRPDR_RATE, msg) - - if groups is not None: - try: - throughput[u"NDR"][u"LOWER"] = float(groups.group(1)) - throughput[u"NDR"][u"UPPER"] = float(groups.group(2)) - throughput[u"PDR"][u"LOWER"] = float(groups.group(3)) - throughput[u"PDR"][u"UPPER"] = float(groups.group(4)) - status = u"PASS" - except (IndexError, ValueError): - pass - - return throughput, status - - def _get_ndrpdr_throughput_gbps(self, msg): - """Get NDR_LOWER, NDR_UPPER, PDR_LOWER and PDR_UPPER in Gbps from the - test message. - - :param msg: The test message to be parsed. - :type msg: str - :returns: Parsed data as a dict and the status (PASS/FAIL). - :rtype: tuple(dict, str) - """ - - gbps = { - u"NDR": {u"LOWER": -1.0, u"UPPER": -1.0}, - u"PDR": {u"LOWER": -1.0, u"UPPER": -1.0} - } - status = u"FAIL" - groups = re.search(self.REGEX_NDRPDR_GBPS, msg) - - if groups is not None: - try: - gbps[u"NDR"][u"LOWER"] = float(groups.group(1)) - gbps[u"NDR"][u"UPPER"] = float(groups.group(2)) - gbps[u"PDR"][u"LOWER"] = float(groups.group(3)) - gbps[u"PDR"][u"UPPER"] = float(groups.group(4)) - status = u"PASS" - except (IndexError, ValueError): - pass - - return gbps, status - - def _get_plr_throughput(self, msg): - """Get PLRsearch lower bound and PLRsearch upper bound from the test - message. - - :param msg: The test message to be parsed. - :type msg: str - :returns: Parsed data as a dict and the status (PASS/FAIL). - :rtype: tuple(dict, str) - """ - - throughput = { - u"LOWER": -1.0, - u"UPPER": -1.0 - } - status = u"FAIL" - groups = re.search(self.REGEX_PLR_RATE, msg) - - if groups is not None: - try: - throughput[u"LOWER"] = float(groups.group(1)) - throughput[u"UPPER"] = float(groups.group(2)) - status = u"PASS" - except (IndexError, ValueError): - pass - - return throughput, status - - def _get_ndrpdr_latency(self, msg): - """Get LATENCY from the test message. - - :param msg: The test message to be parsed. - :type msg: str - :returns: Parsed data as a dict and the status (PASS/FAIL). - :rtype: tuple(dict, str) - """ - latency_default = { - u"min": -1.0, - u"avg": -1.0, - u"max": -1.0, - u"hdrh": u"" - } - latency = { - u"NDR": { - u"direction1": copy.copy(latency_default), - u"direction2": copy.copy(latency_default) - }, - u"PDR": { - u"direction1": copy.copy(latency_default), - u"direction2": copy.copy(latency_default) - }, - u"LAT0": { - u"direction1": copy.copy(latency_default), - u"direction2": copy.copy(latency_default) - }, - u"PDR10": { - u"direction1": copy.copy(latency_default), - u"direction2": copy.copy(latency_default) - }, - u"PDR50": { - u"direction1": copy.copy(latency_default), - u"direction2": copy.copy(latency_default) - }, - u"PDR90": { - u"direction1": copy.copy(latency_default), - u"direction2": copy.copy(latency_default) - }, - } - - groups = re.search(self.REGEX_NDRPDR_LAT, msg) - if groups is None: - groups = re.search(self.REGEX_NDRPDR_LAT_BASE, msg) - if groups is None: - return latency, u"FAIL" - - def process_latency(in_str): - """Return object with parsed latency values. - - TODO: Define class for the return type. - - :param in_str: Input string, min/avg/max/hdrh format. - :type in_str: str - :returns: Dict with corresponding keys, except hdrh float values. - :rtype dict: - :throws IndexError: If in_str does not have enough substrings. - :throws ValueError: If a substring does not convert to float. - """ - in_list = in_str.split('/', 3) - - rval = { - u"min": float(in_list[0]), - u"avg": float(in_list[1]), - u"max": float(in_list[2]), - u"hdrh": u"" - } - - if len(in_list) == 4: - rval[u"hdrh"] = str(in_list[3]) - - return rval - - try: - latency[u"NDR"][u"direction1"] = process_latency(groups.group(1)) - latency[u"NDR"][u"direction2"] = process_latency(groups.group(2)) - latency[u"PDR"][u"direction1"] = process_latency(groups.group(3)) - latency[u"PDR"][u"direction2"] = process_latency(groups.group(4)) - if groups.lastindex == 4: - return latency, u"PASS" - except (IndexError, ValueError): - pass - - try: - latency[u"PDR90"][u"direction1"] = process_latency(groups.group(5)) - latency[u"PDR90"][u"direction2"] = process_latency(groups.group(6)) - latency[u"PDR50"][u"direction1"] = process_latency(groups.group(7)) - latency[u"PDR50"][u"direction2"] = process_latency(groups.group(8)) - latency[u"PDR10"][u"direction1"] = process_latency(groups.group(9)) - latency[u"PDR10"][u"direction2"] = process_latency(groups.group(10)) - latency[u"LAT0"][u"direction1"] = process_latency(groups.group(11)) - latency[u"LAT0"][u"direction2"] = process_latency(groups.group(12)) - if groups.lastindex == 12: - return latency, u"PASS" - except (IndexError, ValueError): - pass - - return latency, u"FAIL" - - @staticmethod - def _get_hoststack_data(msg, tags): - """Get data from the hoststack 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" - - msg = msg.replace(u"'", u'"').replace(u" ", u"") - if u"LDPRELOAD" in tags: - try: - result = loads(msg) - status = u"PASS" - except JSONDecodeError: - pass - elif u"VPPECHO" in tags: - try: - msg_lst = msg.replace(u"}{", u"} {").split(u" ") - result = dict( - client=loads(msg_lst[0]), - server=loads(msg_lst[1]) - ) - status = u"PASS" - except (JSONDecodeError, IndexError): - pass - - 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. - - :param suite: Suite to process. - :type suite: Suite - :returns: Nothing. - """ - if self.start_suite(suite) is not False: - suite.suites.visit(self) - suite.tests.visit(self) - self.end_suite(suite) - - def start_suite(self, suite): - """Called when suite starts. - - :param suite: Suite to process. - :type suite: Suite - :returns: Nothing. - """ - - try: - parent_name = suite.parent.name - except AttributeError: - return - - self._data[u"suites"][suite.longname.lower(). - replace(u'"', u"'"). - replace(u" ", u"_")] = { - u"name": suite.name.lower(), - u"doc": suite.doc, - u"parent": parent_name, - u"level": len(suite.longname.split(u".")) - } - - suite.setup.visit(self) - - def end_suite(self, suite): - """Called when suite ends. - - :param suite: Suite to process. - :type suite: Suite - :returns: Nothing. - """ - - def visit_test(self, test): - """Implements traversing through the test. - - :param test: Test to process. - :type test: Test - :returns: Nothing. - """ - if self.start_test(test) is not False: - test.body.visit(self) - self.end_test(test) - - def start_test(self, test): - """Called when test starts. - - :param test: Test to process. - :type test: Test - :returns: Nothing. - """ - - self._sh_run_counter = 0 - self._telemetry_kw_counter = 0 - self._telemetry_msg_counter = 0 - - longname_orig = test.longname.lower() - - # Check the ignore list - if longname_orig in self._ignore: - return - - tags = [str(tag) for tag in test.tags] - test_result = dict() - - # Change the TC long name and name if defined in the mapping table - longname = self._mapping.get(longname_orig, None) - if longname is not None: - name = longname.split(u'.')[-1] - logging.debug( - f"{self._data[u'metadata']}\n{longname_orig}\n{longname}\n" - f"{name}" - ) - else: - longname = longname_orig - name = test.name.lower() - - # Remove TC number from the TC long name (backward compatibility): - self._test_id = re.sub(self.REGEX_TC_NUMBER, u"", longname) - # Remove TC number from the TC name (not needed): - test_result[u"name"] = re.sub(self.REGEX_TC_NUMBER, "", name) - - test_result[u"parent"] = test.parent.name.lower() - test_result[u"tags"] = tags - test_result["doc"] = test.doc - test_result[u"type"] = u"" - test_result[u"status"] = test.status - test_result[u"starttime"] = test.starttime - test_result[u"endtime"] = test.endtime - - if test.status == u"PASS": - if u"NDRPDR" in tags: - if u"TCP_PPS" in tags or u"UDP_PPS" in tags: - test_result[u"msg"] = self._get_data_from_pps_test_msg( - test.message) - elif u"TCP_CPS" in tags or u"UDP_CPS" in tags: - test_result[u"msg"] = self._get_data_from_cps_test_msg( - test.message) - else: - test_result[u"msg"] = self._get_data_from_perf_test_msg( - test.message) - elif u"MRR" in tags or u"FRMOBL" in tags or u"BMRR" in tags: - test_result[u"msg"] = self._get_data_from_mrr_test_msg( - test.message) - else: - test_result[u"msg"] = test.message - else: - test_result[u"msg"] = test.message - - 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. - 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" - elif u"NDRPDR" in tags: - if u"TCP_CPS" in tags or u"UDP_CPS" in tags: - test_result[u"type"] = u"CPS" - else: - test_result[u"type"] = u"NDRPDR" - if test.status == u"PASS": - test_result[u"throughput"], test_result[u"status"] = \ - self._get_ndrpdr_throughput(test.message) - test_result[u"gbps"], test_result[u"status"] = \ - self._get_ndrpdr_throughput_gbps(test.message) - test_result[u"latency"], test_result[u"status"] = \ - self._get_ndrpdr_latency(test.message) - elif u"MRR" in tags or u"FRMOBL" in tags or u"BMRR" in tags: - if u"MRR" in tags: - test_result[u"type"] = u"MRR" - else: - test_result[u"type"] = u"BMRR" - if test.status == u"PASS": - test_result[u"result"] = dict() - groups = re.search(self.REGEX_BMRR, test.message) - if groups is not None: - items_str = groups.group(1) - items_float = [ - float(item.strip().replace(u"'", u"")) - for item in items_str.split(",") - ] - # Use whole list in CSIT-1180. - stats = jumpavg.AvgStdevStats.for_runs(items_float) - test_result[u"result"][u"samples"] = items_float - test_result[u"result"][u"receive-rate"] = stats.avg - test_result[u"result"][u"receive-stdev"] = stats.stdev - else: - groups = re.search(self.REGEX_MRR, test.message) - test_result[u"result"][u"receive-rate"] = \ - float(groups.group(3)) / float(groups.group(1)) - elif u"SOAK" in tags: - test_result[u"type"] = u"SOAK" - if test.status == u"PASS": - test_result[u"throughput"], test_result[u"status"] = \ - self._get_plr_throughput(test.message) - 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"HOSTSTACK" in tags: - test_result[u"type"] = u"HOSTSTACK" - if test.status == u"PASS": - test_result[u"result"], test_result[u"status"] = \ - self._get_hoststack_data(test.message, tags) - elif u"RECONF" in tags: - test_result[u"type"] = u"RECONF" - if test.status == u"PASS": - test_result[u"result"] = None - try: - grps_loss = re.search(self.REGEX_RECONF_LOSS, test.message) - grps_time = re.search(self.REGEX_RECONF_TIME, test.message) - test_result[u"result"] = { - u"loss": int(grps_loss.group(1)), - u"time": float(grps_time.group(1)) - } - except (AttributeError, IndexError, ValueError, TypeError): - test_result[u"status"] = u"FAIL" - else: - test_result[u"status"] = u"FAIL" - - self._data[u"tests"][self._test_id] = test_result - - def end_test(self, test): - """Called when test ends. - - :param test: Test to process. - :type test: Test - :returns: Nothing. - """ - - def visit_keyword(self, keyword): - """Implements traversing through the keyword and its child keywords. - - :param keyword: Keyword to process. - :type keyword: Keyword - :returns: Nothing. - """ - if self.start_keyword(keyword) is not False: - self.end_keyword(keyword) - - def start_keyword(self, keyword): - """Called when keyword starts. Default implementation does nothing. - - :param keyword: Keyword to process. - :type keyword: Keyword - :returns: Nothing. - """ - try: - if keyword.type == u"setup": - self.visit_setup_kw(keyword) - elif keyword.type == u"teardown": - self.visit_teardown_kw(keyword) - else: - self.visit_test_kw(keyword) - except AttributeError: - pass - - def end_keyword(self, keyword): - """Called when keyword ends. Default implementation does nothing. - - :param keyword: Keyword to process. - :type keyword: Keyword - :returns: Nothing. - """ - - def visit_test_kw(self, test_kw): - """Implements traversing through the test keyword and its child - keywords. - - :param test_kw: Keyword to process. - :type test_kw: Keyword - :returns: Nothing. - """ - for keyword in test_kw.body: - if self.start_test_kw(keyword) is not False: - self.visit_test_kw(keyword) - self.end_test_kw(keyword) - - def start_test_kw(self, test_kw): - """Called when test keyword starts. Default implementation does - nothing. - - :param test_kw: Keyword to process. - :type test_kw: Keyword - :returns: Nothing. - """ - if not self._process_oper: - 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: - return - test_kw.messages.visit(self) - - def end_test_kw(self, test_kw): - """Called when keyword ends. Default implementation does nothing. - - :param test_kw: Keyword to process. - :type test_kw: Keyword - :returns: Nothing. - """ - - def visit_setup_kw(self, setup_kw): - """Implements traversing through the teardown keyword and its child - keywords. - - :param setup_kw: Keyword to process. - :type setup_kw: Keyword - :returns: Nothing. - """ - for keyword in setup_kw.body: - if self.start_setup_kw(keyword) is not False: - self.visit_setup_kw(keyword) - self.end_setup_kw(keyword) - - def start_setup_kw(self, setup_kw): - """Called when teardown keyword starts. Default implementation does - nothing. - - :param setup_kw: Keyword to process. - :type setup_kw: Keyword - :returns: Nothing. - """ - if setup_kw.name.count(u"Show Vpp Version On All Duts") \ - and not self._version: - self._msg_type = u"vpp-version" - 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"Setup Framework") and not self._testbed: - self._msg_type = u"testbed" - else: - return - setup_kw.messages.visit(self) - - def end_setup_kw(self, setup_kw): - """Called when keyword ends. Default implementation does nothing. - - :param setup_kw: Keyword to process. - :type setup_kw: Keyword - :returns: Nothing. - """ - - def visit_teardown_kw(self, teardown_kw): - """Implements traversing through the teardown keyword and its child - keywords. - - :param teardown_kw: Keyword to process. - :type teardown_kw: Keyword - :returns: Nothing. - """ - for keyword in teardown_kw.body: - if self.start_teardown_kw(keyword) is not False: - self.visit_teardown_kw(keyword) - self.end_teardown_kw(keyword) - - def start_teardown_kw(self, teardown_kw): - """Called when teardown keyword starts - - :param teardown_kw: Keyword to process. - :type teardown_kw: Keyword - :returns: Nothing. - """ - if teardown_kw.name.count(u"Show Papi History On All Duts"): - self._conf_history_lookup_nr = 0 - self._msg_type = u"teardown-papi-history" - teardown_kw.messages.visit(self) - - def end_teardown_kw(self, teardown_kw): - """Called when keyword ends. Default implementation does nothing. - - :param teardown_kw: Keyword to process. - :type teardown_kw: Keyword - :returns: Nothing. - """ - - def visit_message(self, msg): - """Implements visiting the message. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - if self.start_message(msg) is not False: - self.end_message(msg) - - def start_message(self, msg): - """Called when message starts. Get required information from messages: - - VPP version. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - if self._msg_type: - self.parse_msg[self._msg_type](msg) - - def end_message(self, msg): - """Called when message ends. Default implementation does nothing. - - :param msg: Message to process. - :type msg: Message - :returns: Nothing. - """ - - -class InputData: - """Input data - - The data is extracted from output.xml files generated by Jenkins jobs and - stored in pandas' DataFrames. - - The data structure: - - job name - - build number - - metadata - (as described in ExecutionChecker documentation) - - suites - (as described in ExecutionChecker documentation) - - tests - (as described in ExecutionChecker documentation) - """ - - 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(dtype="object") - - @property - def data(self): - """Getter - Input data. - - :returns: Input data - :rtype: pandas.Series - """ - return self._input_data - - def metadata(self, job, build): - """Getter - metadata - - :param job: Job which metadata we want. - :param build: Build which metadata we want. - :type job: str - :type build: str - :returns: Metadata - :rtype: pandas.Series - """ - return self.data[job][build][u"metadata"] - - def suites(self, job, build): - """Getter - suites - - :param job: Job which suites we want. - :param build: Build which suites we want. - :type job: str - :type build: str - :returns: Suites. - :rtype: pandas.Series - """ - return self.data[job][str(build)][u"suites"] - - def tests(self, job, build): - """Getter - tests - - :param job: Job which tests we want. - :param build: Build which tests we want. - :type job: str - :type build: str - :returns: Tests. - :rtype: pandas.Series - """ - return self.data[job][build][u"tests"] - - def _parse_tests(self, job, build): - """Process data from robot output.xml file and return JSON structured - data. - - :param job: The name of job which build output data will be processed. - :param build: The build which output data will be processed. - :type job: str - :type build: dict - :returns: JSON data structure. - :rtype: dict - """ - - metadata = { - u"job": job, - u"build": build - } - - with open(build[u"file-name"], u'r') as data_file: - try: - result = ExecutionResult(data_file) - except errors.DataError as err: - logging.error( - f"Error occurred while parsing output.xml: {repr(err)}" - ) - return None - - process_oper = False - if u"-vpp-perf-report-coverage-" in job: - process_oper = True - # elif u"-vpp-perf-report-iterative-" in job: - # # Exceptions for TBs where we do not have coverage data: - # for item in (u"-2n-icx", ): - # if item in job: - # process_oper = True - # break - checker = ExecutionChecker( - metadata, self._cfg.mapping, self._cfg.ignore, process_oper - ) - result.visit(checker) - - checker.data[u"metadata"][u"tests_total"] = \ - result.statistics.total.total - checker.data[u"metadata"][u"tests_passed"] = \ - result.statistics.total.passed - checker.data[u"metadata"][u"tests_failed"] = \ - result.statistics.total.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): - """Download and parse the input data file. - - :param pid: PID of the process executing this method. - :param job: Name of the Jenkins job which generated the processed input - file. - :param build: Information about the Jenkins build which generated the - processed input file. - :param repeat: Repeat the download specified number of times if not - successful. - :type pid: int - :type job: str - :type build: dict - :type repeat: int - """ - - logging.info(f"Processing the job/build: {job}: {build[u'build']}") - - state = u"failed" - success = False - data = None - do_repeat = repeat - while do_repeat: - success = download_and_unzip_data_file(self._cfg, job, build, pid) - if success: - break - do_repeat -= 1 - if not success: - logging.error( - f"It is not possible to download the input data file from the " - f"job {job}, build {build[u'build']}, or it is damaged. " - f"Skipped." - ) - if success: - logging.info(f" Processing data from build {build[u'build']}") - data = self._parse_tests(job, build) - if data is None: - logging.error( - f"Input data file from the job {job}, build " - f"{build[u'build']} is damaged. Skipped." - ) - else: - state = u"processed" - - try: - remove(build[u"file-name"]) - except OSError as err: - logging.error( - f"Cannot remove the file {build[u'file-name']}: {repr(err)}" - ) - - # If the time-period is defined in the specification file, remove all - # files which are outside the time period. - is_last = False - timeperiod = self._cfg.environment.get(u"time-period", None) - if timeperiod and data: - now = dt.utcnow() - timeperiod = timedelta(int(timeperiod)) - metadata = data.get(u"metadata", None) - if metadata: - generated = metadata.get(u"generated", None) - if generated: - generated = dt.strptime(generated, u"%Y%m%d %H:%M") - if (now - generated) > timeperiod: - # Remove the data and the file: - state = u"removed" - data = None - is_last = True - logging.info( - f" The build {job}/{build[u'build']} is " - f"outdated, will be removed." - ) - return { - u"data": data, - u"state": state, - u"job": job, - u"build": build, - u"last": is_last - } - - def download_and_parse_data(self, repeat=1): - """Download the input data files, parse input data from input files and - store in pandas' Series. - - :param repeat: Repeat the download specified number of times if not - successful. - :type repeat: int - """ - - logging.info(u"Downloading and parsing input files ...") - - for job, builds in self._cfg.input.items(): - for build in builds: - - result = self._download_and_parse_build(job, build, repeat) - if result[u"last"]: - break - build_nr = result[u"build"][u"build"] - - if result[u"data"]: - data = result[u"data"] - build_data = pd.Series({ - u"metadata": pd.Series( - list(data[u"metadata"].values()), - index=list(data[u"metadata"].keys()) - ), - u"suites": pd.Series( - list(data[u"suites"].values()), - index=list(data[u"suites"].keys()) - ), - u"tests": pd.Series( - list(data[u"tests"].values()), - index=list(data[u"tests"].keys()) - ) - }) - - if self._input_data.get(job, None) is None: - self._input_data[job] = pd.Series(dtype="object") - self._input_data[job][str(build_nr)] = build_data - self._cfg.set_input_file_name( - job, build_nr, result[u"build"][u"file-name"] - ) - self._cfg.set_input_state(job, build_nr, result[u"state"]) - - mem_alloc = \ - resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1000 - logging.info(f"Memory allocation: {mem_alloc:.0f}MB") - - logging.info(u"Done.") - - msg = f"Successful downloads from the sources:\n" - for source in self._cfg.environment[u"data-sources"]: - if source[u"successful-downloads"]: - msg += ( - f"{source[u'url']}/{source[u'path']}/" - f"{source[u'file-name']}: " - f"{source[u'successful-downloads']}\n" - ) - logging.info(msg) - - def process_local_file(self, local_file, job=u"local", build_nr=1, - replace=True): - """Process local XML file given as a command-line parameter. - - :param local_file: The file to process. - :param job: Job name. - :param build_nr: Build number. - :param replace: If True, the information about jobs and builds is - replaced by the new one, otherwise the new jobs and builds are - added. - :type local_file: str - :type job: str - :type build_nr: int - :type replace: bool - :raises: PresentationError if an error occurs. - """ - if not isfile(local_file): - raise PresentationError(f"The file {local_file} does not exist.") - - try: - build_nr = int(local_file.split(u"/")[-1].split(u".")[0]) - except (IndexError, ValueError): - pass - - build = { - u"build": build_nr, - u"status": u"failed", - u"file-name": local_file - } - if replace: - self._cfg.input = dict() - self._cfg.add_build(job, build) - - logging.info(f"Processing {job}: {build_nr:2d}: {local_file}") - data = self._parse_tests(job, build) - if data is None: - raise PresentationError( - f"Error occurred while parsing the file {local_file}" - ) - - build_data = pd.Series({ - u"metadata": pd.Series( - list(data[u"metadata"].values()), - index=list(data[u"metadata"].keys()) - ), - u"suites": pd.Series( - list(data[u"suites"].values()), - index=list(data[u"suites"].keys()) - ), - u"tests": pd.Series( - list(data[u"tests"].values()), - index=list(data[u"tests"].keys()) - ) - }) - - if self._input_data.get(job, None) is None: - self._input_data[job] = pd.Series(dtype="object") - self._input_data[job][str(build_nr)] = build_data - - self._cfg.set_input_state(job, build_nr, u"processed") - - def process_local_directory(self, local_dir, replace=True): - """Process local directory with XML file(s). The directory is processed - as a 'job' and the XML files in it as builds. - If the given directory contains only sub-directories, these - sub-directories processed as jobs and corresponding XML files as builds - of their job. - - :param local_dir: Local directory to process. - :param replace: If True, the information about jobs and builds is - replaced by the new one, otherwise the new jobs and builds are - added. - :type local_dir: str - :type replace: bool - """ - if not isdir(local_dir): - raise PresentationError( - f"The directory {local_dir} does not exist." - ) - - # Check if the given directory includes only files, or only directories - _, dirnames, filenames = next(walk(local_dir)) - - if filenames and not dirnames: - filenames.sort() - # local_builds: - # key: dir (job) name, value: list of file names (builds) - local_builds = { - local_dir: [join(local_dir, name) for name in filenames] - } - - elif dirnames and not filenames: - dirnames.sort() - # local_builds: - # key: dir (job) name, value: list of file names (builds) - local_builds = dict() - for dirname in dirnames: - builds = [ - join(local_dir, dirname, name) - for name in listdir(join(local_dir, dirname)) - if isfile(join(local_dir, dirname, name)) - ] - if builds: - local_builds[dirname] = sorted(builds) - - elif not filenames and not dirnames: - raise PresentationError(f"The directory {local_dir} is empty.") - else: - raise PresentationError( - f"The directory {local_dir} can include only files or only " - f"directories, not both.\nThe directory {local_dir} includes " - f"file(s):\n{filenames}\nand directories:\n{dirnames}" - ) - - if replace: - self._cfg.input = dict() - - for job, files in local_builds.items(): - for idx, local_file in enumerate(files): - self.process_local_file(local_file, job, idx + 1, replace=False) - - @staticmethod - def _end_of_tag(tag_filter, start=0, closer=u"'"): - """Return the index of character in the string which is the end of tag. - - :param tag_filter: The string where the end of tag is being searched. - :param start: The index where the searching is stated. - :param closer: The character which is the tag closer. - :type tag_filter: str - :type start: int - :type closer: str - :returns: The index of the tag closer. - :rtype: int - """ - try: - idx_opener = tag_filter.index(closer, start) - return tag_filter.index(closer, idx_opener + 1) - except ValueError: - return None - - @staticmethod - def _condition(tag_filter): - """Create a conditional statement from the given tag filter. - - :param tag_filter: Filter based on tags from the element specification. - :type tag_filter: str - :returns: Conditional statement which can be evaluated. - :rtype: str - """ - index = 0 - while True: - index = InputData._end_of_tag(tag_filter, index) - if index is None: - return tag_filter - index += 1 - tag_filter = tag_filter[:index] + u" in tags" + tag_filter[index:] - - def filter_data(self, element, params=None, data=None, data_set=u"tests", - continue_on_error=False): - """Filter required data from the given jobs and builds. - - The output data structure is: - - job 1 - - build 1 - - test (or suite) 1 ID: - - param 1 - - param 2 - ... - - param n - ... - - test (or suite) n ID: - ... - ... - - build n - ... - - job n - - :param element: Element which will use the filtered data. - :param params: Parameters which will be included in the output. If None, - all parameters are included. - :param data: If not None, this data is used instead of data specified - in the element. - :param data_set: The set of data to be filtered: tests, suites, - metadata. - :param continue_on_error: Continue if there is error while reading the - data. The Item will be empty then - :type element: pandas.Series - :type params: list - :type data: dict - :type data_set: str - :type continue_on_error: bool - :returns: Filtered data. - :rtype pandas.Series - """ - - try: - if data_set == "suites": - cond = u"True" - elif element[u"filter"] in (u"all", u"template"): - cond = u"True" - else: - cond = InputData._condition(element[u"filter"]) - logging.debug(f" Filter: {cond}") - except KeyError: - logging.error(u" No filter defined.") - return None - - if params is None: - params = element.get(u"parameters", None) - if params: - params.extend((u"type", u"status")) - - data_to_filter = data if data else element[u"data"] - data = pd.Series(dtype="object") - try: - for job, builds in data_to_filter.items(): - data[job] = pd.Series(dtype="object") - for build in builds: - data[job][str(build)] = pd.Series(dtype="object") - try: - data_dict = dict( - self.data[job][str(build)][data_set].items()) - except KeyError: - if continue_on_error: - continue - return None - - for test_id, test_data in data_dict.items(): - if eval(cond, {u"tags": test_data.get(u"tags", u"")}): - data[job][str(build)][test_id] = \ - pd.Series(dtype="object") - if params is None: - for param, val in test_data.items(): - data[job][str(build)][test_id][param] = val - else: - for param in params: - try: - data[job][str(build)][test_id][param] =\ - test_data[param] - except KeyError: - data[job][str(build)][test_id][param] =\ - u"No Data" - return data - - except (KeyError, IndexError, ValueError) as err: - logging.error( - f"Missing mandatory parameter in the element specification: " - f"{repr(err)}" - ) - return None - except AttributeError as err: - logging.error(repr(err)) - return None - except SyntaxError as err: - logging.error( - f"The filter {cond} is not correct. Check if all tags are " - f"enclosed by apostrophes.\n{repr(err)}" - ) - return None - - def filter_tests_by_name(self, element, params=None, data_set=u"tests", - continue_on_error=False): - """Filter required data from the given jobs and builds. - - The output data structure is: - - job 1 - - build 1 - - test (or suite) 1 ID: - - param 1 - - param 2 - ... - - param n - ... - - test (or suite) n ID: - ... - ... - - build n - ... - - job n - - :param element: Element which will use the filtered data. - :param params: Parameters which will be included in the output. If None, - all parameters are included. - :param data_set: The set of data to be filtered: tests, suites, - metadata. - :param continue_on_error: Continue if there is error while reading the - data. The Item will be empty then - :type element: pandas.Series - :type params: list - :type data_set: str - :type continue_on_error: bool - :returns: Filtered data. - :rtype pandas.Series - """ - - include = element.get(u"include", None) - if not include: - logging.warning(u"No tests to include, skipping the element.") - return None - - if params is None: - params = element.get(u"parameters", None) - if params and u"type" not in params: - params.append(u"type") - - cores = element.get(u"core", None) - if cores: - tests = list() - for core in cores: - for test in include: - tests.append(test.format(core=core)) - else: - tests = include - - data = pd.Series(dtype="object") - try: - for job, builds in element[u"data"].items(): - data[job] = pd.Series(dtype="object") - for build in builds: - data[job][str(build)] = pd.Series(dtype="object") - for test in tests: - try: - reg_ex = re.compile(str(test).lower()) - for test_id in self.data[job][ - str(build)][data_set].keys(): - if re.match(reg_ex, str(test_id).lower()): - test_data = self.data[job][ - str(build)][data_set][test_id] - data[job][str(build)][test_id] = \ - pd.Series(dtype="object") - if params is None: - for param, val in test_data.items(): - data[job][str(build)][test_id]\ - [param] = val - else: - for param in params: - try: - data[job][str(build)][ - test_id][param] = \ - test_data[param] - except KeyError: - data[job][str(build)][ - test_id][param] = u"No Data" - except KeyError as err: - if continue_on_error: - logging.debug(repr(err)) - continue - logging.error(repr(err)) - return None - return data - - except (KeyError, IndexError, ValueError) as err: - logging.error( - f"Missing mandatory parameter in the element " - f"specification: {repr(err)}" - ) - return None - except AttributeError as err: - logging.error(repr(err)) - return None - - @staticmethod - def merge_data(data): - """Merge data from more jobs and builds to a simple data structure. - - The output data structure is: - - - test (suite) 1 ID: - - param 1 - - param 2 - ... - - param n - ... - - test (suite) n ID: - ... - - :param data: Data to merge. - :type data: pandas.Series - :returns: Merged data. - :rtype: pandas.Series - """ - - logging.info(u" Merging data ...") - - merged_data = pd.Series(dtype="object") - for builds in data.values: - for item in builds.values: - for item_id, item_data in item.items(): - merged_data[item_id] = item_data - return merged_data - - def print_all_oper_data(self): - """Print all operational data to console. - """ - - for job in self._input_data.values: - for build in job.values: - for test_id, test_data in build[u"tests"].items(): - print(f"{test_id}") - 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"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 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) - avg += row[-1] - if len(thread) == 0: - avg = u"" - else: - avg = f", Average Vector Size per Node: " \ - f"{(avg / len(thread)):.2f}" - th_name = u"main" if thread_nr == 0 \ - else f"worker_{thread_nr}" - print(f"{dut_name}, {th_name}{avg}") - txt_table.float_format = u".2" - txt_table.align = u"r" - txt_table.align[u"Name"] = u"l" - print(f"{txt_table.get_string()}\n") |