contest: fetcher: compute test stability

As the first step for HW test ingest we need to auto-judge if
given test is stable and passing or not supported by HW.

Create a DB of all test cases we have seen and compute pass/fail
statistics. We will judge a test as stable / passing if it had
15 clean runs. "undoing" the passing status will be manual
for now.

Signed-off-by: Jakub Kicinski <[email protected]>

Author: kuba-moo

contest/results-fetcher.py

@@ -26,10 +26,36 @@
 combined=name-of-manifest.json
 [db]
 db=db-name
+stability-name=table-name
 results-name=table-name
 branches-name=table-name
 """
 
+def result_flatten(full):
+    """
+    Take in a full result dict (for one run, with subtests).
+    Return a list of dicts:
+        [
+            { "group": str, "test": str, "subtest": str/None, "result": bool },
+        ]
+    """
+    flat = []
+
+    for test in full["results"]:
+        l1 = { "group":   test["group"],
+               "test":    test["test"],
+               "subtest": None,
+               "result":  test["result"].lower() == "pass"
+        }
+        flat.append(l1)
+        for case in test.get("results", []):
+            data = l1.copy()
+            data["subtest"] = case["test"]
+            data["result"] = case["result"].lower() == "pass"
+            flat.append(data)
+
+    return flat
+
 
 class FetcherState:
     def __init__(self):
@@ -39,6 +65,7 @@ def __init__(self):
         # "fetched" is more of a "need state rebuild"
         self.fetched = True
 
+        self.tbl_stb = self.config.get("db", "stability-name", fallback="stability")
         self.tbl_res = self.config.get("db", "results-name", fallback="results")
         self.tbl_brn = self.config.get("db", "branches-name", fallback="branches")
 
@@ -104,6 +131,70 @@ def psql_json_split(self, data):
             full = json.dumps(data)
         return json.dumps(normal), full
 
+    def psql_stability_selector(self, cur, data, row):
+        base = cur.mogrify("WHERE remote = %s AND executor = %s AND grp = %s AND test = %s",
+                           (data["remote"], data["executor"], row["group"], row["test"],)).decode('utf-8')
+
+        if row["subtest"] is None:
+            return base + " AND subtest is NULL"
+        return base + cur.mogrify(" AND subtest = %s", (row["subtest"],)).decode('utf-8')
+
+    def psql_get_stability(self, data, row):
+        with self.psql_conn.cursor() as cur:
+            cur.execute(f"SELECT pass_cnt, fail_cnt, pass_srk, fail_srk, pass_cur, fail_cur, passing FROM {self.tbl_stb} " +
+                        self.psql_stability_selector(cur, data, row))
+            rows = cur.fetchall()
+        if rows and len(rows) > 0:
+            res = rows[0]
+        else:
+            res = [0] * 10
+            res[6] = None # passing
+        return {
+            "pass_cnt": res[0],
+            "fail_cnt": res[1],
+            "pass_srk": res[2],
+            "fail_srk": res[3],
+            "pass_cur": res[4],
+            "fail_cur": res[5],
+            "passing": res[6],
+            "exists": bool(rows),
+        }
+
+    def psql_insert_stability(self, data):
+        flat = result_flatten(data)
+
+        for row in flat:
+            # Fetch current state
+            stability = self.psql_get_stability(data, row)
+            if not stability["exists"]:
+                with self.psql_conn.cursor() as cur:
+                    cur.execute(f"INSERT INTO {self.tbl_stb} (remote, executor, grp, test, subtest) " +
+                                cur.mogrify("VALUES (%s, %s, %s, %s, %s)",
+                                            (data["remote"], data["executor"], row["group"], row["test"], row["subtest"],)).decode('utf-8'))
+            # Update state
+            if row["result"]:
+                key_pfx = "pass"
+                stability["fail_cur"] = 0
+            else:
+                key_pfx = "fail"
+                stability["pass_cur"] = 0
+
+            stability[key_pfx + "_cnt"] += 1
+            stability[key_pfx + "_cur"] += 1
+            stability[key_pfx + "_srk"] = max(stability[key_pfx + "_cur"], stability[key_pfx + "_srk"])
+
+            now = datetime.datetime.now().isoformat() + "+00:00"
+            if stability[key_pfx + "_srk"] > 15 and not stability["passing"]: # 5 clean days for HW
+                print("Test reached stability", row["remote"], row["test"], row["subtest"])
+                stability["passing"] = now
+
+            with self.psql_conn.cursor() as cur:
+                cur.execute(f"UPDATE {self.tbl_stb} SET " +
+                            cur.mogrify("pass_cnt = %s, fail_cnt = %s, pass_srk = %s, fail_srk = %s, pass_cur = %s, fail_cur = %s, passing = %s, last_update = %s",
+                                        (stability["pass_cnt"], stability["fail_cnt"], stability["pass_srk"], stability["fail_srk"],
+                                         stability["pass_cur"], stability["fail_cur"], stability["passing"], now)).decode('utf-8') +
+                            self.psql_stability_selector(cur, data, row))
+
     def insert_real(self, remote, run):
         data = run.copy()
         data["remote"] = remote["name"]
@@ -119,6 +210,8 @@ def insert_real(self, remote, run):
                 q = f"UPDATE {self.tbl_res} " + vals + ' ' + selector
                 cur.execute(q)
 
+        self.psql_insert_stability(data)
+
 
 def write_json_atomic(path, data):
     tmp = path + '.new'

deploy/contest/db

@@ -65,3 +65,19 @@ CREATE TABLE db_monitor (
     disk_pct            REAL,
     disk_pct_metal      REAL
 );
+
+CREATE TABLE stability (
+    remote              varchar(80),
+    executor            varchar(80),
+    grp                 varchar(80),
+    test                varchar(128),
+    subtest             varchar(256),
+    pass_cnt		integer NOT NULL DEFAULT 0,
+    fail_cnt		integer NOT NULL DEFAULT 0,
+    pass_srk		integer NOT NULL DEFAULT 0,
+    fail_srk		integer NOT NULL DEFAULT 0,
+    pass_cur		integer NOT NULL DEFAULT 0,
+    fail_cur		integer NOT NULL DEFAULT 0,
+    last_update		timestamp,
+    passing		timestamp
+);