Add option to write flux log files

executorlib/__init__.py

@@ -94,6 +94,7 @@ def __init__(
         flux_executor=None,
         flux_executor_pmi_mode: Optional[str] = None,
         flux_executor_nesting: bool = False,
+        flux_log_files: bool = False,
         pysqa_config_directory: Optional[str] = None,
         hostname_localhost: Optional[bool] = None,
         block_allocation: bool = False,
@@ -115,6 +116,7 @@ def __new__(
         flux_executor=None,
         flux_executor_pmi_mode: Optional[str] = None,
         flux_executor_nesting: bool = False,
+        flux_log_files: bool = False,
         pysqa_config_directory: Optional[str] = None,
         hostname_localhost: Optional[bool] = None,
         block_allocation: bool = False,
@@ -194,6 +196,7 @@ def __new__(
                 flux_executor=flux_executor,
                 flux_executor_pmi_mode=flux_executor_pmi_mode,
                 flux_executor_nesting=flux_executor_nesting,
+                flux_log_files=flux_log_files,
                 pysqa_config_directory=pysqa_config_directory,
                 hostname_localhost=hostname_localhost,
                 block_allocation=block_allocation,
@@ -211,6 +214,7 @@ def __new__(
                 flux_executor=flux_executor,
                 flux_executor_pmi_mode=flux_executor_pmi_mode,
                 flux_executor_nesting=flux_executor_nesting,
+                flux_log_files=flux_log_files,
                 hostname_localhost=hostname_localhost,
                 block_allocation=block_allocation,
                 init_function=init_function,
@@ -230,6 +234,7 @@ def __new__(
                 flux_executor=flux_executor,
                 flux_executor_pmi_mode=flux_executor_pmi_mode,
                 flux_executor_nesting=flux_executor_nesting,
+                flux_log_files=flux_log_files,
                 hostname_localhost=hostname_localhost,
                 block_allocation=block_allocation,
                 init_function=init_function,

executorlib/cache/executor.py

@@ -10,6 +10,7 @@
 from executorlib.standalone.inputcheck import (
     check_executor,
     check_flux_executor_pmi_mode,
+    check_flux_log_files,
     check_hostname_localhost,
     check_max_workers_and_cores,
     check_nested_flux_executor,
@@ -88,6 +89,7 @@ def create_file_executor(
     flux_executor=None,
     flux_executor_pmi_mode: Optional[str] = None,
     flux_executor_nesting: bool = False,
+    flux_log_files: bool = False,
     pysqa_config_directory: Optional[str] = None,
     hostname_localhost: Optional[bool] = None,
     block_allocation: bool = False,
@@ -109,6 +111,7 @@ def create_file_executor(
     check_hostname_localhost(hostname_localhost=hostname_localhost)
     check_executor(executor=flux_executor)
     check_nested_flux_executor(nested_flux_executor=flux_executor_nesting)
+    check_flux_log_files(flux_log_files=flux_log_files)
     return FileExecutor(
         cache_directory=cache_directory,
         resource_dict=resource_dict,

executorlib/interactive/executor.py

@@ -10,6 +10,7 @@
 from executorlib.standalone.inputcheck import (
     check_command_line_argument_lst,
     check_executor,
+    check_flux_log_files,
     check_gpus_per_worker,
     check_init_function,
     check_nested_flux_executor,
@@ -154,6 +155,7 @@ def create_executor(
     flux_executor=None,
     flux_executor_pmi_mode: Optional[str] = None,
     flux_executor_nesting: bool = False,
+    flux_log_files: bool = False,
     hostname_localhost: Optional[bool] = None,
     block_allocation: bool = False,
     init_function: Optional[callable] = None,
@@ -213,6 +215,7 @@ def create_executor(
         resource_dict["flux_executor"] = flux_executor
         resource_dict["flux_executor_pmi_mode"] = flux_executor_pmi_mode
         resource_dict["flux_executor_nesting"] = flux_executor_nesting
+        resource_dict["flux_log_files"] = flux_log_files
         if block_allocation:
             resource_dict["init_function"] = init_function
             return InteractiveExecutor(
@@ -235,6 +238,7 @@ def create_executor(
     elif backend == "slurm_allocation":
         check_executor(executor=flux_executor)
         check_nested_flux_executor(nested_flux_executor=flux_executor_nesting)
+        check_flux_log_files(flux_log_files=flux_log_files)
         if block_allocation:
             resource_dict["init_function"] = init_function
             return InteractiveExecutor(
@@ -257,6 +261,7 @@ def create_executor(
     elif backend == "local":
         check_executor(executor=flux_executor)
         check_nested_flux_executor(nested_flux_executor=flux_executor_nesting)
+        check_flux_log_files(flux_log_files=flux_log_files)
         check_gpus_per_worker(gpus_per_worker=resource_dict["gpus_per_core"])
         check_command_line_argument_lst(
             command_line_argument_lst=resource_dict["slurm_cmd_args"]

executorlib/interactive/flux.py

@@ -19,6 +19,7 @@ class FluxPythonSpawner(BaseSpawner):
         flux_executor (flux.job.FluxExecutor, optional): The FluxExecutor instance. Defaults to None.
         flux_executor_pmi_mode (str, optional): The PMI option. Defaults to None.
         flux_executor_nesting (bool, optional): Whether to use nested FluxExecutor. Defaults to False.
+        flux_log_files (bool, optional): Write flux stdout and stderr files. Defaults to False.
     """
 
     def __init__(
@@ -31,6 +32,7 @@ def __init__(
         flux_executor: Optional[flux.job.FluxExecutor] = None,
         flux_executor_pmi_mode: Optional[str] = None,
         flux_executor_nesting: bool = False,
+        flux_log_files: bool = False,
     ):
         super().__init__(
             cwd=cwd,
@@ -42,6 +44,7 @@ def __init__(
         self._flux_executor = flux_executor
         self._flux_executor_pmi_mode = flux_executor_pmi_mode
         self._flux_executor_nesting = flux_executor_nesting
+        self._flux_log_files = flux_log_files
         self._future = None
 
     def bootup(
@@ -85,6 +88,9 @@ def bootup(
             jobspec.setattr_shell_option("pmi", self._flux_executor_pmi_mode)
         if self._cwd is not None:
             jobspec.cwd = self._cwd
+        if self._flux_log_files and self._cwd is not None:
+            jobspec.stderr = os.path.join(self._cwd, "flux.err")
+            jobspec.stdout = os.path.join(self._cwd, "flux.out")
         self._future = self._flux_executor.submit(jobspec)
 
     def shutdown(self, wait: bool = True):

executorlib/standalone/inputcheck.py

@@ -147,6 +147,16 @@ def check_flux_executor_pmi_mode(flux_executor_pmi_mode: Optional[str]) -> None:
         )
 
 
+def check_flux_log_files(flux_log_files: Optional[bool]) -> None:
+    """
+    Check if flux_log_files is True and raise a ValueError if it is.
+    """
+    if flux_log_files:
+        raise ValueError(
+            "The flux_log_files parameter is only supported for the flux framework backend."
+        )
+
+
 def check_pysqa_config_directory(pysqa_config_directory: Optional[str]) -> None:
     """
     Check if pysqa_config_directory is None and raise a ValueError if it is not.