Merge pull request HSF#89 from HSF/flin

queue config mapper: refill if missing 'ANY' types

pandaharvester/commit_timestamp.py

@@ -1 +1 @@
-timestamp = "25-11-2020 13:34:10 on master (by tmaeno)"
+timestamp = "09-12-2020 17:58:33 on flin (by fahui)"

pandaharvester/harvestercore/db_proxy.py

@@ -840,11 +840,11 @@ def update_worker(self, workspec, criteria=None):
             return None
 
     # fill panda queue table
-    def fill_panda_queue_table(self, panda_queue_list, queue_config_mapper):
+    def fill_panda_queue_table(self, panda_queue_list, queue_config_mapper, refill_table=False):
         try:
             # get logger
             tmpLog = core_utils.make_logger(_logger, method_name='fill_panda_queue_table')
-            tmpLog.debug('start')
+            tmpLog.debug('start, refill={0}'.format(refill_table))
             # get existing queues
             sqlE = "SELECT queueName FROM {0} ".format(pandaQueueTableName)
             varMap = dict()
@@ -867,11 +867,20 @@ def fill_panda_queue_table(self, panda_queue_list, queue_config_mapper):
                     # check if already exist
                     sqlC = "SELECT * FROM {0} ".format(pandaQueueTableName)
                     sqlC += "WHERE queueName=:queueName "
+                    sqlC += " AND resourceType=:resourceType AND jobType=:jobType "
                     varMap = dict()
                     varMap[':queueName'] = queueName
+                    varMap[':resourceType'] = PandaQueueSpec.RT_catchall
+                    varMap[':jobType'] = PandaQueueSpec.JT_catchall
                     self.execute(sqlC, varMap)
                     resC = self.cur.fetchone()
-                    if resC is not None:
+                    if refill_table:
+                        sqlD = "DELETE FROM {0} ".format(pandaQueueTableName)
+                        sqlD += "WHERE queueName=:queueName "
+                        varMap = dict()
+                        varMap[':queueName'] = queueName
+                        self.execute(sqlD, varMap)
+                    if resC is not None and not refill_table:
                         # update limits just in case
                         varMap = dict()
                         sqlU = "UPDATE {0} SET ".format(pandaQueueTableName)

pandaharvester/harvestercore/queue_config_mapper.py

@@ -270,7 +270,7 @@ def _get_last_reload_time(self):
         return timestamp
 
     # load data
-    def load_data(self):
+    def load_data(self, refill_table=False):
         mainLog = _make_logger(method_name='QueueConfigMapper.load_data')
         with self.lock:
             # check if to update
@@ -608,7 +608,7 @@ def load_data(self):
             self.lastUpdate = datetime.datetime.utcnow()
         # update database
         if self.toUpdateDB:
-            self.dbProxy.fill_panda_queue_table(self.activeQueues.keys(), self)
+            self.dbProxy.fill_panda_queue_table(self.activeQueues.keys(), self, refill_table=refill_table)
             mainLog.debug('updated to DB')
         # done
         mainLog.debug('done')

pandaharvester/harvesterscripts/harvester_admin.py

@@ -198,7 +198,7 @@ def qconf_refresh(arguments):
     qcm = QueueConfigMapper()
     qcm._update_last_reload_time()
     qcm.lastUpdate = None
-    qcm.load_data()
+    qcm.load_data(refill_table=arguments.refill)
 
 def qconf_dump(arguments):
     from pandaharvester.harvesterscripts import queue_config_tool
@@ -327,6 +327,7 @@ def main():
     # qconf refresh command
     qconf_refresh_parser = qconf_subparsers.add_parser('refresh', help='refresh queue configuration immediately')
     qconf_refresh_parser.set_defaults(which='qconf_refresh')
+    qconf_refresh_parser.add_argument('-R', '--refill', dest='refill', action='store_true', help='Refill pq_table before refresh (cleaner)')
     # qconf purge command
     qconf_purge_parser = qconf_subparsers.add_parser('purge', help='Purge the queue thoroughly from harvester DB (Be careful !!)')
     qconf_purge_parser.set_defaults(which='qconf_purge')

pandaharvester/harvestersubmitter/htcondor_submitter.py

@@ -227,14 +227,18 @@ def _get_complicated_pilot_options(pilot_type, pilot_url=None):
     return pilot_opt_dict
 
 
-# get special flag of pilot wrapper to run with python 3 if pilot version is "3"
+# get special flag of pilot wrapper about pilot version , and whehter to run with python 3 if pilot version is "3"
 # FIXME: during pilot testing phase, only prodsourcelabel ptest and rc_test2 should run python3
 # This constraint will be removed when pilot is ready
-def _get_pilot_python_option(pilot_version, prod_source_label):
+def _get_pilot_version_python_option(pilot_version, prod_source_label):
+    version = 'current'
     option = ''
-    if pilot_version in ['3'] and prod_source_label in ['rc_test2', 'ptest']:
-        option = '-3'
-    return option
+    if pilot_version.startswith('3'):
+        if prod_source_label in ['rc_test2', 'ptest']:
+            option = '--pythonversion 3'
+    else:
+        version = pilot_version
+    return version, option
 
 
 # submit a bag of workers
@@ -432,8 +436,8 @@ def make_a_jdl(workspec, template, n_core_per_node, log_dir, panda_queue_name, e
             'ioIntensity': io_intensity,
             'pilotType': pilot_type_opt,
             'pilotUrlOption': pilot_url_str,
-            'pilotVersion': pilot_version,
-            'pilotPythonOption': _get_pilot_python_option(pilot_version, prod_source_label),
+            'pilotVersion': _get_pilot_version_python_option(pilot_version, prod_source_label)[0],
+            'pilotPythonOption': _get_pilot_version_python_option(pilot_version, prod_source_label)[1],
             'submissionHost': workspec.submissionHost,
             'submissionHostShort': workspec.submissionHost.split('.')[0],
         }