feat: custom log file behaviour

snakemake_executor_plugin_slurm/__init__.py

@@ -35,8 +35,8 @@ class ExecutorSettings(ExecutorSettingsBase):
         default=40,
         metadata={
             "help": """
-                    Defines the time in seconds before the first status
-                    check is performed after job submission.
+                    Defines the time in seconds before the first
+                    status check is performed after job submission.
                     """,
             "env_var": False,
             "required": False,
@@ -47,7 +47,8 @@ class ExecutorSettings(ExecutorSettingsBase):
         metadata={
             "help": """
                     Allow requeuing preempted of failed jobs,
-                    if no cluster default. Results in `sbatch ... --requeue ...`
+                    if no cluster default. Results in
+                    `sbatch ... --requeue ...`.
                     This flag has no effect, if not set.
                     """,
             "env_var": False,
@@ -62,7 +63,8 @@ class ExecutorSettings(ExecutorSettingsBase):
     # define whether your executor plugin executes locally
     # or remotely. In virtually all cases, it will be remote execution
     # (cluster, cloud, etc.). Only Snakemake's standard execution
-    # plugins (snakemake-executor-plugin-dryrun, snakemake-executor-plugin-local)
+    # plugins (snakemake-executor-plugin-dryrun,
+    # snakemake-executor-plugin-local)
     # are expected to specify False here.
     non_local_exec=True,
     # Define whether your executor plugin implies that there is no shared
@@ -107,6 +109,30 @@ def warn_on_jobcontext(self, done=None):
     def additional_general_args(self):
         return "--executor slurm-jobstep --jobs 1"
 
+    def run_jobs(
+        self,
+        jobs: List[JobExecutorInterface],
+    ):
+        """Run a list of jobs that is ready at a given point in time.
+
+        By default, this method just runs each job individually.
+        This method is overwritten in the executor plugin to implement
+        - SLURM array jobs.
+        - pooled jobs (currently in planning)
+        """
+        self.logger.debug(f"jobs: {jobs}")
+        # self.logger.debug(f"job attributes: {dir(jobs[0])}")
+        self.logger.debug(f"jobs length: {len(jobs)}")
+        # sys.exit(1)
+        # g = groupby(jobs, key=Job.rule)
+        # self.logger.debug(f"Jobs grouped by rule: {g}")
+        # sys.exit(1)
+        for job in jobs:
+            if job.name == "foo":
+                break
+            self.run_job_pre(job)
+            self.run_job(job)
+
     def run_job(self, job: JobExecutorInterface):
         # Implement here how to run a job.
         # You can access the job's resources, etc.
@@ -140,9 +166,9 @@ def run_job(self, job: JobExecutorInterface):
         # we use a run_uuid as the job-name, to allow `--name`-based
         # filtering in the job status checks (`sacct --name` and `squeue --name`)
         if wildcard_str == "":
-            comment_str = f"rule_{job.name}"
+            comment_str = f"{group_or_rule}"
         else:
-            comment_str = f"rule_{job.name}_wildcards_{wildcard_str}"
+            comment_str = f"{group_or_rule}_wildcards_{wildcard_str}"
         call = (
             f"sbatch "
             f"--parsable "
@@ -475,7 +501,8 @@ async def job_stati(self, command):
             }
         except subprocess.CalledProcessError as e:
             self.logger.error(
-                f"The job status query failed with command: {command}\n"
+                "The job status query failed with"
+                f" command: {command}\n"
                 f"Error message: {e.stderr.strip()}\n"
             )
             pass

snakemake_executor_plugin_slurm/utils.py

@@ -14,3 +14,21 @@ def delete_slurm_environment():
     for var in os.environ:
         if var.startswith("SLURM_"):
             del os.environ[var]
+
+
+def clean_old_logs(logdir, age_cutoff):
+    """
+    Function to delete files older than 'age_cutoff'
+    in the SLURM 'logdir'
+    """
+    cutoff_secs = age_cutoff * 86400
+    for root, _, files in os.walk(logdir, topdown=False):
+        for fname in files:
+            file_path = os.path.join(root, fname)
+            if age_cutoff > 0:
+                filestamp = os.stat(file_path).st_mtime
+                if filestamp > cutoff_secs:
+                    os.remove(file_path)
+        # remove empty rule top dir, if empty
+        if len(os.listdir(root)) == 0:
+            os.rmdir(root)