Distribute log files across subdirs.

doc/lsst.ctrl.bps.parsl/use.rst

@@ -52,6 +52,9 @@ The following configuration settings can be used in configuring the plugin:
 * ``parsl.log_level`` (`str`): logging level for Parsl; may be one of ``CRITICAL``, ``DEBUG``, ``ERROR``, ``FATAL``, ``INFO``, ``WARN``.
 * ``project`` (`str`): project name; defaults to ``bps``.
 * ``campaign`` (`str`): campaign name; defaults to the user name (which can also be set via the ``username`` setting).
+* ``subDirTemplate`` (`str`): template used to define log subdirectories in order to avoid having too many files in a single directory; defaults to a very generic template defined by ctrl_bps in bps_defaults.yaml_.  To run with no subdirectories (original plugin behavior), in the submit yaml set ``subDirTemplate`` to the empty string (``subDirTemplate: ''``).   
+
+.. _bps_defaults.yaml: https://github.com/lsst/ctrl_bps/blob/main/python/lsst/ctrl/bps/etc/bps_defaults.yaml
 
 The workflow job name is taken to be ``<project>.<campaign>``.
 

python/lsst/ctrl/bps/parsl/job.py

@@ -28,6 +28,7 @@
 import os
 import re
 import subprocess
+from collections import defaultdict
 from collections.abc import Sequence
 from functools import partial
 from textwrap import dedent
@@ -120,9 +121,25 @@ def __init__(
         self.file_paths = file_paths
         self.future = None
         self.done = False
+
+        # Determine directory for job stdout and stderr
         log_dir = os.path.join(get_bps_config_value(self.config, "submitPath", str, required=True), "logs")
-        self.stdout = os.path.join(log_dir, self.name + ".stdout")
-        self.stderr = os.path.join(log_dir, self.name + ".stderr")
+        _, template = self.config.search(
+            "subDirTemplate",
+            opt={
+                "curvals": {"curr_site": self.config["computeSite"], "curr_cluster": self.generic.label},
+                "replaceVars": False,
+                "default": "",
+            },
+        )
+        job_vals = defaultdict(str)
+        job_vals["label"] = self.generic.label
+        if self.generic.tags:
+            job_vals.update(self.generic.tags)
+        subdir = os.path.normpath(template.format_map(job_vals))
+
+        self.stdout = os.path.join(log_dir, subdir, self.name + ".stdout")
+        self.stderr = os.path.join(log_dir, subdir, self.name + ".stderr")
 
     def __reduce__(self):
         """Recipe for pickling"""
@@ -288,6 +305,8 @@ def run_local(self):
             return
         command = self.get_command_line(False)
         command = self.evaluate_command_line(command)
+        os.makedirs(os.path.dirname(self.stdout), exist_ok=True)
+        os.makedirs(os.path.dirname(self.stderr), exist_ok=True)
         with open(self.stdout, "w") as stdout, open(self.stderr, "w") as stderr:
             subprocess.check_call(command, shell=True, executable="/bin/bash", stdout=stdout, stderr=stderr)
         self.done = True