Propagate returncodes of external processes

If external subprocesses such as mafft got killed (e.g. by SLURM)
because of OOM, the main FastOMA python process did simply raise an
Exception and terminated with exit code 1. This prevented nextflow
from restarting it with more memory/time.

We now report properly the exit code of Mafft and FastTree and exit
the python code with the same exit code (stored in WrapperError)

Author: alpae

FastOMA/_infer_subhog.py

@@ -30,6 +30,7 @@
 from ._utils_subhog import MSAFilter, MSAFilterElbow, MSAFilterTrimAL
 
 from .zoo.utils import unique
+from .zoo.wrappers import WrapperError
 
 low_so_detection = True # detection of proteins with low species overlap score in gene tree
 fragment_detection = True  # this also need to be consistent in _hog_class.py
@@ -76,10 +77,15 @@ def read_infer_xml_rhog(rhogid, inferhog_concurrent_on, pickles_rhog_folder,  pi
     species_names_rhog = list(set(species_names_rhog))
     logger.info("Number of unique species in rHOG " + rhogid + " is " + str(len(species_names_rhog)) + ".")
 
-    if inferhog_concurrent_on:  # for big HOG we use parallelization at the level taxonomic level using concurrent
-        infer_hogs_concurrent(species_tree, rhogid, pickles_subhog_folder_all, rhogs_fa_folder, conf_infer_subhhogs)
-    else:
-        infer_hogs_for_rhog_levels_recursively(species_tree, rhogid, pickles_subhog_folder_all, rhogs_fa_folder, conf_infer_subhhogs)
+    try:
+        if inferhog_concurrent_on:  # for big HOG we use parallelization at the level taxonomic level using concurrent
+            infer_hogs_concurrent(species_tree, rhogid, pickles_subhog_folder_all, rhogs_fa_folder, conf_infer_subhhogs)
+        else:
+            infer_hogs_for_rhog_levels_recursively(species_tree, rhogid, pickles_subhog_folder_all, rhogs_fa_folder, conf_infer_subhhogs)
+    except WrapperError as e:
+        logger.exception("Error of external tool during subhog inference: %s", str(e))
+        sys.exit(getattr(e, "exit_code", 1))
+
 
     #####  Now read the final pickle file for this rootHOG
     root_node_name = species_tree.name

FastOMA/zoo/utils.py

@@ -109,4 +109,18 @@ def unique(seq):
     return [x for x in seq if x not in seen and not seen.add(x)]
 
 
-
+def summarize_long_message(message: str, head_chars: int = 1000, tail_lines: int = 50) -> str:
+    """
+    Summarize a potentially long messge string.
+    Shows first head_chars characters and last tail_lines lines with ellipsis.
+    """
+    output = message or ""
+    # Get last tail_lines
+    lines = output.splitlines()
+    tail = "\n".join(lines[-tail_lines:]) if len(lines) > tail_lines else "\n".join(lines)
+    # Get first head_chars
+    head = output[:head_chars] + ("…" if len(output) > head_chars else "")
+    if len(lines) > tail_lines or len(output) > head_chars:
+        return f"{head}\n...\n{tail}"
+    else:
+        return output

FastOMA/zoo/wrappers/__init__.py

@@ -1,4 +1,6 @@
 class WrapperError(Exception):
-    pass
+    def __init__(self, message, exit_code=1):
+        super().__init__(message)
+        self.exit_code = exit_code
 
 

FastOMA/zoo/wrappers/aligners/mafft.py

@@ -7,6 +7,7 @@
 from .base_aligner import Aligner, AlignmentInput, DataType
 from ...seq_utils.utils import iter_seqrecs_from_any
 from ...wrappers import WrapperError
+from ...utils import summarize_long_message
 from ..options import StringOption, FlagOption, IntegerOption, FloatOption, MultiOption, OptionSet
 import tempfile
 import logging
@@ -120,7 +121,8 @@ def __call__(self, *args, **kwargs):
         logger.debug('Output of Mafft: stdout={}; stderr={}'.format(output, error))
         if len(output) == 0 and len(error) > 0:
             logger.warning('is MAFFT_BINARIES set correctly: {}'.format(os.getenv('MAFFT_BINARIES', '')))
-            raise WrapperError('Mafft did not compute any alignments. StdErr: {}'.format(error))
+            logger.warning("Mafft did not compute any alignments. StdErr:\n%s", summarize_long_message(error))
+            raise WrapperError('Mafft did not compute any alignments')
         self.result = self._read_result(output)  # store result
         self.stdout = output
         self.stderr = error
@@ -140,6 +142,18 @@ def _call(self, filename, *args, **kwargs):
         """
         self.cli('{} {}'.format(self.command(), filename),
                  wait=True)
+
+        ret = self.cli.process.returncode
+        if ret != 0:
+            logger.error('Mafft returned non-zero exit status: {}'.format(ret))
+            logger.error('Output of Mafft:\n\n%s\nstdout=\n%s\n{}\n\n%s\nstderr=\n%s\n{}\n\n',
+                         "=" * 30, "=" * 30, summarize_long_message(self.cli.get_stdout()),
+                         "=" * 30, "=" * 30, summarize_long_message(self.cli.get_stderr()))
+            if ret < 0:
+                sig = -ret
+                raise WrapperError(f'Mafft was terminated by signal {sig}', exit_code=128 + sig)
+            else:
+                raise WrapperError(f'Mafft exited with code {ret}', exit_code=ret)
         return self.cli.get_stdout(), self.cli.get_stderr()
 
     def command(self):

FastOMA/zoo/wrappers/treebuilders/fasttree.py

@@ -14,6 +14,7 @@
 
 from ..abstract_cli import AbstractCLI
 from ..options import OptionSet, StringOption, IntegerOption
+from ...utils import summarize_long_message
 from ...file_utils import TempFile, TempDir
 
 logger = logging.getLogger(__name__)
@@ -89,13 +90,24 @@ def _call(self, filename, *args, **kwargs):
         self.returncode = self.cli.process.returncode
 
         if self.returncode != 0:
+            self.stdout = self.cli.get_stdout()
             self.stderr = self.cli.get_stderr()
             last_error_line = self.stderr.split('\n')[-1].strip()
-            msg = f"Fasttree failed on {filename}: {last_error_line}"
+            logger.error('FastTree returned non-zero exit status: {}'.format(self.returncode))
+            logger.error('Output of FastTree:\n\n%s\nstdout=\n%s\n{}\n\n%s\nstderr=\n%s\n{}\n\n',
+                         "=" * 30, "=" * 30, summarize_long_message(self.stdout),
+                         "=" * 30, "=" * 30, summarize_long_message(self.stderr))
+            if self.returncode < 0:
+                termination = f"was terminated by signal {-self.returncode}"
+                exit_code = 128 - self.returncode
+            else:
+                termination = f"exited with code {self.returncode}"
+                exit_code = self.returncode
+            msg = f"Fasttree {termination} on {filename}: {last_error_line}"
             logger.error(msg)
-            raise WrapperError(msg, self.stderr)
+            raise WrapperError(msg, exit_code=exit_code)
 
-        return (self.cli.get_stdout(), self.cli.get_stderr())
+        return self.cli.get_stdout(), self.cli.get_stderr()
 
     def command(self):
         return str(self.options)