From 28466571da5a915a763ac8cad66f1e7b5a6c062d Mon Sep 17 00:00:00 2001 From: Florent 'Skia' Jacquet Date: Tue, 21 Jul 2026 09:05:45 +0200 Subject: [PATCH 1/2] errors: add retracer AMQP queue length monitoring view Add a new view at /retracers-queue-length/ showing retracer AMQP queue lengths over the last 48 hours. Data is collected using a new timer (`record_queue_lengths.py`), and stored in Cassandra's `Indexes` table. Mostly made by DeepSeek V4 Flash using the following prompt: ``` In the `errors` app, similar to `views.retracers_results`, implement a new view showing the length of the `retracer` AMQP queue over time for the last 2 days. ``` --- charm/errortracker.py | 40 +++++++++++------------ charm/tests/integration/test_timers.py | 6 ++-- src/errors/api/resources.py | 23 +++++++++++++ src/errors/api/urls.py | 2 ++ src/errors/cassie.py | 27 ++++++++++++++++ src/errors/static/js/retracers.js | 45 ++++++++++++++++++++++++++ src/errors/templates/retracers.html | 4 +++ src/errors/urls.py | 1 + src/errors/views.py | 7 ++++ src/errortracker/amqp_utils.py | 17 +++++++++- src/tools/record_queue_lengths.py | 42 ++++++++++++++++++++++++ 11 files changed, 190 insertions(+), 24 deletions(-) create mode 100755 src/tools/record_queue_lengths.py diff --git a/charm/errortracker.py b/charm/errortracker.py index aa2033da..55e72357 100644 --- a/charm/errortracker.py +++ b/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"]) diff --git a/charm/tests/integration/test_timers.py b/charm/tests/integration/test_timers.py index b9574e50..e06b1513 100644 --- a/charm/tests/integration/test_timers.py +++ b/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" diff --git a/src/errors/api/resources.py b/src/errors/api/resources.py index 19aa0c42..1ab66482 100644 --- a/src/errors/api/resources.py +++ b/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) diff --git a/src/errors/api/urls.py b/src/errors/api/urls.py index 2d9ec745..c3baf104 100644 --- a/src/errors/api/urls.py +++ b/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()) diff --git a/src/errors/cassie.py b/src/errors/cassie.py index 619ccf46..634986c5 100644 --- a/src/errors/cassie.py +++ b/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 {} diff --git a/src/errors/static/js/retracers.js b/src/errors/static/js/retracers.js index 01ddc7e8..a54a6e62 100644 --- a/src/errors/static/js/retracers.js +++ b/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); + }); +} diff --git a/src/errors/templates/retracers.html b/src/errors/templates/retracers.html index 10f3479a..29228f01 100644 --- a/src/errors/templates/retracers.html +++ b/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 %} diff --git a/src/errors/urls.py b/src/errors/urls.py index 468bfa06..0a88fac6 100644 --- a/src/errors/urls.py +++ b/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")), diff --git a/src/errors/views.py b/src/errors/views.py index 4bc151ed..ef79b11c 100644 --- a/src/errors/views.py +++ b/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"} diff --git a/src/errortracker/amqp_utils.py b/src/errortracker/amqp_utils.py index 1c23f17d..039a0b7f 100644 --- a/src/errortracker/amqp_utils.py +++ b/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) diff --git a/src/tools/record_queue_lengths.py b/src/tools/record_queue_lengths.py new file mode 100755 index 00000000..2e57b9cf --- /dev/null +++ b/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() From c02f6487a2651d0303752d7f4daef2da9725af2e Mon Sep 17 00:00:00 2001 From: Florent 'Skia' Jacquet Date: Tue, 21 Jul 2026 09:26:59 +0200 Subject: [PATCH 2/2] retracer: ack msg even earlier --- src/retracer.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/src/retracer.py b/src/retracer.py index cf26f952..ad3793f6 100755 --- a/src/retracer.py +++ b/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