wip: SQL results

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

contest/backend/query.py

@@ -3,9 +3,13 @@
 
 
 from flask import Flask
+from flask import Response
 from flask import request
+import json
+import psycopg2
 import couchdb
 import os
+import re
 import datetime
 
 
@@ -16,6 +20,10 @@
 couch = couchdb.Server(f'http://{user}:{pwd}@127.0.0.1:5984')
 res_db = couch["results"]
 
+db_name = os.getenv('DB_NAME')
+psql = psycopg2.connect(database=db_name)
+psql.autocommit = True
+
 
 def branches_to_rows(br_cnt):
     data = res_db.view('branch/rows', None,
@@ -31,6 +39,17 @@ def hello():
     return '<h1>boo!</h1>'
 
 
+@app.route('/branches')
+def branches():
+    global psql
+
+    with psql.cursor() as cur:
+        cur.execute(f"SELECT branch, t_date, base, url FROM branches ORDER BY branch DESC LIMIT 40")
+        rows = [{"branch": r[0], "date": r[1].isoformat() + "+00:00", "base": r[2], "url": r[3]} for r in cur.fetchall()]
+        rows.reverse()
+    return rows
+
+
 @app.route('/results')
 def results():
     global couch
@@ -63,3 +82,61 @@ def results():
     print(f"Query for {br_cnt} branches, {need_rows} records took: {str(t3-t1)} ({str(t2-t1)}+{str(t3-t2)})")
 
     return data
+
+
+def branches_to_rows2(br_cnt):
+    global psql
+
+    cnt = 0
+    with psql.cursor() as cur:
+        q = f"SELECT branch,count(*) FROM results GROUP BY branch ORDER BY branch DESC LIMIT {br_cnt}"
+        cur.execute(q)
+        for r in cur.fetchall():
+            cnt += r[1]
+    return cnt
+
+
+@app.route('/results2')
+def results2():
+    global psql
+
+    br_name = request.args.get('branch-name')
+    if br_name:
+        if re.match(r'^[\w-_ ]+$', br_name) is None:
+            return {}
+
+        t1 = datetime.datetime.now()
+        with psql.cursor() as cur:
+            cur.execute(f"SELECT json_normal FROM results WHERE branch = '{br_name}' LIMIT 100")
+            rows = [json.loads(r[0]) for r in cur.fetchall()]
+        t2 = datetime.datetime.now()
+        print("Query for exact branch took: ", str(t2-t1))
+        return rows
+
+    t1 = datetime.datetime.now()
+
+    br_cnt = request.args.get('branches')
+    try:
+        br_cnt = int(br_cnt)
+    except:
+        br_cnt = None
+    if not br_cnt:
+        br_cnt = 10
+
+    raw = request.args.get('raw')
+
+    need_rows = branches_to_rows2(br_cnt)
+    t2 = datetime.datetime.now()
+    with psql.cursor() as cur:
+        cur.execute(f"SELECT json_normal FROM results ORDER BY branch DESC LIMIT {need_rows}")
+        if raw:
+            rows = "[" + ",".join([r[0] for r in cur.fetchall()]) + "]"
+        else:
+            rows = [json.loads(r[0]) for r in cur.fetchall()]
+
+    t3 = datetime.datetime.now()
+    print(f"Query for {br_cnt} branches, {need_rows} records took: {str(t3-t1)} ({str(t2-t1)}+{str(t3-t2)})")
+
+    if raw:
+        return Response(rows, mimetype='application/json')
+    return rows

contest/results-fetcher.py

@@ -2,10 +2,12 @@
 # SPDX-License-Identifier: GPL-2.0
 
 import configparser
+import copy
 import couchdb
 import datetime
 import json
 import os
+import psycopg2
 import requests
 import time
 import uuid
@@ -23,8 +25,9 @@
 url_pfx=relative/within/server
 combined=name-of-manifest.json
 [db]
-results-name=db-name
-branches-name=db-name
+db=db-name
+results-name=table-name
+branches-name=table-name
 user=name
 pwd=pass
 """
@@ -38,11 +41,18 @@ def __init__(self):
         # "fetched" is more of a "need state rebuild"
         self.fetched = True
 
+        self.tbl_res = self.config.get("db", "results-name", fallback="results")
+        self.tbl_brn = self.config.get("db", "branches-name", fallback="branches")
+
         user = self.config.get("db", "user")
         pwd = self.config.get("db", "pwd")
         server = couchdb.Server(f'http://{user}:{pwd}@127.0.0.1:5984')
-        self.res_db = server[self.config.get("db", "results-name", fallback="results")]
-        self.brn_db = server[self.config.get("db", "branches-name", fallback="branches")]
+        self.res_db = server[self.tbl_res]
+        self.brn_db = server[self.tbl_brn]
+
+        db_name = self.config.get("db", "db")
+        self.psql_conn = psycopg2.connect(database=db_name)
+        self.psql_conn.autocommit = True
 
     def _one(self, rows):
         rows = list(rows)
@@ -51,13 +61,97 @@ def _one(self, rows):
         return rows[0]
 
     def get_branch(self, name):
+        try:
+            with self.psql_conn.cursor() as cur:
+                cur.execute(f"SELECT info FROM {self.tbl_brn} WHERE branch = '{name}'")
+                rows = cur.fetchall()
+            return json.loads(rows[0][0])
+        except Exception as e:
+            print("PSQL branch GET FAIL!")
+            print(e)
+
         branch_info = self.brn_db.find({
             'selector': {
                 'branch': name
             }
         })
         return self._one(branch_info)
 
+    def psql_run_selector(self, cur, remote, run):
+        return cur.mogrify("WHERE branch = %s AND remote = %s AND executor = %s",
+                           (run['branch'], remote["name"], run["executor"],)).decode('utf-8')
+
+    def psql_has_wip(self, remote, run):
+        with self.psql_conn.cursor() as cur:
+            cur.execute(f"SELECT branch FROM {self.tbl_res} " + self.psql_run_selector(cur, remote, run))
+            rows = cur.fetchall()
+        return len(rows) > 0
+
+    def insert_result_psql(self, cur, data):
+        normal, full = self.psql_json_split(data)
+        arg = cur.mogrify("(%s,%s,%s,%s,%s,%s,%s)", (data["branch"], data["remote"], data["executor"],
+                                                     data["start"], data["end"], normal, full))
+        cur.execute(f"INSERT INTO {self.tbl_res} VALUES " + arg.decode('utf-8'))
+
+    def insert_wip_psql(self, remote, run, branch_info):
+        try:
+            if self.psql_has_wip(remote, run):
+                # no point, we have no interesting info to add
+                return
+
+            data = run.copy()
+            data["remote"] = remote["name"]
+            when = datetime.datetime.fromisoformat(branch_info['date'])
+            data["start"] = str(when)
+            when += datetime.timedelta(hours=2, minutes=58)
+            data["end"] = str(when)
+            data["results"] = None
+
+            with self.psql_conn.cursor() as cur:
+                self.insert_result_psql(cur, data)
+            print("PSQL WIP save success!")
+        except Exception as e:
+            print("PSQL WIP save FAIL!")
+            print(e)
+
+    def psql_json_split(self, data):
+        # return "normal" and "full" as json string or None
+        # "full" will be None if they are the same to save storage
+        if "results" not in data:
+            return None, None
+
+        normal = copy.deepcopy(data)
+        full = None
+
+        for row in normal["results"]:
+            if row["results"]:
+                full = True
+                del row["results"]
+
+        if full:
+            full = json.dumps(data)
+        return json.dumps(normal), full
+
+    def insert_real_psql(self, remote, run):
+        data = run.copy()
+        data["remote"] = remote["name"]
+
+        try:
+            with self.psql_conn.cursor() as cur:
+                if self.psql_has_wip(remote, run):
+                    self.insert_result_psql(cur, data)
+                else:
+                    normal, full = self.psql_json_split(data)
+                    vals = cur.mogrify("SET t_start = %s, t_end = %s, json_normal = %s, json_full = %s",
+                                       (data["start"], data["end"], normal, full)).decode('utf-8')
+                    selector = self.psql_run_selector(cur, remote, run)
+                    q = f"UPDATE {self.tbl_res} " + vals + selector
+                    cur.execute(q)
+            print("PSQL FULL save success!")
+        except Exception as e:
+            print("PSQL FULL save FAIL!")
+            print(e)
+
     def get_wip_row(self, remote, run):
         rows = self.res_db.find({
             'selector': {
@@ -71,10 +165,12 @@ def get_wip_row(self, remote, run):
             return row
 
     def insert_wip(self, remote, run):
-        existing = self.get_wip_row(remote, run)
-
         branch_info = self.get_branch(run["branch"])
 
+        self.insert_wip_psql(remote, run, branch_info)
+
+        existing = self.get_wip_row(remote, run)
+
         data = run.copy()
         if existing:
             data['_id'] = existing['_id']
@@ -91,6 +187,7 @@ def insert_wip(self, remote, run):
         self.res_db.save(data)
 
     def insert_real(self, remote, run):
+        self.insert_real_psql(remote, run)
         existing = self.get_wip_row(remote, run)
 
         data = run.copy()