| # SPDX-License-Identifier: GPL-2.0 | |
| import re | |
| import csv | |
| import json | |
| import argparse | |
| from pathlib import Path | |
| import subprocess | |
| class TestError: | |
| def __init__(self, metric: list[str], wl: str, value: list[float], low: float, up=float('nan'), description=str()): | |
| self.metric: list = metric # multiple metrics in relationship type tests | |
| self.workloads = [wl] # multiple workloads possible | |
| self.collectedValue: list = value | |
| self.valueLowBound = low | |
| self.valueUpBound = up | |
| self.description = description | |
| def __repr__(self) -> str: | |
| if len(self.metric) > 1: | |
| return "\nMetric Relationship Error: \tThe collected value of metric {0}\n\ | |
| \tis {1} in workload(s): {2} \n\ | |
| \tbut expected value range is [{3}, {4}]\n\ | |
| \tRelationship rule description: \'{5}\'".format(self.metric, self.collectedValue, self.workloads, | |
| self.valueLowBound, self.valueUpBound, self.description) | |
| elif len(self.collectedValue) == 0: | |
| return "\nNo Metric Value Error: \tMetric {0} returns with no value \n\ | |
| \tworkload(s): {1}".format(self.metric, self.workloads) | |
| else: | |
| return "\nWrong Metric Value Error: \tThe collected value of metric {0}\n\ | |
| \tis {1} in workload(s): {2}\n\ | |
| \tbut expected value range is [{3}, {4}]"\ | |
| .format(self.metric, self.collectedValue, self.workloads, | |
| self.valueLowBound, self.valueUpBound) | |
| class Validator: | |
| def __init__(self, rulefname, reportfname='', t=5, debug=False, datafname='', fullrulefname='', | |
| workload='true', metrics='', cputype='cpu'): | |
| self.rulefname = rulefname | |
| self.reportfname = reportfname | |
| self.rules = None | |
| self.collectlist: str = metrics | |
| self.metrics = self.__set_metrics(metrics) | |
| self.skiplist = set() | |
| self.tolerance = t | |
| self.cputype = cputype | |
| self.workloads = [x for x in workload.split(",") if x] | |
| self.wlidx = 0 # idx of current workloads | |
| self.allresults = dict() # metric results of all workload | |
| self.alltotalcnt = dict() | |
| self.allpassedcnt = dict() | |
| self.results = dict() # metric results of current workload | |
| # vars for test pass/failure statistics | |
| # metrics with no results or negative results, neg result counts failed tests | |
| self.ignoremetrics = set() | |
| self.totalcnt = 0 | |
| self.passedcnt = 0 | |
| # vars for errors | |
| self.errlist = list() | |
| # vars for Rule Generator | |
| self.pctgmetrics = set() # Percentage rule | |
| # vars for debug | |
| self.datafname = datafname | |
| self.debug = debug | |
| self.fullrulefname = fullrulefname | |
| def __set_metrics(self, metrics=''): | |
| if metrics != '': | |
| return set(metrics.split(",")) | |
| else: | |
| return set() | |
| def read_json(self, filename: str) -> dict: | |
| try: | |
| with open(Path(filename).resolve(), "r") as f: | |
| data = json.loads(f.read()) | |
| except OSError as e: | |
| print(f"Error when reading file {e}") | |
| sys.exit() | |
| return data | |
| def json_dump(self, data, output_file): | |
| parent = Path(output_file).parent | |
| if not parent.exists(): | |
| parent.mkdir(parents=True) | |
| with open(output_file, "w+") as output_file: | |
| json.dump(data, | |
| output_file, | |
| ensure_ascii=True, | |
| indent=4) | |
| def get_results(self, idx: int = 0): | |
| return self.results.get(idx) | |
| def get_bounds(self, lb, ub, error, alias={}, ridx: int = 0) -> list: | |
| """ | |
| Get bounds and tolerance from lb, ub, and error. | |
| If missing lb, use 0.0; missing ub, use float('inf); missing error, use self.tolerance. | |
| @param lb: str/float, lower bound | |
| @param ub: str/float, upper bound | |
| @param error: float/str, error tolerance | |
| @returns: lower bound, return inf if the lower bound is a metric value and is not collected | |
| upper bound, return -1 if the upper bound is a metric value and is not collected | |
| tolerance, denormalized base on upper bound value | |
| """ | |
| # init ubv and lbv to invalid values | |
| def get_bound_value(bound, initval, ridx): | |
| val = initval | |
| if isinstance(bound, int) or isinstance(bound, float): | |
| val = bound | |
| elif isinstance(bound, str): | |
| if bound == '': | |
| val = float("inf") | |
| elif bound in alias: | |
| vall = self.get_value(alias[ub], ridx) | |
| if vall: | |
| val = vall[0] | |
| elif bound.replace('.', '1').isdigit(): | |
| val = float(bound) | |
| else: | |
| print("Wrong bound: {0}".format(bound)) | |
| else: | |
| print("Wrong bound: {0}".format(bound)) | |
| return val | |
| ubv = get_bound_value(ub, -1, ridx) | |
| lbv = get_bound_value(lb, float('inf'), ridx) | |
| t = get_bound_value(error, self.tolerance, ridx) | |
| # denormalize error threshold | |
| denormerr = t * ubv / 100 if ubv != 100 and ubv > 0 else t | |
| return lbv, ubv, denormerr | |
| def get_value(self, name: str, ridx: int = 0) -> list: | |
| """ | |
| Get value of the metric from self.results. | |
| If result of this metric is not provided, the metric name will be added into self.ignoremetics. | |
| All future test(s) on this metric will fail. | |
| @param name: name of the metric | |
| @returns: list with value found in self.results; list is empty when value is not found. | |
| """ | |
| results = [] | |
| data = self.results[ridx] if ridx in self.results else self.results[0] | |
| if name not in self.ignoremetrics: | |
| if name in data: | |
| results.append(data[name]) | |
| elif name.replace('.', '1').isdigit(): | |
| results.append(float(name)) | |
| else: | |
| self.ignoremetrics.add(name) | |
| return results | |
| def check_bound(self, val, lb, ub, err): | |
| return True if val <= ub + err and val >= lb - err else False | |
| # Positive Value Sanity check | |
| def pos_val_test(self): | |
| """ | |
| Check if metrics value are non-negative. | |
| One metric is counted as one test. | |
| Failure: when metric value is negative or not provided. | |
| Metrics with negative value will be added into self.ignoremetrics. | |
| """ | |
| negmetric = dict() | |
| pcnt = 0 | |
| tcnt = 0 | |
| rerun = list() | |
| results = self.get_results() | |
| if not results: | |
| return | |
| for name, val in results.items(): | |
| if val < 0: | |
| negmetric[name] = val | |
| rerun.append(name) | |
| else: | |
| pcnt += 1 | |
| tcnt += 1 | |
| # The first round collect_perf() run these metrics with simple workload | |
| # "true". We give metrics a second chance with a longer workload if less | |
| # than 20 metrics failed positive test. | |
| if len(rerun) > 0 and len(rerun) < 20: | |
| second_results = dict() | |
| self.second_test(rerun, second_results) | |
| for name, val in second_results.items(): | |
| if name not in negmetric: | |
| continue | |
| if val >= 0: | |
| del negmetric[name] | |
| pcnt += 1 | |
| if len(negmetric.keys()): | |
| self.ignoremetrics.update(negmetric.keys()) | |
| self.errlist.extend( | |
| [TestError([m], self.workloads[self.wlidx], negmetric[m], 0) for m in negmetric.keys()]) | |
| return | |
| def evaluate_formula(self, formula: str, alias: dict, ridx: int = 0): | |
| """ | |
| Evaluate the value of formula. | |
| @param formula: the formula to be evaluated | |
| @param alias: the dict has alias to metric name mapping | |
| @returns: value of the formula is success; -1 if the one or more metric value not provided | |
| """ | |
| stack = [] | |
| b = 0 | |
| errs = [] | |
| sign = "+" | |
| f = str() | |
| # TODO: support parenthesis? | |
| for i in range(len(formula)): | |
| if i+1 == len(formula) or formula[i] in ('+', '-', '*', '/'): | |
| s = alias[formula[b:i]] if i + \ | |
| 1 < len(formula) else alias[formula[b:]] | |
| v = self.get_value(s, ridx) | |
| if not v: | |
| errs.append(s) | |
| else: | |
| f = f + "{0}(={1:.4f})".format(s, v[0]) | |
| if sign == "*": | |
| stack[-1] = stack[-1] * v | |
| elif sign == "/": | |
| stack[-1] = stack[-1] / v | |
| elif sign == '-': | |
| stack.append(-v[0]) | |
| else: | |
| stack.append(v[0]) | |
| if i + 1 < len(formula): | |
| sign = formula[i] | |
| f += sign | |
| b = i + 1 | |
| if len(errs) > 0: | |
| return -1, "Metric value missing: "+','.join(errs) | |
| val = sum(stack) | |
| return val, f | |
| # Relationships Tests | |
| def relationship_test(self, rule: dict): | |
| """ | |
| Validate if the metrics follow the required relationship in the rule. | |
| eg. lower_bound <= eval(formula)<= upper_bound | |
| One rule is counted as ont test. | |
| Failure: when one or more metric result(s) not provided, or when formula evaluated outside of upper/lower bounds. | |
| @param rule: dict with metric name(+alias), formula, and required upper and lower bounds. | |
| """ | |
| alias = dict() | |
| for m in rule['Metrics']: | |
| alias[m['Alias']] = m['Name'] | |
| lbv, ubv, t = self.get_bounds( | |
| rule['RangeLower'], rule['RangeUpper'], rule['ErrorThreshold'], alias, ridx=rule['RuleIndex']) | |
| val, f = self.evaluate_formula( | |
| rule['Formula'], alias, ridx=rule['RuleIndex']) | |
| lb = rule['RangeLower'] | |
| ub = rule['RangeUpper'] | |
| if isinstance(lb, str): | |
| if lb in alias: | |
| lb = alias[lb] | |
| if isinstance(ub, str): | |
| if ub in alias: | |
| ub = alias[ub] | |
| if val == -1: | |
| self.errlist.append(TestError([m['Name'] for m in rule['Metrics']], self.workloads[self.wlidx], [], | |
| lb, ub, rule['Description'])) | |
| elif not self.check_bound(val, lbv, ubv, t): | |
| self.errlist.append(TestError([m['Name'] for m in rule['Metrics']], self.workloads[self.wlidx], [val], | |
| lb, ub, rule['Description'])) | |
| else: | |
| self.passedcnt += 1 | |
| self.totalcnt += 1 | |
| return | |
| # Single Metric Test | |
| def single_test(self, rule: dict): | |
| """ | |
| Validate if the metrics are in the required value range. | |
| eg. lower_bound <= metrics_value <= upper_bound | |
| One metric is counted as one test in this type of test. | |
| One rule may include one or more metrics. | |
| Failure: when the metric value not provided or the value is outside the bounds. | |
| This test updates self.total_cnt. | |
| @param rule: dict with metrics to validate and the value range requirement | |
| """ | |
| lbv, ubv, t = self.get_bounds( | |
| rule['RangeLower'], rule['RangeUpper'], rule['ErrorThreshold']) | |
| metrics = rule['Metrics'] | |
| passcnt = 0 | |
| totalcnt = 0 | |
| failures = dict() | |
| rerun = list() | |
| for m in metrics: | |
| totalcnt += 1 | |
| result = self.get_value(m['Name']) | |
| if len(result) > 0 and self.check_bound(result[0], lbv, ubv, t) or m['Name'] in self.skiplist: | |
| passcnt += 1 | |
| else: | |
| failures[m['Name']] = result | |
| rerun.append(m['Name']) | |
| if len(rerun) > 0 and len(rerun) < 20: | |
| second_results = dict() | |
| self.second_test(rerun, second_results) | |
| for name, val in second_results.items(): | |
| if name not in failures: | |
| continue | |
| if self.check_bound(val, lbv, ubv, t): | |
| passcnt += 1 | |
| del failures[name] | |
| else: | |
| failures[name] = [val] | |
| self.results[0][name] = val | |
| self.totalcnt += totalcnt | |
| self.passedcnt += passcnt | |
| if len(failures.keys()) != 0: | |
| self.errlist.extend([TestError([name], self.workloads[self.wlidx], val, | |
| rule['RangeLower'], rule['RangeUpper']) for name, val in failures.items()]) | |
| return | |
| def create_report(self): | |
| """ | |
| Create final report and write into a JSON file. | |
| """ | |
| print(self.errlist) | |
| if self.debug: | |
| allres = [{"Workload": self.workloads[i], "Results": self.allresults[i]} | |
| for i in range(0, len(self.workloads))] | |
| self.json_dump(allres, self.datafname) | |
| def check_rule(self, testtype, metric_list): | |
| """ | |
| Check if the rule uses metric(s) that not exist in current platform. | |
| @param metric_list: list of metrics from the rule. | |
| @return: False when find one metric out in Metric file. (This rule should not skipped.) | |
| True when all metrics used in the rule are found in Metric file. | |
| """ | |
| if testtype == "RelationshipTest": | |
| for m in metric_list: | |
| if m['Name'] not in self.metrics: | |
| return False | |
| return True | |
| # Start of Collector and Converter | |
| def convert(self, data: list, metricvalues: dict): | |
| """ | |
| Convert collected metric data from the -j output to dict of {metric_name:value}. | |
| """ | |
| for json_string in data: | |
| try: | |
| result = json.loads(json_string) | |
| if "metric-unit" in result and result["metric-unit"] != "(null)" and result["metric-unit"] != "": | |
| name = result["metric-unit"].split(" ")[1] if len(result["metric-unit"].split(" ")) > 1 \ | |
| else result["metric-unit"] | |
| metricvalues[name.lower()] = float(result["metric-value"]) | |
| except ValueError as error: | |
| continue | |
| return | |
| def _run_perf(self, metric, workload: str): | |
| tool = 'perf' | |
| command = [tool, 'stat', '--cputype', self.cputype, '-j', '-M', f"{metric}", "-a"] | |
| wl = workload.split() | |
| command.extend(wl) | |
| print(" ".join(command)) | |
| cmd = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8') | |
| lines = cmd.stderr.splitlines() + cmd.stdout.splitlines() | |
| data = [] | |
| for line in lines: | |
| line = line.strip() | |
| if line.startswith('{') and line.endswith('}'): | |
| data.append(line) | |
| return data | |
| def collect_perf(self, workload: str): | |
| """ | |
| Collect metric data with "perf stat -M" on given workload with -a and -j. | |
| """ | |
| self.results = dict() | |
| print(f"Starting perf collection") | |
| print(f"Long workload: {workload}") | |
| collectlist = dict() | |
| if self.collectlist != "": | |
| collectlist[0] = {x for x in self.collectlist.split(",")} | |
| else: | |
| collectlist[0] = set(list(self.metrics)) | |
| # Create metric set for relationship rules | |
| for rule in self.rules: | |
| if rule["TestType"] == "RelationshipTest": | |
| metrics = [m["Name"] for m in rule["Metrics"]] | |
| if not any(m not in collectlist[0] for m in metrics): | |
| collectlist[rule["RuleIndex"]] = [ | |
| ",".join(list(set(metrics)))] | |
| for idx, metrics in collectlist.items(): | |
| if idx == 0: | |
| wl = "true" | |
| else: | |
| wl = workload | |
| for metric in metrics: | |
| data = self._run_perf(metric, wl) | |
| if idx not in self.results: | |
| self.results[idx] = dict() | |
| self.convert(data, self.results[idx]) | |
| return | |
| def second_test(self, collectlist, second_results): | |
| workload = self.workloads[self.wlidx] | |
| for metric in collectlist: | |
| data = self._run_perf(metric, workload) | |
| self.convert(data, second_results) | |
| # End of Collector and Converter | |
| # Start of Rule Generator | |
| def parse_perf_metrics(self): | |
| """ | |
| Read and parse perf metric file: | |
| 1) find metrics with '1%' or '100%' as ScaleUnit for Percent check | |
| 2) create metric name list | |
| """ | |
| command = ['perf', 'list', '-j', '--details', 'metrics'] | |
| cmd = subprocess.run(command, stdout=subprocess.PIPE, | |
| stderr=subprocess.PIPE, encoding='utf-8') | |
| try: | |
| data = json.loads(cmd.stdout) | |
| for m in data: | |
| if 'MetricName' not in m: | |
| print("Warning: no metric name") | |
| continue | |
| if 'Unit' in m and m['Unit'] != self.cputype: | |
| continue | |
| name = m['MetricName'].lower() | |
| self.metrics.add(name) | |
| if 'ScaleUnit' in m and (m['ScaleUnit'] == '1%' or m['ScaleUnit'] == '100%'): | |
| self.pctgmetrics.add(name.lower()) | |
| except ValueError as error: | |
| print(f"Error when parsing metric data") | |
| sys.exit() | |
| return | |
| def remove_unsupported_rules(self, rules): | |
| new_rules = [] | |
| for rule in rules: | |
| add_rule = True | |
| for m in rule["Metrics"]: | |
| if m["Name"] in self.skiplist or m["Name"] not in self.metrics: | |
| add_rule = False | |
| break | |
| if add_rule: | |
| new_rules.append(rule) | |
| return new_rules | |
| def create_rules(self): | |
| """ | |
| Create full rules which includes: | |
| 1) All the rules from the "relationshi_rules" file | |
| 2) SingleMetric rule for all the 'percent' metrics | |
| Reindex all the rules to avoid repeated RuleIndex | |
| """ | |
| data = self.read_json(self.rulefname) | |
| rules = data['RelationshipRules'] | |
| self.skiplist = set([name.lower() for name in data['SkipList']]) | |
| self.rules = self.remove_unsupported_rules(rules) | |
| pctgrule = {'RuleIndex': 0, | |
| 'TestType': 'SingleMetricTest', | |
| 'RangeLower': '0', | |
| 'RangeUpper': '100', | |
| 'ErrorThreshold': self.tolerance, | |
| 'Description': 'Metrics in percent unit have value with in [0, 100]', | |
| 'Metrics': [{'Name': m.lower()} for m in self.pctgmetrics]} | |
| self.rules.append(pctgrule) | |
| # Re-index all rules to avoid repeated RuleIndex | |
| idx = 1 | |
| for r in self.rules: | |
| r['RuleIndex'] = idx | |
| idx += 1 | |
| if self.debug: | |
| # TODO: need to test and generate file name correctly | |
| data = {'RelationshipRules': self.rules, 'SupportedMetrics': [ | |
| {"MetricName": name} for name in self.metrics]} | |
| self.json_dump(data, self.fullrulefname) | |
| return | |
| # End of Rule Generator | |
| def _storewldata(self, key): | |
| ''' | |
| Store all the data of one workload into the corresponding data structure for all workloads. | |
| @param key: key to the dictionaries (index of self.workloads). | |
| ''' | |
| self.allresults[key] = self.results | |
| self.alltotalcnt[key] = self.totalcnt | |
| self.allpassedcnt[key] = self.passedcnt | |
| # Initialize data structures before data validation of each workload | |
| def _init_data(self): | |
| testtypes = ['PositiveValueTest', | |
| 'RelationshipTest', 'SingleMetricTest'] | |
| self.results = dict() | |
| self.ignoremetrics = set() | |
| self.errlist = list() | |
| self.totalcnt = 0 | |
| self.passedcnt = 0 | |
| def test(self): | |
| ''' | |
| The real entry point of the test framework. | |
| This function loads the validation rule JSON file and Standard Metric file to create rules for | |
| testing and namemap dictionaries. | |
| It also reads in result JSON file for testing. | |
| In the test process, it passes through each rule and launch correct test function bases on the | |
| 'TestType' field of the rule. | |
| The final report is written into a JSON file. | |
| ''' | |
| if not self.collectlist: | |
| self.parse_perf_metrics() | |
| if not self.metrics: | |
| print("No metric found for testing") | |
| return 0 | |
| self.create_rules() | |
| for i in range(0, len(self.workloads)): | |
| self.wlidx = i | |
| self._init_data() | |
| self.collect_perf(self.workloads[i]) | |
| # Run positive value test | |
| self.pos_val_test() | |
| for r in self.rules: | |
| # skip rules that uses metrics not exist in this platform | |
| testtype = r['TestType'] | |
| if not self.check_rule(testtype, r['Metrics']): | |
| continue | |
| if testtype == 'RelationshipTest': | |
| self.relationship_test(r) | |
| elif testtype == 'SingleMetricTest': | |
| self.single_test(r) | |
| else: | |
| print("Unsupported Test Type: ", testtype) | |
| print("Workload: ", self.workloads[i]) | |
| print("Total Test Count: ", self.totalcnt) | |
| print("Passed Test Count: ", self.passedcnt) | |
| self._storewldata(i) | |
| self.create_report() | |
| return len(self.errlist) > 0 | |
| # End of Class Validator | |
| def main() -> None: | |
| parser = argparse.ArgumentParser( | |
| description="Launch metric value validation") | |
| parser.add_argument( | |
| "-rule", help="Base validation rule file", required=True) | |
| parser.add_argument( | |
| "-output_dir", help="Path for validator output file, report file", required=True) | |
| parser.add_argument("-debug", help="Debug run, save intermediate data to files", | |
| action="store_true", default=False) | |
| parser.add_argument( | |
| "-wl", help="Workload to run while data collection", default="true") | |
| parser.add_argument("-m", help="Metric list to validate", default="") | |
| parser.add_argument("-cputype", help="Only test metrics for the given CPU/PMU type", | |
| default="cpu") | |
| args = parser.parse_args() | |
| outpath = Path(args.output_dir) | |
| reportf = Path.joinpath(outpath, 'perf_report.json') | |
| fullrule = Path.joinpath(outpath, 'full_rule.json') | |
| datafile = Path.joinpath(outpath, 'perf_data.json') | |
| validator = Validator(args.rule, reportf, debug=args.debug, | |
| datafname=datafile, fullrulefname=fullrule, workload=args.wl, | |
| metrics=args.m, cputype=args.cputype) | |
| ret = validator.test() | |
| return ret | |
| if __name__ == "__main__": | |
| import sys | |
| sys.exit(main()) |