From 6d2feb062189b1760bed50f35badd8d73465e2d0 Mon Sep 17 00:00:00 2001 From: mightqxc Date: Tue, 28 Jul 2026 11:51:38 +0200 Subject: [PATCH 1/3] taskrefiner: add input size limit check for analysis tasks from tapes --- pandajedi/jediorder/TaskRefiner.py | 30 +++++++++++++++++++++++++++++- 1 file changed, 29 insertions(+), 1 deletion(-) diff --git a/pandajedi/jediorder/TaskRefiner.py b/pandajedi/jediorder/TaskRefiner.py index 6b17ebfbb..4cf907653 100644 --- a/pandajedi/jediorder/TaskRefiner.py +++ b/pandajedi/jediorder/TaskRefiner.py @@ -412,7 +412,35 @@ def runImpl(self): tmpLog.info("no need to prestage, try to resume task from staging") # no dataset needs pre-staging; resume task from staging self.taskBufferIF.sendCommandTaskPanda(jediTaskID, "TaskRefiner. No need to prestage. Resumed from staging", True, "resume") - if prestaging_list: + # check total size of input datasets from tapes for analysis tasks + if prestaging_list and taskParamMap.get("taskType") == "anal" and taskParamMap.get("prodSourceLabel") == "user": + analysis_tape_input_limit_TB = self.taskBufferIF.getConfigValue( + "taskrefiner", "USER_MAX_TAPE_INPUT_TB", "jedi", vo, default=300 + ) + if analysis_tape_input_limit_TB < 0: + # negative limit means unlimited + tmpLog.debug(f"input size limit from tapes is {analysis_tape_input_limit_TB} TB (unlimited) ; skipped checking") + else: + tape_input_size = 0 + for tmp_dataset, _, _, tmp_to_pin, _ in prestaging_list: + if tmp_to_pin: + # replicas already on datadisks, only to pin; not from tape + continue + tmp_metadata = rucioAPI.get_dataset_metadata(tmp_dataset) + if not tmp_metadata or tmp_metadata.get("bytes") is None: + tmpLog.warning(f"cannot get size of {tmp_dataset} ; skipped") + continue + tape_input_size += tmp_metadata["bytes"] + tape_input_size_TB = tape_input_size / 1024**4 + tmpLog.info(f"input data from tapes: {tape_input_size_TB:.3f} TB") + if tape_input_size_TB > analysis_tape_input_limit_TB: + errStr = ( + f"input data from tapes exceeds the limit for analysis tasks " + f"({tape_input_size_TB:.3f} TB > {analysis_tape_input_limit_TB} TB)" + ) + tmpLog.error(errStr) + tmpStat = Interaction.SC_FAILED + if prestaging_list and tmpStat == Interaction.SC_SUCCEEDED: # something to prestage if to_reuse_staging_ds_list := ds_list_dict["to_reuse_staging_ds_list"]: # update to_staging_datasets with datasets to reuse existing staging DDM rules (de facto already staging, still need to submit DC requests) From 897526f7d337f260379c507b78a3ad79262c88fc Mon Sep 17 00:00:00 2001 From: mightqxc Date: Tue, 28 Jul 2026 13:27:53 +0200 Subject: [PATCH 2/3] taskrefiner: change into size check for individual input datasset --- pandajedi/jediorder/TaskRefiner.py | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/pandajedi/jediorder/TaskRefiner.py b/pandajedi/jediorder/TaskRefiner.py index 4cf907653..48d228f5d 100644 --- a/pandajedi/jediorder/TaskRefiner.py +++ b/pandajedi/jediorder/TaskRefiner.py @@ -412,16 +412,15 @@ def runImpl(self): tmpLog.info("no need to prestage, try to resume task from staging") # no dataset needs pre-staging; resume task from staging self.taskBufferIF.sendCommandTaskPanda(jediTaskID, "TaskRefiner. No need to prestage. Resumed from staging", True, "resume") - # check total size of input datasets from tapes for analysis tasks + # check size of each input dataset from tape for analysis tasks if prestaging_list and taskParamMap.get("taskType") == "anal" and taskParamMap.get("prodSourceLabel") == "user": analysis_tape_input_limit_TB = self.taskBufferIF.getConfigValue( "taskrefiner", "USER_MAX_TAPE_INPUT_TB", "jedi", vo, default=300 ) if analysis_tape_input_limit_TB < 0: # negative limit means unlimited - tmpLog.debug(f"input size limit from tapes is {analysis_tape_input_limit_TB} TB (unlimited) ; skipped checking") + tmpLog.debug(f"input size limit from tape is {analysis_tape_input_limit_TB} TB (unlimited) ; skipped checking") else: - tape_input_size = 0 for tmp_dataset, _, _, tmp_to_pin, _ in prestaging_list: if tmp_to_pin: # replicas already on datadisks, only to pin; not from tape @@ -430,16 +429,17 @@ def runImpl(self): if not tmp_metadata or tmp_metadata.get("bytes") is None: tmpLog.warning(f"cannot get size of {tmp_dataset} ; skipped") continue - tape_input_size += tmp_metadata["bytes"] - tape_input_size_TB = tape_input_size / 1024**4 - tmpLog.info(f"input data from tapes: {tape_input_size_TB:.3f} TB") - if tape_input_size_TB > analysis_tape_input_limit_TB: - errStr = ( - f"input data from tapes exceeds the limit for analysis tasks " - f"({tape_input_size_TB:.3f} TB > {analysis_tape_input_limit_TB} TB)" - ) - tmpLog.error(errStr) - tmpStat = Interaction.SC_FAILED + tmp_dataset_size_TB = tmp_metadata["bytes"] / 1024**4 + tmpLog.debug(f"input dataset {tmp_dataset} from tape is {tmp_dataset_size_TB:.3f} TB") + if tmp_dataset_size_TB > analysis_tape_input_limit_TB: + errStr = ( + f"input dataset {tmp_dataset} from tape exceeds the limit for analysis tasks " + f"({tmp_dataset_size_TB:.3f} TB > {analysis_tape_input_limit_TB} TB). " + f"Please contact support" + ) + tmpLog.error(errStr) + tmpStat = Interaction.SC_FAILED + break if prestaging_list and tmpStat == Interaction.SC_SUCCEEDED: # something to prestage if to_reuse_staging_ds_list := ds_list_dict["to_reuse_staging_ds_list"]: From 1651b1119dc01e6c0bd49a310d2a8e8a8bd4d4c9 Mon Sep 17 00:00:00 2001 From: mightqxc Date: Tue, 28 Jul 2026 13:49:39 +0200 Subject: [PATCH 3/3] fix --- pandajedi/jediorder/TaskRefiner.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pandajedi/jediorder/TaskRefiner.py b/pandajedi/jediorder/TaskRefiner.py index 48d228f5d..63b7ddf75 100644 --- a/pandajedi/jediorder/TaskRefiner.py +++ b/pandajedi/jediorder/TaskRefiner.py @@ -425,7 +425,7 @@ def runImpl(self): if tmp_to_pin: # replicas already on datadisks, only to pin; not from tape continue - tmp_metadata = rucioAPI.get_dataset_metadata(tmp_dataset) + tmp_metadata = rucioAPI.get_dataset_metadata(tmp_dataset, ignore_missing=True) if not tmp_metadata or tmp_metadata.get("bytes") is None: tmpLog.warning(f"cannot get size of {tmp_dataset} ; skipped") continue