-# Copyright (c) 2023 Cisco and/or its affiliates.
+# Copyright (c) 2024 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:
This script expects a particular tree created on a filesystem by
per_patch_perf.sh bootstrap script, including test results
exported as json files according to a current model schema.
-This script extracts the results (according to tresult type)
+This script extracts the results (according to result type)
and joins them into one list of floats for parent and one for current.
This script then uses jumpavg library to determine whether there was
this script votes -1 (by exiting with code 1), otherwise it votes +1 (exit 0).
"""
-import json
-import os
import sys
-from typing import Dict, List
-
from resources.libraries.python import jumpavg
-
-
-def parse(dirpath: str, fake_value: float) -> Dict[str, List[float]]:
- """Looks for test jsons, extract scalar results.
-
- Files other than .json are skipped, jsons without test_id are skipped.
- If the test failed, four fake values are used as a fake result.
-
- Units are ignored, as both parent and current are tested
- with the same CSIT code so the unit should be identical.
-
- :param dirpath: Path to the directory tree to examine.
- :param fail_value: Fake value to use for test cases that failed.
- :type dirpath: str
- :returns: Mapping from test IDs to list of measured values.
- :rtype: Dict[str, List[float]]
- :raises RuntimeError: On duplicate test ID or unknown test type.
- """
- results = {}
- for root, _, files in os.walk(dirpath):
- for filename in files:
- if not filename.endswith(".json"):
- continue
- filepath = os.path.join(root, filename)
- with open(filepath, "rt", encoding="utf8") as file_in:
- data = json.load(file_in)
- if "test_id" not in data:
- continue
- name = data["test_id"]
- if name in results:
- raise RuntimeError(f"Duplicate: {name}")
- if not data["passed"]:
- results[name] = [fake_value] * 4
- continue
- result_object = data["result"]
- result_type = result_object["type"]
- if result_type == "mrr":
- results[name] = result_object["receive_rate"]["rate"]["values"]
- elif result_type == "ndrpdr":
- results[name] = [result_object["pdr"]["lower"]["rate"]["value"]]
- elif result_type == "soak":
- results[name] = [
- result_object["critical_rate"]["lower"]["rate"]["value"]
- ]
- elif result_type == "reconf":
- results[name] = [result_object["loss"]["time"]["value"]]
- elif result_type == "hoststack":
- results[name] = [result_object["bandwidth"]["value"]]
- else:
- raise RuntimeError(f"Unknown result type: {result_type}")
- return results
+from resources.libraries.python.model.parse import parse
def main() -> int:
parent_results = {}
current_results = {}
parent_results = parse(f"csit_parent/{iteration}", fake_value=2.0)
- parent_names = set(parent_results.keys())
+ parent_names = list(parent_results)
if test_names is None:
test_names = parent_names
if not parent_names:
break
assert parent_names == test_names, f"{parent_names} != {test_names}"
current_results = parse(f"csit_current/{iteration}", fake_value=1.0)
- current_names = set(current_results.keys())
+ current_names = list(current_results)
assert (
current_names == parent_names
), f"{current_names} != {parent_names}"
current_aggregate[name].extend(current_results[name])
exit_code = 0
for name in test_names:
- print(f"Test name: {name}")
parent_values = parent_aggregate[name]
current_values = current_aggregate[name]
print(f"Time-ordered MRR values for parent build: {parent_values}")
# TODO: Version of classify that takes max_value and list of stats?
# That matters if only stats (not list of floats) are given.
classified_list = jumpavg.classify([parent_values, current_values])
- if len(classified_list) < 2:
- print(f"Test {name}: normal (no anomaly)")
- continue
- anomaly = classified_list[1].comment
- if anomaly == "regression":
- print(f"Test {name}: anomaly regression")
- exit_code = 3 # 1 or 2 can be caused by other errors
- continue
- print(f"Test {name}: anomaly {anomaly}")
+ anomaly_name = "normal (no anomaly)"
+ if len(classified_list) > 1:
+ anomaly = classified_list[1].comment
+ anomaly_name = "anomaly progression"
+ if anomaly == "regression":
+ anomaly_name = "anomaly regression"
+ exit_code = 3 # 1 or 2 can be caused by other errors
+ print(f"Test name {name}: {anomaly_name}")
print(f"Exit code: {exit_code}")
return exit_code
Code Review / csit.git / blobdiff