Prioritize ALRB by considering prodsourcelable for worker_adjuster, submitter, worker_maker

pandaharvester/commit_timestamp.py

@@ -1 +1 @@
-timestamp = "13-02-2026 07:45:33 on flin (by mightqxc)"
+timestamp = "07-04-2026 12:44:11 on flin (by mightqxc)"

pandaharvester/harvesterbody/job_fetcher.py

@@ -67,13 +67,13 @@ def run(self):
                 if n_jobs == 0:
                     tmpLog.debug("no job to fetch; skip")
                     continue
-                # prodsourcelabel
+                # prod_source_label
                 try:
                     is_grandly_unified_queue = pandaQueueDict.is_grandly_unified_queue(siteName)
                 except Exception:
                     is_grandly_unified_queue = False
                 default_prodSourceLabel = queueConfig.get_source_label(is_gu=is_grandly_unified_queue)
-                # randomize prodsourcelabel if configured
+                # randomize prod_source_label if configured
                 pdpm = getattr(queueConfig, "prodSourceLabelRandomWeightsPermille", {})
                 choice_list = core_utils.make_choice_list(pdpm=pdpm, default=default_prodSourceLabel)
                 prodSourceLabel = random.choice(choice_list)

pandaharvester/harvesterbody/worker_maker.py

@@ -18,8 +18,10 @@ def get_plugin(self, queue_config):
         return self.pluginFactory.get_plugin(queue_config.workerMaker)
 
     # make workers
-    def make_workers(self, jobchunk_list, queue_config, n_ready, job_type, resource_type, maker=None):
-        tmpLog = core_utils.make_logger(_logger, f"queue={queue_config.queueName} jtype={job_type} rtype={resource_type}", method_name="make_workers")
+    def make_workers(self, jobchunk_list, queue_config, n_ready, job_type, resource_type, prod_source_label="ANY", maker=None):
+        tmpLog = core_utils.make_logger(
+            _logger, f"queue={queue_config.queueName} jtype={job_type} rtype={resource_type} pslabel={prod_source_label}", method_name="make_workers"
+        )
         tmpLog.debug("start")
         try:
             # get plugin
@@ -37,7 +39,7 @@ def make_workers(self, jobchunk_list, queue_config, n_ready, job_type, resource_
             for iChunk, jobChunk in enumerate(jobchunk_list):
                 # make a worker
                 if iChunk >= n_ready:
-                    workSpec = maker.make_worker(jobChunk, queue_config, job_type, resource_type)
+                    workSpec = maker.make_worker(jobChunk, queue_config, job_type, resource_type, prod_source_label=prod_source_label)
                 else:
                     # use ready worker
                     if iChunk < len(readyWorkers):
@@ -63,35 +65,35 @@ def make_workers(self, jobchunk_list, queue_config, n_ready, job_type, resource_
             return [], jobchunk_list
 
     # get number of jobs per worker
-    def get_num_jobs_per_worker(self, queue_config, n_workers, job_type, resource_type, maker=None):
+    def get_num_jobs_per_worker(self, queue_config, n_workers, job_type, resource_type, prod_source_label="ANY", maker=None):
         # get plugin
         if maker is None:
             maker = self.pluginFactory.get_plugin(queue_config.workerMaker)
         return maker.get_num_jobs_per_worker(n_workers)
 
     # get number of workers per job
-    def get_num_workers_per_job(self, queue_config, n_workers, job_type, resource_type, maker=None):
+    def get_num_workers_per_job(self, queue_config, n_workers, job_type, resource_type, prod_source_label="ANY", maker=None):
         # get plugin
         if maker is None:
             maker = self.pluginFactory.get_plugin(queue_config.workerMaker)
         return maker.get_num_workers_per_job(n_workers)
 
     # check number of ready resources
-    def num_ready_resources(self, queue_config, job_type, resource_type, maker=None):
+    def num_ready_resources(self, queue_config, job_type, resource_type, prod_source_label="ANY", maker=None):
         # get plugin
         if maker is None:
             maker = self.pluginFactory.get_plugin(queue_config.workerMaker)
         return maker.num_ready_resources()
 
     # get upper limit on the cumulative total of workers per job
-    def get_max_workers_per_job_in_total(self, queue_config, job_type, resource_type, maker=None):
+    def get_max_workers_per_job_in_total(self, queue_config, job_type, resource_type, prod_source_label="ANY", maker=None):
         # get plugin
         if maker is None:
             maker = self.pluginFactory.get_plugin(queue_config.workerMaker)
         return maker.get_max_workers_per_job_in_total()
 
     # get upper limit on the number of new workers per job in a cycle
-    def get_max_workers_per_job_per_cycle(self, queue_config, job_type, resource_type, maker=None):
+    def get_max_workers_per_job_per_cycle(self, queue_config, job_type, resource_type, prod_source_label="ANY", maker=None):
         # get plugin
         if maker is None:
             maker = self.pluginFactory.get_plugin(queue_config.workerMaker)

pandaharvester/harvestercommunicator/panda_communicator.py

@@ -744,6 +744,31 @@ def get_job_stats(self):
 
         return stats, "OK"
 
+    # get job statistics: new function with prod_source_label, under testing and may replace the old one
+    def get_job_stats_new(self):
+        tmp_log = self.make_logger(method_name="get_job_stats_new")
+        tmp_log.debug("Start")
+
+        tmp_status, tmp_response = self.request_ssl("GET", "statistics/active_job_detailed_stats_by_site", {})
+        stats = {}
+        ret_message = "FAILED"
+
+        # Communication issue
+        if tmp_status is False:
+            core_utils.dump_error_message(tmp_log, tmp_response)
+            return stats, ret_message
+
+        tmp_success = tmp_response.get("success", False)
+        tmp_message = tmp_response.get("message")
+        stats = tmp_response.get("data")
+
+        if not tmp_success:
+            ret_message = tmp_message
+            core_utils.dump_error_message(tmp_log, ret_message)
+            return stats, ret_message
+
+        return stats, "OK"
+
     # update workers
     def update_workers(self, workspec_list):
         tmp_log = self.make_logger(method_name="update_workers")

pandaharvester/harvestercore/core_utils.py

@@ -706,3 +706,39 @@ def naive_utcfromtimestamp(timestamp: float) -> datetime:
         datetime: current UTC date and time, without tzinfo
     """
     return aware_utcfromtimestamp(timestamp).replace(tzinfo=None)
+
+
+def special_pilot_type_to_prod_source_label(pilot_type: str) -> str:
+    """
+    Convert special pilotType of worker to prodSourceLabel of PanDA job.
+
+    Args:
+        pilot_type (str): pilotType of worker, e.g. "RC", "ALRB", "PT" (except "PR" which is production)
+
+    Returns:
+        str: prodSourceLabel of PanDA job, e.g. "rc_test2", "rc_alrb", "ptest"; "ANY" if no mapping is defined for the given pilot_type
+    """
+    pilot_type_to_prod_source_label_map = {
+        "RC": "rc_test2",
+        "ALRB": "rc_alrb",
+        "PT": "ptest",
+    }
+    return pilot_type_to_prod_source_label_map.get(pilot_type, "ANY")
+
+
+def prod_source_label_to_pilot_type(prod_source_label: str) -> str:
+    """
+    Convert prodSourceLabel of PanDA job to pilotType of worker.
+
+    Args:
+        prod_source_label (str): prodSourceLabel of PanDA job, e.g. "rc_test2", "rc_alrb", "ptest"
+
+    Returns:
+        str: pilotType of worker, e.g. "RC", "ALRB", "PT"; default to "PR" (production) if no mapping is defined for the given prod_source_label
+    """
+    prod_source_label_to_pilot_type_map = {
+        "rc_test2": "RC",
+        "rc_alrb": "ALRB",
+        "ptest": "PT",
+    }
+    return prod_source_label_to_pilot_type_map.get(prod_source_label, "PR")