canonical-ubuntu-qa team mailing list archive

[Tim Andersson <mp+460847@xxxxxxxxxxxxxxxxxx>]

Tim Andersson has proposed merging ~andersson123/autopkgtest-cloud:d-r-d-a-r-merging 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/460847
-- 
Your team Canonical's Ubuntu QA is requested to review the proposed merge of ~andersson123/autopkgtest-cloud:d-r-d-a-r-merging into autopkgtest-cloud:master.

diff --git a/charms/focal/autopkgtest-web/webcontrol/download-all-results b/charms/focal/autopkgtest-web/webcontrol/download-all-results
index 1af7918..3fc1eb7 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):
@@ -182,42 +201,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 +267,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..3939bde
--- /dev/null
+++ b/charms/focal/autopkgtest-web/webcontrol/sqlite-writer
@@ -0,0 +1,136 @@
+#!/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 = [
+        "test_id",
+        "run_id",
+        "version",
+        "triggers",
+        "duration",
+        "exitcode",
+        "requester",
+        "env",
+        "uuid",
+    ]
+    queue_msg_keys = list(queue_msg.keys())
+    required_keys.sort()
+    queue_msg_keys.sort()
+    if queue_msg_keys == required_keys:
+        return True
+    return False
+
+
+def process_message(msg, db_con):
+    # aight, time to test this with download-results and download-all-results now.
+    body = msg.body
+    if isinstance(body, bytes):
+        body = body.decide("UTF-8", errors="replace")
+    info = json.loads(body)
+    logging.info("Message is: \n%s" % json.dumps(info, indent=2))
+    print(check_msg(info))
+    if not check_msg(info):
+        logging.info(
+            "Message has incorrect keys!\n%s" % json.dumps(info, indent=2)
+        )
+    # insert into db
+    try:
+        with db_con:
+            c = db_con.cursor()
+            c.execute(
+                "INSERT INTO result VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    info["test_id"],
+                    info["run_id"],
+                    info["version"],
+                    info["triggers"],
+                    info["duration"],
+                    info["exitcode"],
+                    info["requester"],
+                    info["env"],
+                    info["uuid"],
+                ),
+            )
+            db_con.commit()
+        logging.info("Inserted the following entry into the db:\n%s" % body)
+    except sqlite3.OperationalError as e:
+        logging.info("Insert operation failed with: %s" % str(e))
+    except sqlite3.IntegrityError as e:
+        logging.info("Insert operation failed with: %s" % str(e))
+
+    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()