feat: custom log file behaviour

snakemake_executor_plugin_slurm/__init__.py

@@ -12,6 +12,7 @@
 import time
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta
+import pathlib
 from typing import List, Generator, Optional
 import uuid
 from snakemake_interface_executor_plugins.executors.base import SubmittedJobInfo
@@ -26,18 +27,53 @@
 from snakemake_interface_common.exceptions import WorkflowError
 from snakemake_executor_plugin_slurm_jobstep import get_cpus_per_task
 
-from .utils import delete_slurm_environment
+from .utils import delete_slurm_environment, clean_old_logs
 
 
 @dataclass
 class ExecutorSettings(ExecutorSettingsBase):
+    logdir: Optional[str] = field(
+        default=f"/home/{os.environ['USER']}/.snakemake/slurm_logs",
+        metadata={
+            "help": """
+                   Per default the SLURM log directory (writing output is required by SLURM)
+                   is '~/.snakemake/slurm_logs'. This flag allows to set an alternative 
+                   directory.
+                   """,
+            "env_var": False,
+            "required": False,
+        },
+    )
+    keep_successful_logs: bool = field(
+        default=False,
+        metadata={
+            "help": """
+                   Per default SLURM log files will be deleted upon sucessful completion
+                   of a job. Whenever a SLURM job fails, its log file will be preserved.
+                   This flag allows to keep all SLURM log files, even those of successful
+                   jobs.
+                   """,
+            "env_var": False,
+            "required": False,
+        },
+    )
+    delete_logfiles_older_than: Optional[int] = field(
+        default=10,
+        metadata={
+            "help": """
+                Per default SLURM log files in the SLURM log directory of a workflow
+                will be deleted after 10 days. Setting this flag allows to change this behaviour.
+                If set to '0', no old files will be deleted.
+                """
+        },
+    )
     init_seconds_before_status_checks: Optional[int] = field(
         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,
         },
@@ -92,6 +128,13 @@ def __post_init__(self):
         self._fallback_partition = None
         self._preemption_warning = False  # no preemption warning has been issued
 
+    def __del__(self):
+        # the logger attribute is not available in this destructor!
+        clean_old_logs(
+            self.workflow.executor_settings.logdir,
+            self.workflow.executor_settings.delete_logfiles_older_than,
+        )
+
     def warn_on_jobcontext(self, done=None):
         if not done:
             if "SLURM_JOB_ID" in os.environ:
@@ -104,6 +147,9 @@ def warn_on_jobcontext(self, done=None):
                 delete_slurm_environment()
         done = True
 
+    # def delete_old_logs(self):
+    #    self.workflow.executor_settings.delete_logfiles_older_than
+
     def additional_general_args(self):
         return "--executor slurm-jobstep --jobs 1"
 
@@ -123,18 +169,21 @@ def run_job(self, job: JobExecutorInterface):
         except AttributeError:
             wildcard_str = ""
 
-        slurm_logfile = os.path.abspath(
-            f".snakemake/slurm_logs/{group_or_rule}/{wildcard_str}/%j.log"
+        slurm_logfile = (
+            self.workflow.executor_settings.logdir
+            + os.path.sep
+            + "{log_path}/{group_or_rule}/{wildcard_str}/%j.log"
         )
-        logdir = os.path.dirname(slurm_logfile)
+
+        slurm_logdir = os.path.dirname(slurm_logfile)
         # this behavior has been fixed in slurm 23.02, but there might be plenty of
         # older versions around, hence we should rather be conservative here.
-        assert "%j" not in logdir, (
+        assert "%j" not in slurm_logdir, (
             "bug: jobid placeholder in parent dir of logfile. This does not work as "
             "we have to create that dir before submission in order to make sbatch "
             "happy. Otherwise we get silent fails without logfiles being created."
         )
-        os.makedirs(logdir, exist_ok=True)
+        os.makedirs(slurm_logdir, exist_ok=True)
 
         # generic part of a submission string:
         # we use a run_uuid as the job-name, to allow `--name`-based
@@ -380,6 +429,11 @@ async def check_active_jobs(
                     self.report_job_success(j)
                     any_finished = True
                     active_jobs_seen_by_sacct.remove(j.external_jobid)
+                    if not self.workflow.executor_settings.keep_successful_logs:
+                        self.logger.debug(
+                            f"removing log for successful job with SLURM ID '{j.external_jobid}'"
+                        )
+                        os.remove(j.aux["slurm_logfile"])
                 elif status == "PREEMPTED" and not self._preemption_warning:
                     self._preemption_warning = True
                     self.logger.warning(

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)