Merge pull request #394 from HSF/dev

fix core to processing type map

Author: wguanicedew

doma/lib/idds/doma/workflowv2/domapandawork.py

@@ -845,7 +845,7 @@ def submit_panda_task(self, processing):
             if self.additional_task_parameters_per_site:
                 try:
                     for site in self.additional_task_parameters_per_site:
-                        if (task_param['PandaSite'] and site in task_param['PandaSite']) or (task_param['site'] and site in task_param['site']):
+                        if ('PandaSite' in task_param and task_param['PandaSite'] and site in task_param['PandaSite']) or ('site' in task_param and task_param['site'] and site in task_param['site']):
                             for key, value in self.additional_task_parameters_per_site[site].items():
                                 if key not in task_param:
                                     task_param[key] = value
@@ -857,7 +857,7 @@ def submit_panda_task(self, processing):
                     # core_to_queues = {"1": {"queues": ["Rubin", "Rubin_Extra_Himem"], "processing_type": ""},
                     #                   "Rubin_Multi": {"queues": ["Rubin_Multi"], "processing_type": "Rubin_Multi"},
                     #                   "Rubin_Merge": {"queues": ["Rubin_Merge"], "processing_type": "Rubin_Merge"},
-                    #                   "any": "Rubin_Multi"}
+                    #                   "any": {"queues": ["Rubin_Multi"], "processing_type": "Rubin_Multi"}}
 
                     if task_param['processingType']:
                         msg = f"processingType {task_param['processingType']} is already set, do nothing"
@@ -867,8 +867,7 @@ def submit_panda_task(self, processing):
                         queue_processing_type = {}
                         for k in self.core_to_queues:
                             key = str(k)
-                            if not key.isdigit():
-                                num_cores.append(key)
+                            num_cores.append(key)
                             if key not in ['any']:
                                 queues = self.core_to_queues[k].get('queues', [])
                                 processing_type = self.core_to_queues[k].get('processing_type', '')
@@ -881,6 +880,14 @@ def submit_panda_task(self, processing):
                                 msg = f"processingType is not defined, set it to {p_type} based on coreCount {task_param['coreCount']}"
                                 task_param['processingType'] = p_type
                                 self.logger.warn(msg)
+                            if 'site' in task_param and task_param['site']:
+                                for q in queue_processing_type:
+                                    if task_param['site'] in q or q in task_param['site']:
+                                        p_type = queue_processing_type[q]
+                                        if p_type:
+                                            msg = f"processingType is not defined, set it to {p_type} based on site {task_param['site']}"
+                                            task_param['processingType'] = p_type
+                                            self.logger.debug(msg)
                         else:
                             if 'site' in task_param and task_param['site']:
                                 for q in queue_processing_type:

main/lib/idds/agents/transformer/transformer.py

@@ -1175,7 +1175,11 @@ def process_update_transform(self, event):
                         ret, is_terminated, ret_processing_id = self.handle_update_transform(tf, event)
                     new_pr_ids, update_pr_ids = self.update_transform(ret)
 
-                    if is_terminated or (event._content and 'event' in event._content and event._content['event'] == 'submitted'):
+                    has_update_workload_id = False
+                    new_workload_id = ret.get('transform_parameters', {}).get('workload_id', None)
+                    if new_workload_id and tf['workload_id'] != new_workload_id:
+                        has_update_workload_id = True
+                    if has_update_workload_id or is_terminated or (event._content and 'event' in event._content and event._content['event'] == 'submitted'):
                         self.logger.info(log_pre + "UpdateRequestEvent(request_id: %s)" % tf['request_id'])
                         event = UpdateRequestEvent(publisher_id=self.id, request_id=tf['request_id'], content=event._content)
                         self.event_bus.send(event)

main/lib/idds/tests/panda_test.py

@@ -14,8 +14,8 @@
 # os.environ['PANDA_URL_SSL'] = 'https://panda-doma-k8s-panda.cern.ch/server/panda'
 # os.environ['PANDA_URL'] = 'http://panda-doma-k8s-panda.cern.ch:25080/server/panda'
 
-os.environ['PANDA_URL'] = 'https://usdf-panda-server.slac.stanford.edu:8443/server/panda'
-os.environ['PANDA_URL_SSL'] = 'https://usdf-panda-server.slac.stanford.edu:8443/server/panda'
+# os.environ['PANDA_URL'] = 'https://usdf-panda-server.slac.stanford.edu:8443/server/panda'
+# os.environ['PANDA_URL_SSL'] = 'https://usdf-panda-server.slac.stanford.edu:8443/server/panda'
 
 # os.environ['PANDA_URL_SSL'] = 'https://pandaserver01.sdcc.bnl.gov:25443/server/panda'
 # os.environ['PANDA_URL'] = 'https://pandaserver01.sdcc.bnl.gov:25443/server/panda'
@@ -87,6 +87,9 @@
 task_ids = [124, 619]
 task_ids = [22707, 22708, 22709, 22710, 23211, 23212, 22155, 22158]
 task_ids = [24483, 24484, 25895, 26126, 26450, 26451, 26452, 26454, 26994, 27025, 27029]
+task_ids = [161489, 161496, 161502, 161508, 161514, 161520, 161526, 161532, 161538, 161544, 161550, 161556, 161562]
+task_ids = [3174, 3198, 3209, 3230, 3252, 3266, 3284, 3292, 3300, 3312, 3350, 3379, 3387]
+task_ids = [1548, 1555]
 for task_id in task_ids:
     print("Killing %s" % task_id)
     ret = Client.killTask(task_id, verbose=True)