Merge ~tsimonq2/autopkgtest-cloud:cache-amqp-queue into autopkgtest-cloud:master

[Simon Quigley]

[Iain Lane (laney)]

Cheers. I made a lot of changes to this but it's basically what you did. See the commits on top for the details!

review: Approve

diff --git a/webcontrol/browse.cgi b/webcontrol/browse.cgi
index 9652c0a..c180068 100755
--- a/webcontrol/browse.cgi
+++ b/webcontrol/browse.cgi
@@ -11,7 +11,6 @@ import re
 import distro_info
 from collections import OrderedDict
 
-import amqplib.client_0_8 as amqp
 from wsgiref.handlers import CGIHandler
 import flask
 
@@ -106,36 +105,6 @@ def srchash(src):
         return src[0]
 
 
-def get_queue_requests(amqp_channel, queue_name, queue_size):
-    '''Return list of pending requests in AMQP queue'''
-
-    delivery_tags = []
-    requests = []
-
-    # This is going to be too slow
-    if queue_size > 500:
-        return ['Too many requests to display']
-
-    # non-acking read of all requests to inspect the queue
-    while True:
-        r = amqp_channel.basic_get(queue_name)
-        if r is None:
-            break
-        requests.append(r.body)
-        delivery_tags.append(r.delivery_tag)
-    # now explicitly reject the requests so that they become available again
-    # for real workers
-    for t in delivery_tags:
-        amqp_channel.basic_reject(t, True)
-
-    res = []
-    for r in requests:
-        if isinstance(r, bytes):
-            r = r.decode('UTF-8')
-        res.append(r)
-    return res
-
-
 def get_release_arches():
     '''Determine available releases and architectures
 
@@ -156,39 +125,16 @@ def get_queue_info(include_requests=True):
 
     Return (releases, arches, context -> release -> arch -> (queue_size, [requests])).
     '''
-    if not amqp_uri:
-        return ([], [], {})
-
-    # connect to AMQP
-    parts = urllib.parse.urlsplit(amqp_uri, allow_fragments=False)
-    amqp_con = amqp.Connection(parts.hostname, userid=parts.username,
-                               password=parts.password)
-    ch = amqp_con.channel()
-    release_arches = get_release_arches()
-    all_arches = set()
-
-    result = {}
-    for context in AMQP_CONTEXTS:
-        for release, arches in release_arches.items():
-            for arch in arches:
-                if context == 'ubuntu':
-                    # ubuntu test requests use context-less name (backwards compatibility)
-                    queue_name = 'debci-%s-%s' % (release, arch)
-                else:
-                    queue_name = 'debci-%s-%s-%s' % (context, release, arch)
-                (_, queue_size, _) = ch.queue_declare(queue=queue_name,
-                                                      durable=True,
-                                                      exclusive=False,
-                                                      auto_delete=False)
-                if include_requests:
-                    requests = get_queue_requests(ch, queue_name, queue_size)
-                else:
-                    requests = []
-                result.setdefault(context, {}).setdefault(release, {})[arch] = (queue_size,
-                                                                                requests)
-                all_arches.add(arch)
-
-    return (release_arches.keys(), sorted(all_arches), result)
+
+    if include_requests:
+        cache = '/tmp/queued.json'
+    else:
+        cache = '/tmp/queued_no_requests.json'
+
+    with open(cache, 'r') as queued:
+        queue_info = tuple(json.load(queued))
+
+    return queue_info
 
 
 def get_source_versions(db, release):
@@ -328,7 +274,7 @@ def running():
     for c in AMQP_CONTEXTS:
         for r in releases:
             for a in arches:
-                (queue_length, queue_items) =  queue_info.get(c, {}).get(r, {}).get(a, (0, []))
+                (queue_length, queue_items) = queue_info.get(c, {}).get(r, {}).get(a, (0, []))
                 queue_lengths.setdefault(c, {}).setdefault(r, {})[a] = queue_length
 
     try:
diff --git a/webcontrol/cache-amqp b/webcontrol/cache-amqp
 new file mode 100755
index 0000000..af01bd9
--- /dev/null
+++ b/webcontrol/cache-amqp
@@ -0,0 +1,85 @@
+#!/usr/bin/python3
+
+import json
+import urllib
+import amqplib.client_0_8 as amqp
+
+amqp_uri = None
+
+
+def get_queue_requests(amqp_channel, queue_name, queue_size):
+    '''Return list of pending requests in AMQP queue'''
+
+    delivery_tags = []
+    requests = []
+
+    # This is going to be too slow
+    if queue_size > 1000:
+        return ['1000+ requests, too many to display']
+
+    # non-acking read of all requests to inspect the queue
+    while True:
+        r = amqp_channel.basic_get(queue_name)
+        if r is None:
+            break
+        requests.append(r.body)
+        delivery_tags.append(r.delivery_tag)
+    # now explicitly reject the requests so that they become available again
+    # for real workers
+    for t in delivery_tags:
+        amqp_channel.basic_reject(t, True)
+
+    res = []
+    for r in requests:
+        if isinstance(r, bytes):
+            r = r.decode('UTF-8')
+        res.append(r)
+    return res
+
+
+def query_amqp(output, include_requests=True):
+    '''Get queue contents from AMQP and cache them on disk'''
+
+    queued = ""
+
+    if not amqp_uri:
+        queued = ([], [], {})
+
+    # connect to AMQP
+    parts = urllib.parse.urlsplit(amqp_uri, allow_fragments=False)
+    amqp_con = amqp.Connection(parts.hostname, userid=parts.username,
+                               password=parts.password)
+    ch = amqp_con.channel()
+    release_arches = get_release_arches()
+    all_arches = set()
+
+    result = {}
+    for context in AMQP_CONTEXTS:
+        for release, arches in release_arches.items():
+            for arch in arches:
+                if context == 'ubuntu':
+                    # ubuntu test requests use context-less name (backwards compatibility)
+                    queue_name = 'debci-%s-%s' % (release, arch)
+                else:
+                    queue_name = 'debci-%s-%s-%s' % (context, release, arch)
+                (_, queue_size, _) = ch.queue_declare(queue=queue_name,
+                                                      durable=True,
+                                                      exclusive=False,
+                                                      auto_delete=False)
+                if include_requests:
+                    requests = get_queue_requests(ch, queue_name, queue_size)
+                else:
+                    requests = []
+                result.setdefault(context, {}).setdefault(release, {})[arch] = (queue_size,
+                                                                                requests)
+                all_arches.add(arch)
+
+    queued = (release_arches.keys(), sorted(all_arches), result)
+
+    with open(output, 'w') as f:
+        json.dump(queued, f)
+
+
+if __name__ == '__main__':
+    query_amqp('/tmp/queued.json')
+    query_amqp('/tmp/queued_no_requests.json')