From f07d76c6ee98a4721617b56565b4c7e5f1643146 Mon Sep 17 00:00:00 2001 From: Nicholas Devenish Date: Mon, 26 Oct 2026 15:58:51 +0000 Subject: [PATCH 1/2] Subscribe services with self.subscribe workflows' CommonService.subscribe defaults to acknowledging messages and extending log records with the recipe ID, which every service passed explicitly, and applies queue overrides given when the service is started. This replaces the hand-written overrides in the filewatcher, per-image analysis and X-ray centering services, and makes the queue overridable for every other recipe service. Services consuming several queues (archiver, s3echo collector and uploader) now refuse to start with an unnamed queue override, which they previously ignored. Use --queue NAME=QUEUE for these. Requires workflows with CommonService.subscribe. --- src/dlstbx/services/archiver.py | 17 ++------------- src/dlstbx/services/cloudwatcher.py | 9 +------- src/dlstbx/services/cluster.py | 9 +------- src/dlstbx/services/filewatcher.py | 9 +------- src/dlstbx/services/images.py | 8 +------- src/dlstbx/services/indexer.py | 9 ++------ src/dlstbx/services/ispybsvc.py | 6 +----- src/dlstbx/services/ispybsvc_pia.py | 10 +-------- src/dlstbx/services/mimas.py | 9 +------- src/dlstbx/services/mimas_backlog.py | 10 +-------- src/dlstbx/services/nexusparser.py | 9 +------- src/dlstbx/services/notifygda.py | 6 +----- src/dlstbx/services/per_image_analysis.py | 8 +------- src/dlstbx/services/s3echo_collector.py | 17 ++------------- src/dlstbx/services/s3echo_uploader.py | 25 +++-------------------- src/dlstbx/services/s3echo_watcher.py | 9 +------- src/dlstbx/services/ssx_plotter.py | 8 +------- src/dlstbx/services/strategy.py | 8 +------- src/dlstbx/services/trigger.py | 8 +------- src/dlstbx/services/trigger_xchem.py | 8 +------- src/dlstbx/services/validation.py | 9 +------- src/dlstbx/services/xray_centering.py | 7 ++----- 22 files changed, 28 insertions(+), 190 deletions(-) diff --git a/src/dlstbx/services/archiver.py b/src/dlstbx/services/archiver.py index cc989cfbc..5832b2308 100644 --- a/src/dlstbx/services/archiver.py +++ b/src/dlstbx/services/archiver.py @@ -8,7 +8,6 @@ from datetime import datetime from pathlib import Path -import workflows.recipe from workflows.services.common_service import CommonService from dlstbx.util import INDUSTRIAL_CODES @@ -112,20 +111,8 @@ def initializing(self): """Subscribe to the archiver queue. Received messages must be acknowledged.""" self.log.info("Archiver starting") - workflows.recipe.wrap_subscribe( - self._transport, - "archive.pattern", - self.archive_dcid, - acknowledgement=True, - log_extender=self.extend_log, - ) - workflows.recipe.wrap_subscribe( - self._transport, - "archive.filelist", - self.archive_filelist, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("archive.pattern", self.archive_dcid) + self.subscribe("archive.filelist", self.archive_filelist) @staticmethod def rangifier(numbers): diff --git a/src/dlstbx/services/cloudwatcher.py b/src/dlstbx/services/cloudwatcher.py index 5b309ddba..0a61295ce 100644 --- a/src/dlstbx/services/cloudwatcher.py +++ b/src/dlstbx/services/cloudwatcher.py @@ -4,7 +4,6 @@ from math import log2 from pprint import pformat -import workflows.recipe from requests.exceptions import HTTPError from workflows.services.common_service import CommonService from zocalo.util import slurm @@ -43,13 +42,7 @@ def initializing(self): "iris": self.iris_api, } - workflows.recipe.wrap_subscribe( - self._transport, - "cloudwatcher", - self.watch_jobs, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("cloudwatcher", self.watch_jobs) def watch_jobs(self, rw, header, message): """ diff --git a/src/dlstbx/services/cluster.py b/src/dlstbx/services/cluster.py index 7f5591015..091e6e86f 100644 --- a/src/dlstbx/services/cluster.py +++ b/src/dlstbx/services/cluster.py @@ -13,7 +13,6 @@ import pydantic import requests -import workflows.recipe from workflows.services.common_service import CommonService from zocalo.configuration import Configuration from zocalo.util import slurm @@ -179,13 +178,7 @@ def initializing(self): ) } self.log.debug(f"Supported schedulers: {', '.join(self.schedulers.keys())}") - workflows.recipe.wrap_subscribe( - self._transport, - "cluster.submission", - self.run_submit_job, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("cluster.submission", self.run_submit_job) def _write_job_file( self, header, path: str, description: str, content: str diff --git a/src/dlstbx/services/filewatcher.py b/src/dlstbx/services/filewatcher.py index cd6950035..182623ddb 100644 --- a/src/dlstbx/services/filewatcher.py +++ b/src/dlstbx/services/filewatcher.py @@ -6,7 +6,6 @@ from pathlib import Path import h5py -import workflows.recipe from workflows.recipe import RecipeWrapper from workflows.services.common_service import CommonService @@ -98,13 +97,7 @@ def initializing(self): acknowledged. """ self.log.info("Filewatcher starting") - workflows.recipe.wrap_subscribe( - self._transport, - self._environment.get("queue") or "filewatcher", - self.watch_files, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("filewatcher", self.watch_files) def watch_files(self, rw, header, message): """Check for presence of files.""" diff --git a/src/dlstbx/services/images.py b/src/dlstbx/services/images.py index 282e01bd1..99fbb2e51 100644 --- a/src/dlstbx/services/images.py +++ b/src/dlstbx/services/images.py @@ -56,13 +56,7 @@ def initializing(self): group="zocalo.services.images.plugins" ) } - workflows.recipe.wrap_subscribe( - self._transport, - "images", - self.image_call, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("images", self.image_call) def image_call(self, rw, header, message): """Pass incoming message to the relevant plugin function.""" diff --git a/src/dlstbx/services/indexer.py b/src/dlstbx/services/indexer.py index 82d21cdd4..c97afc2ab 100644 --- a/src/dlstbx/services/indexer.py +++ b/src/dlstbx/services/indexer.py @@ -81,13 +81,8 @@ class DLSIndexer(CommonService): def initializing(self): logging.getLogger("dials").setLevel(logging.WARNING) - workflows.recipe.wrap_subscribe( - self._transport, - "index", - self.index, - acknowledgement=True, - log_extender=self.extend_log, - mangle_for_receiving=msgpack_mangle_for_receiving, + self.subscribe( + "index", self.index, mangle_for_receiving=msgpack_mangle_for_receiving ) @pydantic.validate_call(config={"arbitrary_types_allowed": True}) diff --git a/src/dlstbx/services/ispybsvc.py b/src/dlstbx/services/ispybsvc.py index d85891369..d2ed86fcb 100644 --- a/src/dlstbx/services/ispybsvc.py +++ b/src/dlstbx/services/ispybsvc.py @@ -11,7 +11,6 @@ import mysql.connector import pydantic import sqlalchemy.orm -import workflows.recipe from ispyb.sqlalchemy import PDB, ProteinHasPDB from workflows.services.common_service import CommonService @@ -68,12 +67,9 @@ def initializing(self): exc_info=True, ) self.log.info("ISPyB service ready") - workflows.recipe.wrap_subscribe( - self._transport, + self.subscribe( "ispyb_connector", # will become 'ispyb' in far future self.receive_msg, - acknowledgement=True, - log_extender=self.extend_log, allow_non_recipe_messages=True, ) diff --git a/src/dlstbx/services/ispybsvc_pia.py b/src/dlstbx/services/ispybsvc_pia.py index cbdfad963..9aac1a63f 100644 --- a/src/dlstbx/services/ispybsvc_pia.py +++ b/src/dlstbx/services/ispybsvc_pia.py @@ -2,7 +2,6 @@ import ispyb import sqlalchemy -import workflows.recipe from dlstbx.services.ispybsvc import DLSISPyB @@ -33,11 +32,4 @@ def initializing(self): ) ) self.log.debug("ISPyB PIA inserter starting") - workflows.recipe.wrap_subscribe( - self._transport, - "ispyb_pia", - self.receive_msg, - acknowledgement=True, - log_extender=self.extend_log, - allow_non_recipe_messages=True, - ) + self.subscribe("ispyb_pia", self.receive_msg, allow_non_recipe_messages=True) diff --git a/src/dlstbx/services/mimas.py b/src/dlstbx/services/mimas.py index e99c562da..dc3e2664e 100644 --- a/src/dlstbx/services/mimas.py +++ b/src/dlstbx/services/mimas.py @@ -4,7 +4,6 @@ from pprint import pformat from typing import Any -import workflows.recipe from workflows.services.common_service import CommonService from dlstbx import mimas @@ -40,13 +39,7 @@ def initializing(self): "s3echo": {"total": 0.0, "last_cluster_update": time.time()}, } - workflows.recipe.wrap_subscribe( - self._transport, - "mimas", - self.process, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("mimas", self.process) # Subscribe to the transient.statistics.cluster topic, which we will # examine to determine the number of waiting jobs diff --git a/src/dlstbx/services/mimas_backlog.py b/src/dlstbx/services/mimas_backlog.py index dac4c715a..0dea391d4 100644 --- a/src/dlstbx/services/mimas_backlog.py +++ b/src/dlstbx/services/mimas_backlog.py @@ -3,7 +3,6 @@ import logging import time -import workflows.recipe from workflows.services.common_service import CommonService @@ -27,14 +26,7 @@ def initializing(self): # Subscribe to the mimas.held queue, which contains the held mimas # recipes we would like to drip-feed to the dispatcher - workflows.recipe.wrap_subscribe( - self._transport, - "mimas.held", - self.on_mimas_held, - acknowledgement=True, - exclusive=True, - log_extender=self.extend_log, - ) + self.subscribe("mimas.held", self.on_mimas_held, exclusive=True) # Subscribe to the transient.statistics.cluster topic, which we will # examine to determine the number of waiting jobs diff --git a/src/dlstbx/services/nexusparser.py b/src/dlstbx/services/nexusparser.py index 9fa2746a0..4f589bfee 100644 --- a/src/dlstbx/services/nexusparser.py +++ b/src/dlstbx/services/nexusparser.py @@ -2,7 +2,6 @@ import os.path -import workflows.recipe from workflows.services.common_service import CommonService import dlstbx.util.hdf5 @@ -20,13 +19,7 @@ class DLSNexusParser(CommonService): def initializing(self): """Subscribe to a queue. Received messages must be acknowledged.""" self.log.info("Nexusparser starting") - workflows.recipe.wrap_subscribe( - self._transport, - "nexusparser.find_related_files", - self.find_related_files, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("nexusparser.find_related_files", self.find_related_files) def find_related_files(self, rw, header, message): """Take a single file and recursively find all referenced files.""" diff --git a/src/dlstbx/services/notifygda.py b/src/dlstbx/services/notifygda.py index 6bc270c10..402321208 100644 --- a/src/dlstbx/services/notifygda.py +++ b/src/dlstbx/services/notifygda.py @@ -1,6 +1,5 @@ from __future__ import annotations -import workflows.recipe from workflows.services.common_service import CommonService import dlstbx.util.gda @@ -21,12 +20,9 @@ def initializing(self): Received messages must be acknowledged. """ self.log.debug("GDA Bridge starting") - workflows.recipe.wrap_subscribe( - self._transport, + self.subscribe( "notify_gda", # consider transient queue self.notify_gda, - acknowledgement=True, - log_extender=self.extend_log, ) def notify_gda(self, rw, header, message): diff --git a/src/dlstbx/services/per_image_analysis.py b/src/dlstbx/services/per_image_analysis.py index 4886e1dec..e1661c1f6 100644 --- a/src/dlstbx/services/per_image_analysis.py +++ b/src/dlstbx/services/per_image_analysis.py @@ -90,13 +90,7 @@ def initializing(self): # The main per_image_analysis queue. # For every received message a single frame will be analysed. - workflows.recipe.wrap_subscribe( - self._transport, - self._environment.get("queue") or "per_image_analysis", - self.per_image_analysis, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("per_image_analysis", self.per_image_analysis) # @pydantic.validate_call(config=dict(arbitrary_types_allowed=True)) def per_image_analysis( diff --git a/src/dlstbx/services/s3echo_collector.py b/src/dlstbx/services/s3echo_collector.py index 5b74b6a54..e1d542382 100644 --- a/src/dlstbx/services/s3echo_collector.py +++ b/src/dlstbx/services/s3echo_collector.py @@ -1,7 +1,6 @@ from __future__ import annotations import minio -import workflows.recipe from workflows.services.common_service import CommonService from dlstbx.util import iris @@ -34,21 +33,9 @@ def initializing(self): self._message_delay = 5 - workflows.recipe.wrap_subscribe( - self._transport, - "s3echo.start", - self.on_start, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("s3echo.start", self.on_start) - workflows.recipe.wrap_subscribe( - self._transport, - "s3echo.end", - self.on_end, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("s3echo.end", self.on_end) def on_start(self, rw, header, message): """ diff --git a/src/dlstbx/services/s3echo_uploader.py b/src/dlstbx/services/s3echo_uploader.py index 94ed2d32b..6f0ecedaf 100644 --- a/src/dlstbx/services/s3echo_uploader.py +++ b/src/dlstbx/services/s3echo_uploader.py @@ -4,7 +4,6 @@ from pprint import pformat import prometheus_client -import workflows.recipe from minio.error import S3Error from workflows.services.common_service import CommonService @@ -83,29 +82,11 @@ def initializing(self): self._message_delay = 5 - workflows.recipe.wrap_subscribe( - self._transport, - "s3echo.upload.rpid", - self.on_upload_rpid, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("s3echo.upload.rpid", self.on_upload_rpid) - workflows.recipe.wrap_subscribe( - self._transport, - "s3echo.upload.dcid", - self.on_upload_dcid, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("s3echo.upload.dcid", self.on_upload_dcid) - workflows.recipe.wrap_subscribe( - self._transport, - "s3echo.download", - self.on_download, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("s3echo.download", self.on_download) # Initialise metrics if requested if self._environment.get("metrics"): diff --git a/src/dlstbx/services/s3echo_watcher.py b/src/dlstbx/services/s3echo_watcher.py index bfbaa5624..6e136652d 100644 --- a/src/dlstbx/services/s3echo_watcher.py +++ b/src/dlstbx/services/s3echo_watcher.py @@ -4,7 +4,6 @@ from pprint import pformat import minio -import workflows.recipe from minio.error import S3Error from workflows.services.common_service import CommonService @@ -41,13 +40,7 @@ def initializing(self): self._message_delay = 5 - workflows.recipe.wrap_subscribe( - self._transport, - "s3echo.watch", - self.watch, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("s3echo.watch", self.watch) def watch(self, rw, header, message): """ diff --git a/src/dlstbx/services/ssx_plotter.py b/src/dlstbx/services/ssx_plotter.py index def9234a2..0b29ab389 100644 --- a/src/dlstbx/services/ssx_plotter.py +++ b/src/dlstbx/services/ssx_plotter.py @@ -74,13 +74,7 @@ class SSXPlotter(CommonService): _logger_name = "dlstbx.services.ssx_plotter" def initializing(self): - workflows.recipe.wrap_subscribe( - self._transport, - "ssx.plot", - self.receive_msg, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("ssx.plot", self.receive_msg) @pydantic.validate_call(config={"arbitrary_types_allowed": True}) def receive_msg( diff --git a/src/dlstbx/services/strategy.py b/src/dlstbx/services/strategy.py index aa759f66e..c773ed5db 100644 --- a/src/dlstbx/services/strategy.py +++ b/src/dlstbx/services/strategy.py @@ -174,13 +174,7 @@ class DLSStrategy(CommonService): def initializing(self): """Subscribe to channel.""" self.log.info("Strategy service starting") - workflows.recipe.wrap_subscribe( - self._transport, - "strategy", - self.generate_strategy, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("strategy", self.generate_strategy) def generate_strategy( self, rw: workflows.recipe.RecipeWrapper, header: dict, message: dict diff --git a/src/dlstbx/services/trigger.py b/src/dlstbx/services/trigger.py index 1840e2a76..622ec02fe 100644 --- a/src/dlstbx/services/trigger.py +++ b/src/dlstbx/services/trigger.py @@ -328,13 +328,7 @@ def initializing(self): ispyb.sqlalchemy.url(), connect_args={"use_pure": True} ) ) - workflows.recipe.wrap_subscribe( - self._transport, - "trigger", - self.trigger, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("trigger", self.trigger) self.ispyb = ispyb.open() # Initialise metrics if requested diff --git a/src/dlstbx/services/trigger_xchem.py b/src/dlstbx/services/trigger_xchem.py index c0d15146b..8c7194a9f 100644 --- a/src/dlstbx/services/trigger_xchem.py +++ b/src/dlstbx/services/trigger_xchem.py @@ -125,13 +125,7 @@ def initializing(self): ispyb.sqlalchemy.url(), connect_args={"use_pure": True} ) ) - workflows.recipe.wrap_subscribe( - self._transport, - "trigger_xchem", - self.trigger, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("trigger_xchem", self.trigger) self.ispyb = ispyb.open() # Initialise metrics if requested diff --git a/src/dlstbx/services/validation.py b/src/dlstbx/services/validation.py index 78f92f964..0bc0dd9be 100644 --- a/src/dlstbx/services/validation.py +++ b/src/dlstbx/services/validation.py @@ -8,7 +8,6 @@ import h5py import numpy as np import pytest -import workflows.recipe try: from dxtbx.format.FormatNXmxDLS import find_meta_filename @@ -35,13 +34,7 @@ class DLSValidation(CommonService): def initializing(self): # The main per_image_analysis queue. # For every received message a single frame will be analysed. - workflows.recipe.wrap_subscribe( - self._transport, - "validation", - self.validate, - acknowledgement=True, - log_extender=self.extend_log, - ) + self.subscribe("validation", self.validate) def fail_validation(self, rw, header, output, reason): output["reason"] = reason diff --git a/src/dlstbx/services/xray_centering.py b/src/dlstbx/services/xray_centering.py index d8ec59d67..35ade9c72 100644 --- a/src/dlstbx/services/xray_centering.py +++ b/src/dlstbx/services/xray_centering.py @@ -124,13 +124,10 @@ def initializing(self): self._next_garbage_collection = time.time() + 60 self._register_idle(60, self.garbage_collect) - workflows.recipe.wrap_subscribe( - self._transport, - self._environment.get("queue") or "reduce.xray_centering", + self.subscribe( + "reduce.xray_centering", self.add_pia_result, - acknowledgement=True, exclusive=True, - log_extender=self.extend_log, prefetch_count=65535, ) From 52140965ee099ce5676f5a3ccb9354acb769102d Mon Sep 17 00:00:00 2001 From: Nicholas Devenish Date: Mon, 26 Oct 2026 13:17:50 +0000 Subject: [PATCH 2/2] Use the service starter's --queue option workflows' service starter provides -q/--queue [NAME=]QUEUE, but only when the starter subclass hasn't defined its own. Removing the dlstbx option, which only accepted a single queue, enables it. A single queue is still passed to services the same way, and services consuming several queues can now have each one overridden by name. --- src/dlstbx/cli/service.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/src/dlstbx/cli/service.py b/src/dlstbx/cli/service.py index fe06d76e3..c8ba8aac4 100644 --- a/src/dlstbx/cli/service.py +++ b/src/dlstbx/cli/service.py @@ -110,13 +110,6 @@ def on_parser_preparation(self, parser): default=False, help="Restart service on failure", ) - parser.add_option( - "-q", - "--queue", - dest="queue", - default=None, - help="Consume messages from this queue", - ) self._zc.add_command_line_options(parser) self.log.debug("Launching %r", sys.argv) @@ -137,8 +130,6 @@ def before_frontend_construction(self, kwargs): kwargs["environment"] = kwargs.get("environment", {}) kwargs["environment"]["live"] = self.use_live_infrastructure # XXX deprecated kwargs["environment"]["config"] = self._zc - if self.options.queue: - kwargs["environment"]["queue"] = self.options.queue return kwargs def on_frontend_preparation(self, frontend):