status script generation

mcpartools/generator.py

@@ -98,6 +98,8 @@ def valid(self):
 
 
 class Generator:
+    wspdir_name = 'workspace'
+
     def __init__(self, options):
         self.options = options
         self.mc_engine = EngineDiscover.get_mcengine(input_path=self.options.input_path,
@@ -146,6 +148,12 @@ def run(self):
         # make symlinks to external files found
         self.symlink_external_files()
 
+        # generate script merging info logs
+        self.generate_merge_logs_script()
+
+        # generate status script
+        self.generate_status_script()
+
         # store information about command line arguments, date, time, user and hostname into generatemc.log
         self.save_logs()
 
@@ -169,8 +177,7 @@ def generate_main_dir(self):
         file_logger.addHandler(logging.FileHandler(os.path.join(dir_path, "generatemc.log"), mode='w+'))
 
     def generate_workspace(self):
-        wspdir_name = 'workspace'
-        wspdir_path = os.path.join(self.main_dir, wspdir_name)
+        wspdir_path = os.path.join(self.main_dir, self.wspdir_name)
         logger.debug("Generated workspace directory path: " + wspdir_path)
         os.mkdir(wspdir_path)
         self.workspace_dir = wspdir_path
@@ -188,7 +195,7 @@ def generate_workspace(self):
 
             self.mc_engine.save_run_script(jobdir_path, jobid + 1)
 
-        self.scheduler.write_main_run_script(jobs_no=self.options.jobs_no, output_dir=self.workspace_dir)
+        self.scheduler.write_main_run_script(particle_no=self.options.particle_no, output_dir=self.workspace_dir)
         self.mc_engine.write_collect_script(self.main_dir)
 
     def generate_submit_script(self):
@@ -242,3 +249,13 @@ def save_logs(self):
         file_logger.info('Date and time: ' + time.strftime("%Y-%m-%d %H:%M:%S"))
         file_logger.info('username@hostname: ' + getpass.getuser() + '@' + socket.gethostname())
         file_logger.info('Current working directory: ' + os.getcwd())
+
+    def generate_merge_logs_script(self):
+        output_name = 'output'
+        wspdir_path = os.path.join(self.main_dir, self.wspdir_name)
+        collect_path = os.path.join(self.main_dir, output_name)
+        self.scheduler.write_merge_logs_script(wspdir_path, collect_path, self.main_dir)
+
+    def generate_status_script(self):
+        wspdir_path = os.path.join(self.main_dir, self.wspdir_name)
+        self.scheduler.write_status_script(self.main_dir, wspdir_path)

mcpartools/mcengine/data/collect.sh

@@ -3,13 +3,36 @@
 # Exit immediately if a simple command exits with a non-zero status.
 set -e
 
+START=$(date +%s)
+
 INPUT_WILDCARD={output_dir:s}/workspace/job_*/{wildcard:s}
 OUTPUT_DIRECTORY={output_dir:s}/output
+LOG_FILE=$OUTPUT_DIRECTORY/info.log
 
 # change working directory
 cd {output_dir:s}
 
 # make output folder
 mkdir -p $OUTPUT_DIRECTORY
 
-{collect_action:s}
+echo "###########################################################" > $LOG_FILE
+echo "################### COLLECT INFORMATION ###################" >> $LOG_FILE
+echo "###########################################################" >> $LOG_FILE
+echo "#" >> $LOG_FILE
+echo "# START           = `date +"%Y-%m-%dT%H:%M:%S"`" >> $LOG_FILE
+echo "# END             =                   -" >> $LOG_FILE
+echo "# TIME IN SECONDS =                   -" >> $LOG_FILE
+echo "# STATUS          =                   1" >> $LOG_FILE
+echo "#" >> $LOG_FILE
+
+{collect_action:s}
+COLLECT_STATUS=$?
+
+let "EXECUTION_TIME = $(date +%s) - $START"
+
+#    end time is in line number 6
+sed -i "6s/.*/# END             = `date +"%Y-%m-%dT%H:%M:%S"`/" $LOG_FILE
+#    collapsed time is in line number 7
+sed -i "7s/.*/# TIME IN SECONDS =`printf "%20d" $EXECUTION_TIME`/" $LOG_FILE
+#    status is in line number 8
+sed -i "8s/.*/# STATUS          =                   0/" $LOG_FILE

mcpartools/scheduler/base.py

@@ -25,6 +25,8 @@ def __init__(self, scheduler_options):
 
     submit_script = 'submit.sh'
     main_run_script = 'main_run.sh'
+    merge_logs_script = 'merge_logs.sh'
+    status_script = 'status.sh'
 
     def submit_script_body(self, jobs_no, main_dir, workspace_dir):
         from pkg_resources import resource_string
@@ -43,14 +45,26 @@ def submit_script_body(self, jobs_no, main_dir, workspace_dir):
                                          main_dir=main_dir,
                                          collect_script_name='collect.sh')
 
-    def main_run_script_body(self, jobs_no, workspace_dir):
+    def main_run_script_body(self, particle_no, workspace_dir):
         from pkg_resources import resource_string
         tpl = resource_string(__name__, self.main_run_script_template)
         self.main_run_script = tpl.decode('ascii').format(options_header=self.options_header,
                                                           workspace_dir=workspace_dir,
-                                                          jobs_no=jobs_no)
+                                                          particle_no=particle_no)
         return self.main_run_script
 
+    def merge_logs_body(self, workspace_dir, collect_dir, main_dir):
+        from pkg_resources import resource_string
+        tpl = resource_string(__name__, self.merge_logs_script_template)
+        return tpl.decode("ascii").format(workspace_dir=workspace_dir,
+                                          collect_dir=collect_dir,
+                                          main_dir=main_dir)
+
+    def status_body(self, merge_script_path):
+        from pkg_resources import resource_string
+        tpl = resource_string(__name__, self.status_script_template)
+        return tpl.decode("ascii").format(merge_script_path=merge_script_path)
+
     def write_submit_script(self, main_dir, script_basename, jobs_no, workspace_dir):
         script_path = os.path.join(main_dir, script_basename)
         fd = open(script_path, 'w')
@@ -63,12 +77,38 @@ def write_submit_script(self, main_dir, script_basename, jobs_no, workspace_dir)
         logger.debug("Jobs no " + str(jobs_no))
         logger.debug("Workspace " + abs_path_workspace)
 
-    def write_main_run_script(self, jobs_no, output_dir):
+    def write_main_run_script(self, particle_no, output_dir):
         output_dir_abspath = os.path.abspath(output_dir)
         out_file_path = os.path.join(output_dir_abspath, self.main_run_script)
         fd = open(out_file_path, 'w')
-        fd.write(self.main_run_script_body(jobs_no=jobs_no, workspace_dir=output_dir_abspath))
+        fd.write(self.main_run_script_body(particle_no=particle_no, workspace_dir=output_dir_abspath))
         fd.close()
         os.chmod(out_file_path, 0o750)
         logger.debug("Saved main run script: " + out_file_path)
         logger.debug("Output dir " + output_dir)
+
+    def write_merge_logs_script(self, workspace_dir, collect_dir, main_dir):
+        workspace_dir_abspath = os.path.abspath(workspace_dir)
+        collect_dir_abspath = os.path.abspath(collect_dir)
+        main_dir_abspath = os.path.abspath(main_dir)
+
+        out_file_path = os.path.join(workspace_dir_abspath, self.merge_logs_script)
+
+        fd = open(out_file_path, 'w')
+        fd.write(self.merge_logs_body(workspace_dir_abspath, collect_dir_abspath, main_dir_abspath))
+        fd.close()
+        os.chmod(out_file_path, 0o750)
+        logger.debug("Saved merge logs script: " + out_file_path)
+
+    def write_status_script(self, main_dir, workspace_dir):
+        main_dir_abspath = os.path.abspath(main_dir)
+        out_file_path = os.path.join(main_dir_abspath, self.status_script)
+
+        workspace_dir_abspath = os.path.abspath(workspace_dir)
+        merge_log_script = os.path.join(workspace_dir_abspath, self.merge_logs_script)
+
+        fd = open(out_file_path, 'w')
+        fd.write(self.status_body(merge_log_script))
+        fd.close()
+        os.chmod(out_file_path, 0o750)
+        logger.debug("Saved status script: " + out_file_path)