canonical-ubuntu-qa team mailing list archive

[Tim Andersson <mp+461146@xxxxxxxxxxxxxxxxxx>]

Tim Andersson has proposed merging ~andersson123/autopkgtest-cloud:d-a-r-make-me-faster into autopkgtest-cloud:master.

Requested reviews:
  Canonical's Ubuntu QA (canonical-ubuntu-qa)

For more details, see:
https://code.launchpad.net/~andersson123/autopkgtest-cloud/+git/autopkgtest-cloud/+merge/461146
-- 
Your team Canonical's Ubuntu QA is requested to review the proposed merge of ~andersson123/autopkgtest-cloud:d-a-r-make-me-faster into autopkgtest-cloud:master.

diff --git a/charms/focal/autopkgtest-web/units/sqlite-writer.service b/charms/focal/autopkgtest-web/units/sqlite-writer.service
new file mode 100644
index 0000000..3a47c08
--- /dev/null
+++ b/charms/focal/autopkgtest-web/units/sqlite-writer.service
@@ -0,0 +1,13 @@
+[Unit]
+Description=Write test results to db
+StartLimitIntervalSec=60s
+StartLimitBurst=60
+
+[Service]
+User=ubuntu
+ExecStart=/home/ubuntu/webcontrol/sqlite-writer
+Restart=on-failure
+RestartSec=1s
+
+[Install]
+WantedBy=autopkgtest-web.target
diff --git a/charms/focal/autopkgtest-web/webcontrol/download-all-results b/charms/focal/autopkgtest-web/webcontrol/download-all-results
index 1af7918..b3e72dd 100755
--- a/charms/focal/autopkgtest-web/webcontrol/download-all-results
+++ b/charms/focal/autopkgtest-web/webcontrol/download-all-results
@@ -16,7 +16,6 @@ import io
 import json
 import logging
 import os
-import random
 import sqlite3
 import sys
 import tarfile
@@ -24,13 +23,33 @@ import time
 import urllib.parse
 from urllib.request import urlopen
 
+import amqplib.client_0_8 as amqp
 from distro_info import UbuntuDistroInfo
 from helpers.utils import get_test_id, init_db
 
 LOGGER = logging.getLogger(__name__)
+WRITER_EXCHANGE_NAME = "sqlite-write-me.fanout"
 
 config = None
 db_con = None
+amqp_con = None
+
+
+def amqp_connect():
+    """Connect to AMQP server"""
+
+    cp = configparser.ConfigParser()
+    cp.read(os.path.expanduser("~ubuntu/autopkgtest-cloud.conf"))
+    amqp_uri = cp["amqp"]["uri"]
+    parts = urllib.parse.urlsplit(amqp_uri, allow_fragments=False)
+    amqp_con = amqp.Connection(
+        parts.hostname, userid=parts.username, password=parts.password
+    )
+    logging.info(
+        "Connected to AMQP server at %s@%s" % (parts.username, parts.hostname)
+    )
+
+    return amqp_con
 
 
 def list_remote_container(container_url):
@@ -95,6 +114,8 @@ def fetch_one_result(url):
     """Download one result URL from swift and add it to the DB"""
     (release, arch, _, src, run_id, _) = url.split("/")[-6:]
     test_id = get_test_id(db_con, release, arch, src)
+    # modify this to use swiftclient instead of urllib
+    # look at update-github-jobs for help
 
     try:
         f = urlopen(url, timeout=30)
@@ -182,42 +203,30 @@ def fetch_one_result(url):
         if env in testinfo.keys():
             env_vars.append(spec)
 
-    while True:
-        try:
-            with (
-                db_con
-            ):  # this starts a transaction, making sure we release the lock at the end
-                c = db_con.cursor()
-                c.execute(
-                    "INSERT INTO result VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
-                    (
-                        test_id,
-                        run_id,
-                        ver,
-                        test_triggers,
-                        duration,
-                        exitcode,
-                        requester,
-                        ",".join(env_vars),
-                        test_uuid,
-                    ),
-                )
-                db_con.commit()
-            break
-        except sqlite3.OperationalError as e:
-            if "database is locked" in str(e):
-                sleep_time = random.uniform(0.1, 2)
-                LOGGER.info(
-                    "database is currently locked, waiting %f seconds and trying again..."
-                    % sleep_time
-                )
-                time.sleep(sleep_time)
-            else:
-                logging.info("insert operation failed with: %s" % str(e))
-                break
-        except sqlite3.IntegrityError:
-            LOGGER.info("%s was already recorded - skipping", run_id)
-            break
+    # Insert the write request into the queue
+    complete_amqp = amqp_con.channel()
+    complete_amqp.access_request(
+        "/complete", active=True, read=False, write=True
+    )
+    complete_amqp.exchange_declare(
+        WRITER_EXCHANGE_NAME, "fanout", durable=True, auto_delete=False
+    )
+    write_me_msg = {
+        "test_id": test_id,
+        "run_id": run_id,
+        "version": ver,
+        "triggers": test_triggers,
+        "duration": duration,
+        "exitcode": exitcode,
+        "requester": requester,
+        "env": ",".join(env_vars),
+        "uuid": test_uuid,
+    }
+    complete_amqp.basic_publish(
+        amqp.Message(json.dumps(write_me_msg), delivery_mode=2),
+        WRITER_EXCHANGE_NAME,
+        "",
+    )
 
 
 def fetch_container(release, container_url):
@@ -260,6 +269,7 @@ if __name__ == "__main__":
 
     config = configparser.ConfigParser()
     config.read(os.path.expanduser("~ubuntu/autopkgtest-cloud.conf"))
+    amqp_con = amqp_connect()
 
     try:
         for release in releases:
diff --git a/charms/focal/autopkgtest-web/webcontrol/download-results b/charms/focal/autopkgtest-web/webcontrol/download-results
index e71d4a0..4b9b11e 100755
--- a/charms/focal/autopkgtest-web/webcontrol/download-results
+++ b/charms/focal/autopkgtest-web/webcontrol/download-results
@@ -4,16 +4,15 @@ import configparser
 import json
 import logging
 import os
-import random
 import socket
 import sqlite3
-import time
 import urllib.parse
 
 import amqplib.client_0_8 as amqp
 from helpers.utils import get_test_id, init_db
 
 EXCHANGE_NAME = "testcomplete.fanout"
+WRITER_EXCHANGE_NAME = "sqlite-write-me.fanout"
 
 
 def amqp_connect():
@@ -83,43 +82,30 @@ def process_message(msg, db_con):
         return
 
     test_id = get_test_id(db_con, release, arch, package)
-
-    while True:
-        try:
-            with (
-                db_con
-            ):  # this starts a transaction, making sure we release the lock at the end
-                c = db_con.cursor()
-                c.execute(
-                    "INSERT INTO result VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
-                    (
-                        test_id,
-                        run_id,
-                        version,
-                        triggers,
-                        duration,
-                        exitcode,
-                        requester,
-                        info.get("env", ""),
-                        test_uuid,
-                    ),
-                )
-                db_con.commit()
-            break
-        except sqlite3.OperationalError as e:
-            if "database is locked" in str(e):
-                sleep_time = random.uniform(0.1, 2)
-                logging.info(
-                    "database is currently locked, waiting %f seconds and trying again..."
-                    % sleep_time
-                )
-                time.sleep(sleep_time)
-            else:
-                logging.info("insert operation failed with: %s" % str(e))
-                break
-        except sqlite3.IntegrityError:
-            logging.info("...which was already recorded - skipping")
-            break
+    # add to queue instead of writing to db
+    complete_amqp = amqp_con.channel()
+    complete_amqp.access_request(
+        "/complete", active=True, read=False, write=True
+    )
+    complete_amqp.exchange_declare(
+        WRITER_EXCHANGE_NAME, "fanout", durable=True, auto_delete=False
+    )
+    write_me_msg = {
+        "test_id": test_id,
+        "run_id": run_id,
+        "version": version,
+        "triggers": triggers,
+        "duration": duration,
+        "exitcode": exitcode,
+        "requester": requester,
+        "env": info.get("env", ""),
+        "uuid": test_uuid,
+    }
+    complete_amqp.basic_publish(
+        amqp.Message(json.dumps(write_me_msg), delivery_mode=2),
+        WRITER_EXCHANGE_NAME,
+        "",
+    )
 
     msg.channel.basic_ack(msg.delivery_tag)
 
diff --git a/charms/focal/autopkgtest-web/webcontrol/sqlite-writer b/charms/focal/autopkgtest-web/webcontrol/sqlite-writer
new file mode 100755
index 0000000..4ba8bdf
--- /dev/null
+++ b/charms/focal/autopkgtest-web/webcontrol/sqlite-writer
@@ -0,0 +1,138 @@
+#!/usr/bin/python3
+
+import configparser
+import json
+import logging
+import os
+import socket
+import sqlite3
+import urllib.parse
+
+import amqplib.client_0_8 as amqp
+from helpers.utils import init_db
+
+EXCHANGE_NAME = "sqlite-write-me.fanout"
+
+config = None
+db_con = None
+
+INSERT_INTO_KEYS = [
+    "test_id",
+    "run_id",
+    "version",
+    "triggers",
+    "duration",
+    "exitcode",
+    "requester",
+    "env",
+    "uuid",
+]
+
+
+def amqp_connect():
+    """Connect to AMQP server"""
+
+    cp = configparser.ConfigParser()
+    cp.read(os.path.expanduser("~ubuntu/autopkgtest-cloud.conf"))
+    amqp_uri = cp["amqp"]["uri"]
+    parts = urllib.parse.urlsplit(amqp_uri, allow_fragments=False)
+    amqp_con = amqp.Connection(
+        parts.hostname, userid=parts.username, password=parts.password
+    )
+    logging.info(
+        "Connected to AMQP server at %s@%s" % (parts.username, parts.hostname)
+    )
+
+    return amqp_con
+
+
+def db_connect():
+    """Connect to SQLite DB"""
+    cp = configparser.ConfigParser()
+    cp.read(os.path.expanduser("~ubuntu/autopkgtest-cloud.conf"))
+
+    db_con = init_db(cp["web"]["database"])
+
+    return db_con
+
+
+def check_msg(queue_msg):
+    required_keys = set(
+        [
+            "test_id",
+            "run_id",
+            "version",
+            "triggers",
+            "duration",
+            "exitcode",
+            "requester",
+            "env",
+            "uuid",
+        ]
+    )
+    queue_keys = set(queue_msg.keys())
+    if required_keys == queue_keys:
+        return True
+    return False
+
+
+def process_message(msg, db_con):
+    # We want to ack and re-send messages if insert fails?
+    body = msg.body
+    if isinstance(body, bytes):
+        body = body.decode("UTF-8", errors="replace")
+    info = json.loads(body)
+    logging.info("Message is: \n%s" % json.dumps(info, indent=2))
+    if not check_msg(info):
+        logging.error(
+            "Message has incorrect keys! Ignoring\n%s"
+            % json.dumps(info, indent=2)
+        )
+        msg.channel.basic_ack(msg.delivery_tag)
+        return
+    # insert into db
+    sqlite3.paramstyle = "named"
+    with db_con:
+        c = db_con.cursor()
+        # change this to column names
+        c.execute(
+            (
+                "INSERT INTO result(test_id, run_id, version, triggers, duration, "
+                "exitcode, requester, env, uuid) VALUES (:test_id, :run_id, "
+                ":version, :triggers, :duration, :exitcode, :requester, :env, :uuid)"
+            ),
+            {
+                "test_id": info["test_id"],
+                "run_id": info["run_id"],
+                "version": info["version"],
+                "triggers": info["triggers"],
+                "duration": info["duration"],
+                "exitcode": info["exitcode"],
+                "requester": info["requester"],
+                "env": info["env"],
+                "uuid": info["uuid"],
+            },
+        )
+    logging.info("Inserted the following entry into the db:\n%s" % body)
+
+    msg.channel.basic_ack(msg.delivery_tag)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.INFO)
+    db_con = db_connect()
+    amqp_con = amqp_connect()
+    status_ch = amqp_con.channel()
+    status_ch.access_request("/complete", active=True, read=True, write=False)
+    status_ch.exchange_declare(
+        EXCHANGE_NAME, "fanout", durable=True, auto_delete=False
+    )
+    queue_name = "sqlite-writer-listener-%s" % socket.getfqdn()
+    status_ch.queue_declare(queue_name, durable=True, auto_delete=False)
+    status_ch.queue_bind(queue_name, EXCHANGE_NAME, queue_name)
+    logging.info("Listening to requests on %s" % queue_name)
+    status_ch.basic_consume(
+        "", callback=lambda msg: process_message(msg, db_con)
+    )
+    while status_ch.callbacks:
+        status_ch.wait()