Implement a view of the AMQP queue length

charm/errortracker.py

@@ -13,8 +13,7 @@ def setup_systemd_timer(unit_name, description, command, calendar):
     systemd_unit_location = Path("/") / "etc" / "systemd" / "system"
     systemd_unit_location.mkdir(parents=True, exist_ok=True)
 
-    (systemd_unit_location / f"{unit_name}.service").write_text(
-        f"""
+    (systemd_unit_location / f"{unit_name}.service").write_text(f"""
 [Unit]
 Description={description}
 
@@ -23,10 +22,8 @@ def setup_systemd_timer(unit_name, description, command, calendar):
 User=ubuntu
 Environment=PYTHONPATH={REPO_LOCATION}/src
 ExecStart={command}
-"""
-    )
-    (systemd_unit_location / f"{unit_name}.timer").write_text(
-        f"""
+""")
+    (systemd_unit_location / f"{unit_name}.timer").write_text(f"""
 [Unit]
 Description={description}
 
@@ -36,8 +33,7 @@ def setup_systemd_timer(unit_name, description, command, calendar):
 
 [Install]
 WantedBy=timers.target
-"""
-    )
+""")
 
     check_call(["systemctl", "daemon-reload"])
     check_call(["systemctl", "enable", "--now", f"{unit_name}.timer"])
@@ -116,8 +112,7 @@ def configure_daisy(self):
         check_call(["apt-get", "install", "-y", "gunicorn"])
         systemd_unit_location = Path("/") / "etc" / "systemd" / "system"
         systemd_unit_location.mkdir(parents=True, exist_ok=True)
-        (systemd_unit_location / "daisy.service").write_text(
-            f"""
+        (systemd_unit_location / "daisy.service").write_text(f"""
 [Unit]
 Description=Daisy
 After=network.target
@@ -131,8 +126,7 @@ def configure_daisy(self):
 
 [Install]
 WantedBy=multi-user.target
-"""
-        )
+""")
 
         check_call(["systemctl", "daemon-reload"])
 
@@ -147,7 +141,9 @@ def configure_retracer(self, retracer_failed_queue: bool):
         failed = "--failed" if retracer_failed_queue else ""
         # Work around https://bugs.launchpad.net/ubuntu/+source/gdb/+bug/1818918
         # Apport will not be run as root, thus the included workaround here will hit ENOPERM
-        (Path("/") / "usr" / "lib" / "debug" / ".dwz").mkdir(parents=True, exist_ok=True)
+        (Path("/") / "usr" / "lib" / "debug" / ".dwz").mkdir(
+            parents=True, exist_ok=True
+        )
         logger.info("Installing additional retracer dependencies")
         check_call(
             [
@@ -162,8 +158,7 @@ def configure_retracer(self, retracer_failed_queue: bool):
         logger.info("Configuring retracer systemd units")
         systemd_unit_location = Path("/") / "etc" / "systemd" / "system"
         systemd_unit_location.mkdir(parents=True, exist_ok=True)
-        (systemd_unit_location / "retracer@.service").write_text(
-            f"""
+        (systemd_unit_location / "retracer@.service").write_text(f"""
 [Unit]
 Description=Retracer
 
@@ -177,8 +172,7 @@ def configure_retracer(self, retracer_failed_queue: bool):
 
 [Install]
 WantedBy=multi-user.target
-"""
-        )
+""")
 
         check_call(["systemctl", "daemon-reload"])
 
@@ -237,6 +231,12 @@ def configure_timers(self):
             f"{REPO_LOCATION}/src/tools/swift_handle_old_cores.py",
             "*-*-* 06:45:00",  # every day at 06:45
         )
+        setup_systemd_timer(
+            "et-record-queue-lengths",
+            "Error Tracker - AMQP - Record queue lengths",
+            f"{REPO_LOCATION}/src/tools/record_queue_lengths.py",
+            "*-*-* *:0/5:00",  # every five minutes
+        )
 
     def configure_errors(self):
         logger.info("Configuring errors")
@@ -258,8 +258,7 @@ def configure_errors(self):
         )
         systemd_unit_location = Path("/") / "etc" / "systemd" / "system"
         systemd_unit_location.mkdir(parents=True, exist_ok=True)
-        (systemd_unit_location / "errors.service").write_text(
-            f"""
+        (systemd_unit_location / "errors.service").write_text(f"""
 [Unit]
 Description=Error Tracker errors
 After=network.target
@@ -287,8 +286,7 @@ def configure_errors(self):
 
 [Install]
 WantedBy=multi-user.target
-"""
-        )
+""")
 
         check_call(["systemctl", "daemon-reload"])
 

charm/tests/integration/test_timers.py

@@ -35,5 +35,7 @@ def test_deploy(
     task = juju.exec("systemctl", "list-units", "-o", "json", unit="timers/0")
     units = json.loads(task.stdout)
     et_units = [u for u in units if u["unit"].startswith("et-")]
-    assert len(et_units) == 5, "wrong number of error tracker systemd units"
-    assert all([u["active"] == "active" for u in et_units]), "not all systemd units are active"
+    assert len(et_units) == 6, "wrong number of error tracker systemd units"
+    assert all(
+        [u["active"] == "active" for u in et_units]
+    ), "not all systemd units are active"

src/errors/api/resources.py

@@ -205,6 +205,29 @@ def obj_get(self, request, **kwargs):
         return ResultObject({"date": date, "value": value})
 
 
+class RetraceQueueLengthResource(ErrorsResource):
+    queue = fields.CharField(attribute="queue", readonly=True)
+    values = fields.ListField(attribute="values", readonly=True)
+
+    class Meta(ErrorsMeta):
+        resource_name = "retracer-queue-length"
+
+    def obj_get_list(self, bundle):
+        hours = int(bundle.request.GET.get("hours", 48))
+        data = cassie.get_queue_lengths(hours=hours)
+        results = []
+        for queue_name in sorted(data):
+            results.append(
+                ResultObject(
+                    {
+                        "queue": queue_name,
+                        "values": data[queue_name],
+                    }
+                )
+            )
+        return results
+
+
 class RetraceAverageProcessingTimeResource(ErrorsResource):
     date = fields.CharField(attribute="date")
     value = fields.DictField(attribute="value", readonly=True)

src/errors/api/urls.py

@@ -22,6 +22,7 @@
     ReleasePackageVersionPockets,
     ReportsStateResource,
     RetraceAverageProcessingTimeResource,
+    RetraceQueueLengthResource,
     RetraceResultResource,
     SystemCrashesResource,
     SystemImageVersionsResource,
@@ -31,6 +32,7 @@
 v1_api = Api(api_name="1.0")
 v1_api.register(RetraceResultResource())
 v1_api.register(RetraceAverageProcessingTimeResource())
+v1_api.register(RetraceQueueLengthResource())
 v1_api.register(InstanceCountResource())
 v1_api.register(ProblemCountResource())
 v1_api.register(DayOopsResource())

src/errors/cassie.py

@@ -780,3 +780,30 @@ def get_system_image_versions(image_type: str):
         return list(versions)
     except DoesNotExist:
         return None
+
+
+def record_queue_length(queue: str, length: int):
+    now = datetime.datetime.now(datetime.timezone.utc)
+    timestamp = now.strftime("%Y%m%d%H%M")
+    column1 = f"{queue}:{timestamp}"
+    Indexes.create(key=b"retrace_queue_length", column1=column1, value=str(length).encode())
+
+
+def get_queue_lengths(hours: int = 48):
+    now = datetime.datetime.now(datetime.timezone.utc)
+    cutoff = now - datetime.timedelta(hours=hours)
+    cutoff_str = cutoff.strftime("%Y%m%d%H%M")
+    try:
+        rows = Indexes.objects.filter(key=b"retrace_queue_length").all()
+        results = {}
+        for row in rows:
+            if row.column1 >= cutoff_str:
+                queue, ts = row.column1.split(":", 1)
+                if queue not in results:
+                    results[queue] = []
+                results[queue].append({"timestamp": ts, "value": int(row.value)})
+        for queue in results:
+            results[queue].sort(key=lambda x: x["timestamp"])
+        return results
+    except DoesNotExist:
+        return {}

src/errors/static/js/retracers.js

@@ -151,3 +151,48 @@ function instances_graph () {
         Y.io(uri);
     });
 }
+
+function retracers_queue_length_graph () {
+    YUI().use('node', 'io-base', 'json-parse', function (Y) {
+        var uri = '/api/1.0/retracer-queue-length/?hours=48&format=json';
+        function complete (id, o, args) {
+            var response = Y.JSON.parse(o.response);
+            var data = [];
+            for (var i in response.objects) {
+                var queue = response.objects[i];
+                var values = [];
+                for (var j in queue.values) {
+                    var ts = queue.values[j].timestamp;
+                    var year = ts.substring(0, 4);
+                    var month = ts.substring(4, 6);
+                    var day = ts.substring(6, 8);
+                    var hour = ts.substring(8, 10);
+                    var minute = ts.substring(10, 12);
+                    values.push({
+                        x: new Date(year, month - 1, day, hour, minute),
+                        y: queue.values[j].value
+                    });
+                }
+                data.push({
+                    values: values,
+                    key: queue.queue,
+                    color: '#7e2f8e'
+                });
+            }
+            nv.addGraph(function () {
+                var chart = nv.models.lineWithFocusChart();
+                chart.xAxis.tickFormat(x_axis_tick_format);
+                chart.x2Axis.tickFormat(x_axis_tick_format);
+                chart.forceY([0]);
+                chart.yAxis.axisLabel('Queue length')
+                var container = d3.select('#retracers svg').datum(data);
+
+                container.transition().duration(500).call(chart);
+                nv.utils.windowResize(chart.update);
+                return chart;
+            });
+        };
+        Y.on('io:complete', complete, Y, {});
+        Y.io(uri);
+    });
+}

src/errors/templates/retracers.html

@@ -22,6 +22,10 @@
         {% else %}
             {% if graph_type == "instances" %}
                 instances_graph();
+            {% else %}
+                {% if graph_type == "queue-length" %}
+                    retracers_queue_length_graph();
+                {% endif %}
             {% endif %}
         {% endif %}
     {% endif %}

src/errors/urls.py

@@ -15,6 +15,7 @@
     re_path(r"problem/(.*)$", views.problem),
     re_path(r"^retracers-average-processing-time/", views.retracers_average_processing_time),
     re_path(r"^retracers-results/", views.retracers_results),
+    re_path(r"^retracers-queue-length/", views.retracers_queue_length),
     re_path(r"^status/?$", views.status),
     re_path(r"^user/(.*)$", views.user),
     re_path(r"^api/", include("errors.api.urls")),

src/errors/views.py

@@ -110,6 +110,13 @@ def retracers_results(request):
     return render(request, "retracers.html", c)
 
 
+@measure_view
+def retracers_queue_length(request):
+    c = {"graph_type": "queue-length"}
+    c.update(common_c())
+    return render(request, "retracers.html", c)
+
+
 @measure_view
 def instances_count(request):
     c = {"graph_type": "instances"}

src/errortracker/amqp_utils.py

@@ -66,6 +66,19 @@ def get_connection():
         raise
 
 
+def get_queue_length(queue: str) -> int | None:
+    channel = get_connection().channel()
+    if not channel:
+        return None
+    try:
+        _, message_count, _ = channel.queue_declare(queue=queue, passive=True)
+        return message_count
+    except amqplib_error_types + (amqp.exceptions.NotFound,):
+        return None
+    finally:
+        channel.close()
+
+
 def enqueue(message: str, queue: str):
     channel = get_connection().channel()
     if not channel:
@@ -75,7 +88,9 @@ def enqueue(message: str, queue: str):
         # We'll use this timestamp to measure how long it takes to process a
         # retrace, from receiving the core file to writing the data back to
         # Cassandra.
-        body = amqp.Message(message, timestamp=int(datetime.now(timezone.utc).timestamp()))
+        body = amqp.Message(
+            message, timestamp=int(datetime.now(timezone.utc).timestamp())
+        )
         # Persistent
         body.properties["delivery_mode"] = 2
         channel.basic_publish(body, exchange="", routing_key=queue)

src/retracer.py

@@ -368,6 +368,13 @@ def save_crash(self, report, oops_id, core_file):
     def callback(self, msg):
         self._processing_callback = True
         log("Processing.")
+
+        # ack the message very early, to prevent them from staying forever
+        # in the queue in case the retracer gets OOM-killed or Cassandra is
+        # unreachable
+        log("ack'ing message from queue")
+        msg.channel.basic_ack(msg.delivery_tag)
+
         self.msg_body = ensure_str(msg.body)
         oops_id, provider = self.msg_body.split(":", 1)
         try:
@@ -383,11 +390,6 @@ def callback(self, msg):
             metrics.meter("could_not_find_oops")
             return
 
-        # ack the message very early, to prevent them from staying forever in
-        # the queue in case the retracer gets OOM-killed
-        log("ack'ing message from queue")
-        msg.channel.basic_ack(msg.delivery_tag)
-
         # There are some items still in amqp queue that have already been
         # retraced, check for this and ack the message.
         # N.B.: This only works in some cases because we don't mark a report as

src/tools/record_queue_lengths.py

@@ -0,0 +1,42 @@
+#!/usr/bin/python3
+
+import sys
+
+from errors import cassie
+from errortracker import amqp_utils, cassandra
+
+cassandra.setup_cassandra()
+
+ARCHES = ["amd64", "arm64", "armhf", "i386"]
+
+
+def main():
+    if "--dry-run" in sys.argv:
+        dry_run = True
+        sys.argv.remove("--dry-run")
+    else:
+        dry_run = False
+
+    for arch in ARCHES:
+        queue = f"retrace_{arch}"
+        length = amqp_utils.get_queue_length(queue)
+        if length is None:
+            print(f"{queue}: connection error")
+            continue
+        print(f"{queue}: {length}")
+        if not dry_run:
+            cassie.record_queue_length(queue, length)
+
+    for arch in ARCHES:
+        queue = f"failed_retrace_{arch}"
+        length = amqp_utils.get_queue_length(queue)
+        if length is None:
+            print(f"{queue}: connection error")
+            continue
+        print(f"{queue}: {length}")
+        if not dry_run:
+            cassie.record_queue_length(queue, length)
+
+
+if __name__ == "__main__":
+    main()