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()
--
Mailing list: https://launchpad.net/~canonical-ubuntu-qa
Post to : canonical-ubuntu-qa@lists.launchpad.net
Unsubscribe : https://launchpad.net/~canonical-ubuntu-qa
More help : https://help.launchpad.net/ListHelp
