From 100f79b413288a1bbf5b8664f2d449affaa15492 Mon Sep 17 00:00:00 2001 From: tmaeno Date: Wed, 28 Oct 2026 22:16:11 +0100 Subject: [PATCH 1/3] Look up each dataset name in both of its forms job/get_datasets_by_name read ATLAS_PANDA.Datasets by the names as given, but a dataset's row can be under either scope:name or name. insertDataset skips a dataset whose vuid is already in the table, and the vuid is md5(scope:name) whichever form the name came in, so the table keeps one row per dataset under the form its first writer used: the prodDBlock of an input, which may carry a scope, or the destinationDBlock of an output. A caller asking with the other form got no row, and the live harness had to send both forms itself. Each name is now looked up as given and in its other form: a scoped name also without its scope, and an unscoped one also with the scope extract_scope gives, which is the scope the vuid is made with. Every row says in asked_as which given names it answers. The IN-list cap and the probe budget count these keys, and the answers carry keys_per_name, so a call may give budget // (keys_per_name * partitions) names, 100 on production's 5 partitions instead of 200. --- pandaserver/api/v1/job_api.py | 87 ++++++++++++++++++----- pandaserver/api/v1/tests/job_api_tests.py | 67 ++++++++++++++--- 2 files changed, 126 insertions(+), 28 deletions(-) diff --git a/pandaserver/api/v1/job_api.py b/pandaserver/api/v1/job_api.py index 1f596c8a4..8ac3830fd 100644 --- a/pandaserver/api/v1/job_api.py +++ b/pandaserver/api/v1/job_api.py @@ -18,6 +18,7 @@ request_validation, ) from pandaserver.config import panda_config +from pandaserver.dataservice.ddm import rucioAPI from pandaserver.jobdispatcher import Protocol from pandaserver.srvcore.panda_request import PandaRequest from pandaserver.taskbuffer import JobUtils @@ -35,10 +36,14 @@ # partition of jobs_StatusLog MAX_STATUS_HISTORY_JOBS = 1000 -# most names one get_datasets_by_name call may give: what Oracle takes in one IN list. The -# db_read_probe_budget normally bounds it lower, since each name costs one index lookup per +# most keys one get_datasets_by_name call may look up: what Oracle takes in one IN list. Each name +# is looked up in both of its forms, so a name costs KEYS_PER_DATASET_NAME keys. The +# db_read_probe_budget normally bounds it lower, since each key costs one index lookup per # partition of Datasets -MAX_DATASET_NAMES = 1000 +MAX_DATASET_KEYS = 1000 + +# keys one name given to get_datasets_by_name is looked up with: the name as given and its other form +KEYS_PER_DATASET_NAME = 2 # slack put on both ends of a get_status_history window before it is widened to whole days. A # creationTime is written with the clock of a server and jobs_StatusLog with the database clock, so @@ -63,6 +68,28 @@ def _parse_utc_time(value: str) -> datetime.datetime: return parsed +def _dataset_name_forms(name: str) -> list[str]: + """ + The forms a dataset name is looked up in: the name as given, then its other form, which is the + name without its scope when it has one, and with a scope put on when it has none. + + The Datasets table keeps one row per vuid, under the name its first writer used, and the vuid is + made from scope:name whichever form the name came in, so the row of a dataset may be under either + form. The scope put on is the one extract_scope gives, since the vuid is made with that scope. + + Args: + name(str): the dataset name as given + + Returns: + list: the name, followed by its other form unless that is empty or the name itself + """ + scope, bare_name = rucioAPI.extract_scope(name) + other_form = bare_name if ":" in name else f"{scope}:{bare_name}" + if other_form and other_form != name: + return [name, other_form] + return [name] + + def init_task_buffer(task_buffer: TaskBuffer) -> None: """ Initialize the task buffer. This method needs to be called before any other method in this module. @@ -328,10 +355,15 @@ def get_datasets_by_name(req: PandaRequest, names: List[str]) -> Dict[str, Any]: Gets the rows of the PanDA Datasets table (the datasets jobs work with, such as the _sub and _dis ones made for them, not JEDI_Datasets) for the given names. Requires a secure connection. - Each name is looked up once in every partition of the table, which is partitioned on modificationdate, so the most - names one call may give is the db_read_probe_budget of the server divided by the number of partitions. A call giving - more is refused, not cut short. Both numbers come back with every answer, so a caller can split its names into - calls that fit. + The table keeps one row per dataset, under the form of the name its first writer used, scope:name or name. So each + name is looked up in both forms: a name with a scope also without it, and a name without one also with the scope + its first part gives (up to the first dot, or up to the second for user and group datasets), which is the scope + the vuid of the dataset is made with. Every row says which of the given names it answers. + + Each form is a key, looked up once in every partition of the table, which is partitioned on modificationdate, so + the most names one call may give is the db_read_probe_budget of the server divided by keys_per_name times the + number of partitions. A call giving more is refused, not cut short. The three numbers come back with every answer, + so a caller can split its names into calls that fit. POST is accepted too, with the arguments in a JSON body (Content-Type: application/json). A GET puts every name in the URL, and a web server refuses a request line over its limit with an HTTP error before the call gets here. A @@ -350,20 +382,26 @@ def get_datasets_by_name(req: PandaRequest, names: List[str]) -> Dict[str, Any]: Response data: dict: The rows of the datasets, and what tells how complete they are. - {"rows": [...], "as_of": datetime, "retained_from": datetime or None, "partitions": int, "budget": int} + {"rows": [...], "as_of": datetime, "retained_from": datetime or None, "partitions": int, "budget": int, + "keys_per_name": int} - rows: one dict per row, ordered by name and creationdate, with every column: vuid, name, version, type, status, numberfiles, currentfiles, creationdate, modificationdate, MoverID, transferStatus and subType, - with null for NULL. A name without rows has none + with null for NULL, and asked_as: the given names, in the order given, that the row answers in either + form. A name without rows has none - as_of: database time in UTC taken before the rows were read - retained_from: time from which the table holds every row, on modificationdate, or None when it cannot be told. Old partitions are dropped, so a dataset last modified before this time may have been in the table - - partitions, budget: the number of partitions, counted as 1 when there are none, and the - db_read_probe_budget; a call may give at most budget // partitions names + - partitions, budget, keys_per_name: the number of partitions, counted as 1 when there are none, the + db_read_probe_budget, and the keys a name is looked up with; a call may give at most + budget // (keys_per_name * partitions) names On failure, with the reason in the message, the reason as values: - {"rejected": "no_keys"}: no names - - {"rejected": "too_many_keys", "keys": int, "max_keys": int}: more names than one IN list takes - - {"rejected": "probe_budget", "keys": int, "partitions": int, "budget": int}: keys x partitions exceed the budget + - {"rejected": "too_many_keys", "keys": int, "max_keys": int, "keys_per_name": int}: more keys than one IN + list takes + - {"rejected": "probe_budget", "keys": int, "partitions": int, "budget": int, "keys_per_name": int}: keys x + partitions exceed the budget - {"error": str}: the database failed, with the error code (ORA-xxxxx, or the SQLSTATE on Postgres) + keys counts the forms looked up, and a form shared by two names, such as s:n given with n, counts once. """ tmp_logger = LogWrapper(_logger, f"get_datasets_by_name n_names={len(names)}") tmp_logger.debug("Start") @@ -374,16 +412,31 @@ def get_datasets_by_name(req: PandaRequest, names: List[str]) -> Dict[str, Any]: message = "names must not be empty" tmp_logger.debug(message) return generate_response(False, message=message, data={"rejected": "no_keys"}) - if len(names) > MAX_DATASET_NAMES: - message = f"at most {MAX_DATASET_NAMES} names are accepted, got {len(names)}" + + # each name is looked up in both of its forms, and a form shared by two names, such as s:n given + # with n, is looked up once for both. The given names of each key, in the order given + asked_as: dict[str, list[str]] = {} + for name in names: + for key in _dataset_name_forms(name): + asked_as.setdefault(key, []).append(name) + keys = list(asked_as) + if len(keys) > MAX_DATASET_KEYS: + message = f"at most {MAX_DATASET_KEYS} keys are looked up, got {len(keys)} for {len(names)} names" tmp_logger.debug(message) - return generate_response(False, message=message, data={"rejected": "too_many_keys", "keys": len(names), "max_keys": MAX_DATASET_NAMES}) + return generate_response( + False, message=message, data={"rejected": "too_many_keys", "keys": len(keys), "max_keys": MAX_DATASET_KEYS, "keys_per_name": KEYS_PER_DATASET_NAME} + ) - success, message, data = global_task_buffer.get_datasets_by_name(names, panda_config.db_read_probe_budget) + success, message, data = global_task_buffer.get_datasets_by_name(keys, panda_config.db_read_probe_budget) if not success: + if data.get("rejected") == "probe_budget": + data["keys_per_name"] = KEYS_PER_DATASET_NAME tmp_logger.debug(f"Failed with {message}") return generate_response(False, message=message, data=data) + for row in data["rows"]: + row["asked_as"] = asked_as.get(row["name"], []) + data["keys_per_name"] = KEYS_PER_DATASET_NAME tmp_logger.debug(f"""Done with {len(data["rows"])} rows""") return generate_response(True, data=data) diff --git a/pandaserver/api/v1/tests/job_api_tests.py b/pandaserver/api/v1/tests/job_api_tests.py index 85e77314e..f62359dcb 100644 --- a/pandaserver/api/v1/tests/job_api_tests.py +++ b/pandaserver/api/v1/tests/job_api_tests.py @@ -162,14 +162,14 @@ def test_get_datasets_by_name(self) -> None: self.assertEqual(status, 0) self.assertTrue(output["success"], output["message"]) self.assertEqual(output["data"]["rows"], []) - self.assertLessEqual(2 * output["data"]["partitions"], output["data"]["budget"]) + self.assertLessEqual(2 * output["data"]["keys_per_name"] * output["data"]["partitions"], output["data"]["budget"]) def test_get_datasets_by_name_over_budget(self) -> None: # the most names a call may give, plus one url = f"{api_url_ssl}/job/get_datasets_by_name" status, output = self.http_client.post(url, {"names": ["no.such.dataset"]}) self.assertTrue(output["success"], output["message"]) - n_names = output["data"]["budget"] // output["data"]["partitions"] + 1 + n_names = output["data"]["budget"] // (output["data"]["keys_per_name"] * output["data"]["partitions"]) + 1 status, output = self.http_client.post(url, {"names": [f"no.such.dataset_sub{i}" for i in range(n_names)]}) print(output) self.assertEqual(status, 0) @@ -308,27 +308,72 @@ def _call(self, names: Any, method: str = "POST", answer: tuple[bool, str, dict[ out = job_api.get_datasets_by_name(self._request(method), names=names) return out, task_buffer + # the same dataset in both forms of its name + SCOPED = "mc23_13p6TeV:mc23_13p6TeV.601229.PhPy8EG.evgen.EVNT.e8514_tid41234567_00" + BARE = "mc23_13p6TeV.601229.PhPy8EG.evgen.EVNT.e8514_tid41234567_00" + + def _answer(self, *row_names: str) -> tuple[bool, str, dict[str, Any]]: + """A successful lookup with one row for each of the given names.""" + return True, "", dict(self.ANSWER, rows=[{"vuid": f"vuid-{i}", "name": name} for i, name in enumerate(row_names)]) + def test_names_looked_up_once(self) -> None: - out, task_buffer = self._call(["a_sub1", "b_sub2", "a_sub1"]) + out, task_buffer = self._call(["a.x_sub1", "b.y_sub2", "a.x_sub1"]) + self.assertTrue(out["success"], out["message"]) + self.assertEqual(task_buffer.get_datasets_by_name.call_args[0], (["a.x_sub1", "a:a.x_sub1", "b.y_sub2", "b:b.y_sub2"], 1000)) + + def test_scoped_name_looked_up_without_scope(self) -> None: + out, task_buffer = self._call([self.SCOPED]) + self.assertEqual(task_buffer.get_datasets_by_name.call_args[0][0], [self.SCOPED, self.BARE]) + + def test_unscoped_name_looked_up_with_scope(self) -> None: + out, task_buffer = self._call([self.BARE]) + self.assertEqual(task_buffer.get_datasets_by_name.call_args[0][0], [self.BARE, self.SCOPED]) + # the scope of a user dataset is its first two parts + out, task_buffer = self._call(["user.alice.test.v1_sub123"]) + self.assertEqual(task_buffer.get_datasets_by_name.call_args[0][0], ["user.alice.test.v1_sub123", "user.alice:user.alice.test.v1_sub123"]) + + def test_no_empty_form(self) -> None: + out, task_buffer = self._call(["s:"]) + self.assertEqual(task_buffer.get_datasets_by_name.call_args[0][0], ["s:"]) + + def test_both_forms_given(self) -> None: + out, task_buffer = self._call([self.SCOPED, self.BARE], answer=self._answer(self.BARE, self.SCOPED)) self.assertTrue(out["success"], out["message"]) - self.assertEqual(task_buffer.get_datasets_by_name.call_args[0], (["a_sub1", "b_sub2"], 1000)) + self.assertEqual(task_buffer.get_datasets_by_name.call_args[0][0], [self.SCOPED, self.BARE]) + self.assertEqual([row["asked_as"] for row in out["data"]["rows"]], [[self.SCOPED, self.BARE], [self.SCOPED, self.BARE]]) + + def test_rows_say_which_names_they_answer(self) -> None: + out, task_buffer = self._call([self.SCOPED, "other.x_sub9"], answer=self._answer(self.BARE, "other:other.x_sub9")) + self.assertTrue(out["success"], out["message"]) + self.assertEqual( + [(row["name"], row["asked_as"]) for row in out["data"]["rows"]], [(self.BARE, [self.SCOPED]), ("other:other.x_sub9", ["other.x_sub9"])] + ) + self.assertEqual(out["data"]["keys_per_name"], 2) + self.assertEqual(out["data"]["partitions"], 5) def test_get(self) -> None: - out, task_buffer = self._call(["a_sub1"], method="GET") + out, task_buffer = self._call(["a.x_sub1"], method="GET") self.assertTrue(out["success"], out["message"]) - self.assertEqual(task_buffer.get_datasets_by_name.call_args[0][0], ["a_sub1"]) + self.assertEqual(task_buffer.get_datasets_by_name.call_args[0][0], ["a.x_sub1", "a:a.x_sub1"]) def test_refused(self) -> None: out, task_buffer = self._call([]) self.assertEqual((out["success"], out["data"]), (False, {"rejected": "no_keys"})) - out, task_buffer = self._call([f"a_sub{i}" for i in range(1001)]) - self.assertEqual((out["success"], out["data"]), (False, {"rejected": "too_many_keys", "keys": 1001, "max_keys": 1000})) + # the cap counts keys, two a name + out, task_buffer = self._call([f"a.x_sub{i}" for i in range(501)]) + self.assertEqual((out["success"], out["data"]), (False, {"rejected": "too_many_keys", "keys": 1002, "max_keys": 1000, "keys_per_name": 2})) task_buffer.get_datasets_by_name.assert_not_called() + out, task_buffer = self._call([f"a.x_sub{i}" for i in range(500)]) + self.assertEqual(len(task_buffer.get_datasets_by_name.call_args[0][0]), 1000) def test_refusal_of_the_lookup_passed_on(self) -> None: - refusal = {"rejected": "probe_budget", "keys": 201, "partitions": 5, "budget": 1000} - out, task_buffer = self._call([f"a_sub{i}" for i in range(201)], answer=(False, "over", refusal)) - self.assertEqual((out["success"], out["message"], out["data"]), (False, "over", refusal)) + refusal = {"rejected": "probe_budget", "keys": 402, "partitions": 5, "budget": 1000} + out, task_buffer = self._call([f"a.x_sub{i}" for i in range(201)], answer=(False, "over", dict(refusal))) + self.assertEqual((out["success"], out["message"], out["data"]), (False, "over", dict(refusal, keys_per_name=2))) + + def test_error_passed_on(self) -> None: + out, task_buffer = self._call(["a.x_sub1"], answer=(False, "database error ORA-00942", {"error": "ORA-00942"})) + self.assertEqual((out["success"], out["message"], out["data"]), (False, "database error ORA-00942", {"error": "ORA-00942"})) if __name__ == "__main__": From b0a75137652d61f80fdaa72fe8507aa3140f0b3e Mon Sep 17 00:00:00 2001 From: tmaeno Date: Wed, 28 Oct 2026 23:53:12 +0100 Subject: [PATCH 2/3] Skip the job message removal when the pilot queue is missing getJobs removes a push job's message from panda_pilot_queue after committing the job, but get_mb_proxy returns None when that queue is not configured or the message broker setup failed in this process, and the call went ahead without checking. The AttributeError reached the outer except after the commit, so getJobs rolled back nothing and returned no jobs, leaving every job committed so far in that call sent but handed to no pilot. The other two senders already check the proxy; this one now does the same and logs that the queue is not configured. --- pandaserver/taskbuffer/db_proxy_mods/job_complex_module.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/pandaserver/taskbuffer/db_proxy_mods/job_complex_module.py b/pandaserver/taskbuffer/db_proxy_mods/job_complex_module.py index 4a625d27d..6d879134b 100644 --- a/pandaserver/taskbuffer/db_proxy_mods/job_complex_module.py +++ b/pandaserver/taskbuffer/db_proxy_mods/job_complex_module.py @@ -2912,9 +2912,12 @@ def build_where_clause() -> tuple[str, dict[str, Any]]: tmp_log.error("recordStatusChange in getJobs") self.push_job_status_message(job, job.PandaID, job.jobStatus) # type: ignore[arg-type] # "NULL" sentinel, see spec_column.py if via_topic and job.is_push_job(): - tmp_log.debug("delete job message") mb_proxy_queue = self.get_mb_proxy("panda_pilot_queue") - srv_msg_utils.delete_job_message(mb_proxy_queue, job.PandaID) # type: ignore[arg-type] # "NULL" sentinel, see spec_column.py + if mb_proxy_queue: + tmp_log.debug("delete job message") + srv_msg_utils.delete_job_message(mb_proxy_queue, job.PandaID) # type: ignore[arg-type] # "NULL" sentinel, see spec_column.py + else: + tmp_log.debug("message queue not configured") return retJobs, nSent except Exception: self.dump_error_message(tmp_log) From 8ff900a70490c5cc8bc50c4d94cb68b67dbe4e07 Mon Sep 17 00:00:00 2001 From: tmaeno Date: Wed, 28 Oct 2026 23:56:10 +0100 Subject: [PATCH 3/3] Name the message bus senders and a few fields typed Any get_mb_proxy returns the MBSenderProxy that start_passive_mode put under the channel, or None, but was declared Any, as were the senders in srv_msg_utils that take its result and the workflow manager's own mb_proxy. They are now MBSenderProxy, imported under TYPE_CHECKING since msg_bkr_utils sets up its loggers and imports stomp on import. CI's mypy does not install panda-common, so there the name is still Any; with panda-common 0.1.11 visible the tree reports nothing new. Three fields whose class was only in a comment now carry it: DaemonMaster.tbif is a TaskBufferInterface or None, AdderGen.add_result an AdderResult or None, and cacheSchedConfig.queueData the list of dicts query_column_sql builds. queueData starts as an empty list rather than None; its only reader, main, calls getQueueData first. --- pandaserver/daemons/scripts/cache_schedconfig.py | 6 +++--- pandaserver/daemons/utils.py | 11 +++++++---- pandaserver/dataservice/adder_gen.py | 3 ++- pandaserver/srvcore/srv_msg_utils.py | 13 +++++++++---- pandaserver/taskbuffer/db_proxy_mods/base_module.py | 7 ++++--- pandaserver/workflow/workflow_core.py | 8 ++++++-- 6 files changed, 31 insertions(+), 17 deletions(-) diff --git a/pandaserver/daemons/scripts/cache_schedconfig.py b/pandaserver/daemons/scripts/cache_schedconfig.py index cfa4a7852..7175ffab6 100644 --- a/pandaserver/daemons/scripts/cache_schedconfig.py +++ b/pandaserver/daemons/scripts/cache_schedconfig.py @@ -35,7 +35,7 @@ class cacheSchedConfig: def __init__(self, tbuf: Any) -> None: self.tbuf = tbuf # the queue rows, read by getQueueData() before anything dumps them - self.queueData: Any = None + self.queueData: list[dict[str, Any]] = [] self.cloudStatus = None # Define this here, but could be more flexible... self.queueDataFields = { @@ -114,9 +114,9 @@ def __init__(self, tbuf: Any) -> None: "all": None, } - def query_column_sql(self, sql: str, varMap: dict[str, Any] | None = None, arraySize: int = 100) -> list[Any]: + def query_column_sql(self, sql: str, varMap: dict[str, Any] | None = None, arraySize: int = 100) -> list[dict[str, Any]]: res = self.tbuf.querySQL(sql, varMap, arraySize=arraySize) - retList = [] + retList: list[dict[str, Any]] = [] for ( panda_queue, data, diff --git a/pandaserver/daemons/utils.py b/pandaserver/daemons/utils.py index 4572cdd55..70491cab5 100644 --- a/pandaserver/daemons/utils.py +++ b/pandaserver/daemons/utils.py @@ -15,7 +15,7 @@ from collections.abc import Callable from multiprocessing.connection import Connection from types import FrameType -from typing import Any +from typing import TYPE_CHECKING, Any import psutil from pandacommon.pandalogger import logger_utils @@ -23,6 +23,10 @@ from pandaserver.config import daemon_config, panda_config +if TYPE_CHECKING: + # imported for annotations only; _make_tbif() imports it at runtime, where the order matters + from pandaserver.taskbuffer.TaskBufferInterface import TaskBufferInterface + # list of signals accepted to end the main process END_SIGNALS = [ signal.SIGINT, @@ -462,9 +466,8 @@ def __init__(self, logger: logging.Logger, n_workers: int = 1, n_dbconn: int = 1 self._make_dem_run_map() # map to store global states self.global_state_map: dict[str, Any] = {} - # shared taskBufferIF, built by _make_tbif() only when use_tbif is on. The class it - # holds is imported inside that method on purpose, so it cannot be named here - self.tbif: Any = None + # shared taskBufferIF, built by _make_tbif() only when use_tbif is on + self.tbif: "TaskBufferInterface | None" = None self._make_tbif() # shared lock pool self.lock_pool = LockPool() diff --git a/pandaserver/dataservice/adder_gen.py b/pandaserver/dataservice/adder_gen.py index 2763c1acd..ca070cec5 100644 --- a/pandaserver/dataservice/adder_gen.py +++ b/pandaserver/dataservice/adder_gen.py @@ -19,6 +19,7 @@ from pandaserver.brokerage.SiteMapper import SiteMapper from pandaserver.config import panda_config from pandaserver.dataservice import DataServiceUtils, closer +from pandaserver.dataservice.adder_result import AdderResult from pandaserver.srvcore.CoreUtils import normalize_cpu_model if TYPE_CHECKING: @@ -89,7 +90,7 @@ def __init__( self.lock_pool = lock_pool self.report_dict: dict[str, Any] = {} self.adder_plugin: Any = None - self.add_result: Any = None + self.add_result: AdderResult | None = None self.adder_plugin_class: Any = None # logger self.logger = LogWrapper(_logger, str(self.job_id)) diff --git a/pandaserver/srvcore/srv_msg_utils.py b/pandaserver/srvcore/srv_msg_utils.py index a27506862..15d97176b 100644 --- a/pandaserver/srvcore/srv_msg_utils.py +++ b/pandaserver/srvcore/srv_msg_utils.py @@ -1,8 +1,13 @@ import json -from typing import Any +from typing import TYPE_CHECKING, Any from pandacommon.pandautils.PandaUtils import naive_utcnow +if TYPE_CHECKING: + # imported for annotations only: msg_bkr_utils sets up its loggers and imports stomp + # when imported, and DBProxy.get_mb_proxy defers that until a daemon has opened its log + from pandacommon.pandamsgbkr.msg_bkr_utils import MBSenderProxy + # make message def make_message(msg_type: str, **kwargs: Any) -> str: @@ -13,7 +18,7 @@ def make_message(msg_type: str, **kwargs: Any) -> str: # send a job message -def send_job_message(msg_queue: Any, msg_topic: Any, task_id: int, job_id: int) -> None: +def send_job_message(msg_queue: "MBSenderProxy", msg_topic: "MBSenderProxy", task_id: int, job_id: int) -> None: # make message msg = make_message("get_job", taskid=task_id, jobid=job_id) # use job ID for selector @@ -25,7 +30,7 @@ def send_job_message(msg_queue: Any, msg_topic: Any, task_id: int, job_id: int) # delete a job message -def delete_job_message(msg_queue: Any, job_id: int, time_out: int = 10) -> None: +def delete_job_message(msg_queue: "MBSenderProxy", job_id: int, time_out: int = 10) -> None: # job ID for selector headers = {"selector": "type='{0}' OR JMSType='{0}'".format(job_id)} # subscribe to remove job messages @@ -35,7 +40,7 @@ def delete_job_message(msg_queue: Any, job_id: int, time_out: int = 10) -> None: # send a task message -def send_task_message(msg_topic: Any, command_str: str, task_id: int) -> None: +def send_task_message(msg_topic: "MBSenderProxy", command_str: str, task_id: int) -> None: # make message msg = make_message(f"{command_str}_task", taskid=task_id) # send message to topic diff --git a/pandaserver/taskbuffer/db_proxy_mods/base_module.py b/pandaserver/taskbuffer/db_proxy_mods/base_module.py index dd79a3d47..65ba1daa8 100644 --- a/pandaserver/taskbuffer/db_proxy_mods/base_module.py +++ b/pandaserver/taskbuffer/db_proxy_mods/base_module.py @@ -24,8 +24,9 @@ if TYPE_CHECKING: # imported for annotations only. WrappedCursor imports panda_config, so - # importing it at runtime here would close an import cycle, and msg_processor - # is the module get_mb_proxy imports late so that a daemon opens its log first + # importing it at runtime here would close an import cycle, and pandamsgbkr + # is the package get_mb_proxy imports late so that a daemon opens its log first + from pandacommon.pandamsgbkr.msg_bkr_utils import MBSenderProxy from pandacommon.pandamsgbkr.msg_processor import PassiveProxies from pandaserver.taskbuffer.JediTaskSpec import JediTaskSpec @@ -397,7 +398,7 @@ def getvalue_corrector(self, value: Any) -> Any: return value # get mb proxy - def get_mb_proxy(self, channel: str) -> Any: + def get_mb_proxy(self, channel: str) -> "MBSenderProxy | None": if self.mb_proxy_dict is None: try: if hasattr(panda_config, "mq_configFile") and panda_config.mq_configFile: diff --git a/pandaserver/workflow/workflow_core.py b/pandaserver/workflow/workflow_core.py index 2977c0e72..69a7ae36f 100644 --- a/pandaserver/workflow/workflow_core.py +++ b/pandaserver/workflow/workflow_core.py @@ -9,7 +9,7 @@ from collections.abc import Generator from contextlib import contextmanager from datetime import datetime -from typing import Any, Dict, List +from typing import TYPE_CHECKING, Any, Dict, List from pandacommon.pandalogger.LogWrapper import LogWrapper from pandacommon.pandalogger.PandaLogger import PandaLogger @@ -41,6 +41,10 @@ parse_raw_request, ) +if TYPE_CHECKING: + # imported for annotations only: set_mb_proxy() starts the message broker late + from pandacommon.pandamsgbkr.msg_bkr_utils import MBSenderProxy + # import polars as pl # isort:skip @@ -185,7 +189,7 @@ def __init__(self, task_buffer: Any, *args: Any, **kwargs: Any) -> None: self.ddm_if = rucioAPI self.full_pid = f"{socket.getfqdn().split('.')[0]}-{os.getpgrp()}-{os.getpid()}" self.plugin_map: dict[str, dict[str, Any]] = {} - self.mb_proxy: Any = None + self.mb_proxy: "MBSenderProxy | None" = None self.set_mb_proxy() def get_plugin(self, plugin_type: str, flavor: str) -> Any: