Should work on new slurm sluster at geneva now

Author: maxnoe

erna/automatic_processing/__main__.py

@@ -28,7 +28,7 @@ def main(config, verbose):
         logging.getLogger('erna').setLevel(logging.DEBUG)
 
     stream_handler = logging.StreamHandler()
-    file_handler = logging.FileHandler(config['submitter']['logfile'])
+    file_handler = logging.FileHandler(config['submitter'].pop('logfile'))
     formatter = logging.Formatter(
         '%(asctime)s|%(levelname)s|%(name)s|%(message)s'
     )
@@ -44,16 +44,7 @@ def main(config, verbose):
     database.close()
 
     job_monitor = JobMonitor(port=config['submitter']['port'])
-    job_submitter = JobSubmitter(
-        interval=config['submitter']['interval'],
-        max_queued_jobs=config['submitter']['max_queued_jobs'],
-        data_directory=config['submitter']['data_directory'],
-        host=config['submitter']['host'],
-        port=config['submitter']['port'],
-        group=config['submitter']['group'],
-        mail_address=config['submitter']['mail_address'],
-        mail_settings=config['submitter']['mail_settings'],
-    )
+    job_submitter = JobSubmitter(**config['submitter'])
 
     log.info('Starting main loop')
     try:
@@ -75,6 +66,6 @@ def main(config, verbose):
         inserted = ProcessingState.get(description='inserted')
 
         for job in Job.select().where((Job.status == running) | (Job.status == queued)):
-            sp.run(['qdel', 'erna_{}'.format(job.id)])
+            sp.run(['scancel', '--jobname=erna_{}'.format(job.id)])
             job.status = inserted
             job.save()

erna/automatic_processing/database.py

@@ -14,7 +14,7 @@
 
 __all__ = [
     'RawDataFile', 'DrsFile',
-    'Jar', 'XML', 'Job', 'Queue',
+    'Jar', 'XML', 'Job',
     'ProcessingState',
     'database', 'setup_database',
 ]
@@ -31,12 +31,6 @@
     'walltime_exceeded',
 ]
 
-WALLTIMES = {
-    'fact_short': 60 * 60,
-    'fact_medium': 6 * 60 * 60,
-    'fact_long': 7 * 24 * 60 * 60,
-}
-
 
 class RetryMySQLDatabase(RetryOperationalError, MySQLDatabase):
     ''' Automatically reconnect when connection went down'''
@@ -71,9 +65,6 @@ def setup_database(database, drop=False):
     for description in PROCESSING_STATES:
         ProcessingState.get_or_create(description=description)
 
-    for name, walltime in WALLTIMES.items():
-        Queue.get_or_create(name=name, walltime=walltime)
-
 
 class File(Model):
     night = NightField()
@@ -85,9 +76,9 @@ class Meta:
         database = database
         indexes = ((('night', 'run_id'), True), )  # unique index
 
-    def get_path(self):
+    def get_path(self, basepath='/fact/raw'):
         return os.path.join(
-            '/fact/raw',
+            basepath,
             str(self.night.year),
             '{:02d}'.format(self.night.month),
             '{:02d}'.format(self.night.day),
@@ -166,25 +157,16 @@ def __repr__(self):
         return '{}'.format(self.description)
 
 
-class Queue(Model):
-    name = CharField(unique=True)
-    walltime = IntegerField()
-
-    class Meta:
-        database = database
-        db_table = 'queues'
-
-
 class Job(Model):
     raw_data_file = ForeignKeyField(RawDataFile, related_name='raw_data_file')
     drs_file = ForeignKeyField(DrsFile, related_name='drs_file')
     jar = ForeignKeyField(Jar, related_name='jar')
     result_file = CharField(null=True)
     status = ForeignKeyField(ProcessingState, related_name='status')
     priority = IntegerField(default=5)
+    walltime = IntegerField(default=60)
     xml = ForeignKeyField(XML)
     md5hash = FixedCharField(32, null=True)
-    queue = ForeignKeyField(Queue, related_name='queue')
 
     class Meta:
         database = database
@@ -193,7 +175,8 @@ class Meta:
             (('raw_data_file', 'jar', 'xml'), True),  # unique constraint
         )
 
-MODELS = [RawDataFile, DrsFile, Jar, XML, Job, ProcessingState, Queue]
+
+MODELS = [RawDataFile, DrsFile, Jar, XML, Job, ProcessingState]
 
 
 @wrapt.decorator

erna/automatic_processing/database_utils.py

@@ -99,7 +99,6 @@ def find_drs_file(raw_data_file, closest=True):
     '''
     query = DrsFile.select()
     query = query.where(DrsFile.night == raw_data_file.night)
-    query = query.where(DrsFile.available)
 
     if raw_data_file.roi == 300:
         query = query.where((DrsFile.drs_step == 2) & (DrsFile.roi == 300))
@@ -129,7 +128,7 @@ def insert_new_job(
         raw_data_file,
         jar,
         xml,
-        queue,
+        walltime,
         priority=5,
         closest_drs_file=True,
         ):
@@ -144,8 +143,8 @@ def insert_new_job(
         the fact-tools jar to use
     xml: XML
         the xml to use
-    queue: Queue
-        the queue to use
+    walltime: walltime
+        the walltime to use
     priority: int
         Priority for the Job. Lower numbers mean more important.
     closest_drs_file: bool
@@ -169,7 +168,7 @@ def insert_new_job(
         raw_data_file=raw_data_file,
         drs_file=drs_file,
         jar=jar,
-        queue=queue,
+        walltime=walltime,
         status=ProcessingState.get(description='inserted'),
         priority=priority,
         xml=xml,
@@ -179,7 +178,7 @@ def insert_new_job(
 
 
 @requires_database_connection
-def insert_new_jobs(raw_data_files, jar, xml, queue, progress=True, **kwargs):
+def insert_new_jobs(raw_data_files, jar, xml, walltime, progress=True, **kwargs):
 
     if isinstance(raw_data_files, list):
         total = len(raw_data_files)
@@ -189,7 +188,7 @@ def insert_new_jobs(raw_data_files, jar, xml, queue, progress=True, **kwargs):
     failed_files = []
     for f in tqdm(raw_data_files, total=total, disable=not progress):
         try:
-            insert_new_job(f, jar=jar, xml=xml, queue=queue, **kwargs)
+            insert_new_job(f, jar=jar, xml=xml, walltime=walltime, **kwargs)
         except peewee.IntegrityError:
             log.warning('Job already submitted: {}_{:03d}'.format(f.night, f.run_id))
         except ValueError as e:
@@ -281,17 +280,12 @@ def build_output_base_name(job):
 
 
 @requires_database_connection
-def resubmit_walltime_exceeded(old_queue, new_queue):
+def resubmit_walltime_exceeded(factor=1.5):
     '''
     Resubmit jobs where walltime was exceeded.
-    Change queue from old_queue to new_queue
     '''
-    if old_queue.walltime >= new_queue.walltime:
-        raise ValueError('New queue must have longer walltime for this to make sense')
-
     return (
         Job
-        .update(queue=new_queue, status=ProcessingState.get(description='inserted'))
+        .update(walltime=factor * Job.walltime, status=ProcessingState.get(description='inserted'))
         .where(Job.status == ProcessingState.get(description='walltime_exceeded'))
-        .where(Job.queue == old_queue)
     ).execute()

erna/automatic_processing/job_submitter.py

@@ -5,7 +5,7 @@
 
 from .database import ProcessingState, requires_database_connection
 from .database_utils import count_jobs, get_pending_jobs
-from .qsub import submit_job, get_current_jobs
+from .slurm import submit_job, get_current_jobs
 
 log = logging.getLogger(__name__)
 
@@ -16,7 +16,10 @@ def __init__(
         self,
         interval,
         max_queued_jobs,
-        data_directory,
+        raw_dir,
+        aux_dir,
+        erna_dir,
+	script,
         host,
         port,
         group,
@@ -32,7 +35,7 @@ def __init__(
             Maximum number of jobs in the queue of the grid engine
             No new jobs are submitted if the number of jobs in the queue is
             higher than this value
-        data_directory: str
+        erna_directory: str
             patch to the basic structure for erna. Logfiles, jars, xmls and
             analysis output are stored in subdirectories to this directory.
         host: str
@@ -48,12 +51,15 @@ def __init__(
         self.event = Event()
         self.interval = interval
         self.max_queued_jobs = max_queued_jobs
-        self.data_directory = data_directory
+        self.erna_dir = erna_dir
+        self.aux_dir = aux_dir
+        self.raw_dir = raw_dir
         self.host = host
         self.port = port
         self.group = group
         self.mail_settings = mail_settings
         self.mail_address = mail_address
+        self.script = script
 
     def run(self):
         while not self.event.is_set():
@@ -62,7 +68,7 @@ def run(self):
             except peewee.OperationalError:
                 log.warning('Lost database connection')
             except Exception as e:
-                log.exception('Error during submission')
+                log.exception('Error during submission: {}'.format(e))
             self.event.wait(self.interval)
 
     def terminate(self):
@@ -92,8 +98,11 @@ def process_pending_jobs(self):
                 try:
                     submit_job(
                         job,
-                        output_base_dir=os.path.join(self.data_directory, 'fact-tools'),
-                        data_dir=self.data_directory,
+			script=self.script,
+                        output_base_dir=os.path.join(self.erna_dir, 'fact-tools'),
+                        raw_dir=self.raw_dir,
+                        aux_dir=self.aux_dir,
+                        erna_dir=self.erna_dir,
                         mail_address=self.mail_address,
                         mail_settings=self.mail_settings,
                         submitter_host=self.host,
@@ -103,5 +112,5 @@ def process_pending_jobs(self):
                     log.info('New job with id {} queued'.format(job.id))
                 except:
                     log.exception('Could not submit job')
-                    job.status = ProcessingState.get(description='error')
+                    job.status = ProcessingState.get(description='failed')
                     job.save()