From 2baf3477c34b5028cbcc0a3de1479aa30685e749 Mon Sep 17 00:00:00 2001 From: Sindre Langeveld Date: Tue, 14 Jul 2026 13:29:41 +0200 Subject: [PATCH] ENH: Print failure logs in logfiles to display in ERT --- src/runrms/executor/fm_executor.py | 53 ++++- tests/conftest.py | 57 +++++ tests/test_fm_executor.py | 331 ++++++++++++++++++++++++++++- 3 files changed, 438 insertions(+), 3 deletions(-) diff --git a/src/runrms/executor/fm_executor.py b/src/runrms/executor/fm_executor.py index 50faf80..453e33d 100644 --- a/src/runrms/executor/fm_executor.py +++ b/src/runrms/executor/fm_executor.py @@ -1,5 +1,6 @@ import glob import os +import re import subprocess import sys import time @@ -63,6 +64,37 @@ def _exec_rms(self) -> int: comp_process = subprocess.run(args=args, check=False) return comp_process.returncode + def _find_job_failures(self, logfile_path: Path) -> str: + """Reads through a .log file and returns logs regarding failing jobs. + + Any failure logs must be at the end of the log file as a failing job + breaks the workflow and will be the last thing that is logged. + + The method first finds the logs for the last job that was started and returns + the findings as a formatted string if any job failures are found. + """ + current_job_line_no = 0 + + # Find the line number where the logging of the last job starts + with open(logfile_path) as logfile: + for line_no, line in enumerate(logfile, start=1): + if line.strip().startswith("
"):
+                    current_job_line_no = line_no
+            last_job_line_no = current_job_line_no
+            if last_job_line_no == 0:
+                return ""
+
+        # Collect the logs for the last job and look for failures
+        with open(logfile_path) as logfile:
+            job_failed_msg = ""
+            for line_no, line in enumerate(logfile, start=1):
+                if line_no >= last_job_line_no:
+                    cleaned_text = re.sub(r"<[^>]*>", "", line)
+                    if cleaned_text != "":
+                        job_failed_msg += cleaned_text
+
+        return job_failed_msg if "failed" in job_failed_msg else ""
+
     def print_failure(self, exit_status: int) -> None:
         run_path = self.config.run_path.resolve()
         # Reverse sort so workflow.log is (probably) first and
@@ -111,7 +143,7 @@ def print_failure(self, exit_status: int) -> None:
         * RMS.stderr.NN and RMS.stdout.NN
         * rms/model/workflow.log
         * Other named log files in rms/model, e.g. workflow_sim2seis.log
-        * rms/model/YYYYMMDD-HHMMSS-XXXXX-RMS.log corresponding to your run
+        * rms/model/YYYYMMDD-HHMMSS-XXXXXX-RMS.log corresponding to your run
 
         The following log files were found in this realization's run path:
 
@@ -119,6 +151,25 @@ def print_failure(self, exit_status: int) -> None:
             )
             fail_msg += "\n".join([f"* {f}" for f in log_files])
 
+            for log_file in log_files:
+                if re.match(
+                    r"^\d{8}-\d{6}-[A-Za-z0-9]{6}-RMS", os.path.basename(log_file)
+                ):
+                    # These logfiles are unstructured and will not give any results
+                    continue
+                try:
+                    job_failed_msg = self._find_job_failures(Path(log_file))
+                except Exception:
+                    job_failed_msg = ""
+                if job_failed_msg:
+                    fail_msg += dedent(
+                        f"""
+    \nThe following logs regarding failed jobs were found in the log file {log_file}:"
+                        """
+                    )
+                    fail_msg += job_failed_msg
+                    break
+
         print(fail_msg, file=sys.stderr)
 
     @property
diff --git a/tests/conftest.py b/tests/conftest.py
index 8004d28..198f55b 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -221,3 +221,60 @@ def master_version() -> str:
                 End GEOMATIC file header
                 """
     )
+
+
+@pytest.fixture
+def workflow_log() -> Callable[[Path], Path]:
+    """Returns a helper function to write a workflow log to file."""
+
+    def _workflow_log(run_path: Path) -> Path:
+        """Writes a workflow log to the workflow.log file at the provided run path."""
+        path = run_path / "workflow.log"
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(
+                dedent(
+                    """
+                Log format
+                
+                Project realization 1
+
+                
+                Job 1 - Some job that runs OK - for project realization 0
+                

+                Completed job 1
+                
Elapsed time0:00:02.2
Total elapsed time0:00:02.9

+
+                
+                Job 2 - structural_model_initial - for project realization 0
+                
+                Project realization 0
+                
+
+                Job 1 - Note - for project realization 0 - skipped
+                

+                Completed job 1
+                
+                
+                Job 2 - failing_python_job.py - for project realization 0
+                Traceback (most recent call last):
+                Python script, line 36
+                Python script, line 24, in main
+                ModuleNotFoundError: No module named 'non_existing_package'
+                Job: failing_python_job.py failed.
+                

+                Failed job 2
+                Workflow structural_model_initial failed
+                
Finish time15:37:07
Elapsed time for workflow0:00:20.0

+                

+                Job: structural_model_initial failed.
+                

+                Failed job 2
+                Workflow MAIN failed
+                
Finish time15:37:07
Elapsed time for workflow0:00:20.5

+                

+                    """  # noqa: E501
+                )
+            )
+        return path
+
+    return _workflow_log
diff --git a/tests/test_fm_executor.py b/tests/test_fm_executor.py
index ef7ddd6..5604bc9 100644
--- a/tests/test_fm_executor.py
+++ b/tests/test_fm_executor.py
@@ -1,10 +1,12 @@
 import json
 import os
+import re
 import stat
 import subprocess
 from collections.abc import Callable
 from pathlib import Path
-from unittest.mock import Mock, patch
+from textwrap import dedent
+from unittest.mock import Mock, call, patch
 
 import pytest
 from pytest import CaptureFixture, MonkeyPatch
@@ -262,7 +264,7 @@ def test_license_file_from_wrapper_overwrites(
     fm_executor_env: Path,
     monkeypatch: MonkeyPatch,
 ) -> None:
-    """Tests that the the batch license file configuration is preferred. This is set by
+    """Tests that the batch license file configuration is preferred. This is set by
     the fm_executor_env, so we unset the environment variable to check.
 
     The configuration has
@@ -488,6 +490,331 @@ def test_print_failure_when_logs_found_in_rms_model(
         assert line.startswith("\t") is False
 
 
+def test_print_failure_when_job_failure_in_logs(
+    fm_executor_env: Path,
+    workflow_log: Callable[[Path], Path],
+    capsys: CaptureFixture[str],
+) -> None:
+    """Job failure logs should be printed when job failures in the logfiles."""
+    config = _create_config(
+        iens=0,
+        project="project",
+        workflow="workflow",
+        run_path="run_path",
+        allow_no_env=True,
+        config_file="runrms.yml",
+    )
+
+    run_path = fm_executor_env / "run_path"
+    workflow_log(run_path)
+
+    rms = ForwardModelExecutor(config)
+    action = {"exit_status": 1}
+    with open(run_path / "action.json", "w") as f:
+        f.write(json.dumps(action))
+
+    with patch.object(
+        rms, "_find_job_failures", return_value="Failed job"
+    ) as mocked_find_job_failures:
+        rms.run()
+        captured = capsys.readouterr()
+
+        assert "workflow.log" in captured.err
+        assert (
+            "The following logs regarding failed jobs were found in the log file"
+            in captured.err
+        )
+        assert "Failed job" in captured.err
+
+        mocked_find_job_failures.assert_called_once()
+
+
+def test_print_failure_skips_logfiles_for_complete_rms_run(
+    fm_executor_env: Path,
+) -> None:
+    """The logfiles for the complete RMS run should not be checked.
+
+    Skip looking for job failures in log files on the format
+    YYYYMMDD-HHMMSS-XXXXXX-RMS.log as these are unstructured
+    and will not give any matching results.
+    """
+    config = _create_config(
+        iens=0,
+        project="project",
+        workflow="workflow",
+        run_path="run_path",
+        allow_no_env=True,
+        config_file="runrms.yml",
+    )
+
+    run_path = Path(fm_executor_env / "run_path")
+    (run_path / "20260101-000000-zdtHhM-RMS.log").touch()
+    (run_path / "20260715-092055-zdtHhM-RMS.log").touch()
+    (run_path / "20262412-135454-jPbOrM-RMS.log").touch()
+    (run_path / "20261705-133409-NgliXA-RMS.log").touch()
+
+    rms = ForwardModelExecutor(config)
+    action = {"exit_status": 1}
+    with open(run_path / "action.json", "w") as f:
+        f.write(json.dumps(action))
+
+    with patch.object(
+        rms, "_find_job_failures", return_value="Failed"
+    ) as mocked_find_job_failures:
+        rms.run()
+        mocked_find_job_failures.assert_not_called()
+
+    (run_path / "workflow.log").touch()
+    with patch.object(
+        rms, "_find_job_failures", return_value=""
+    ) as mocked_find_job_failures:
+        rms.run()
+        mocked_find_job_failures.assert_called_once_with(run_path / "workflow.log")
+
+
+def test_print_failure_print_first_logfile_when_more_than_one_log_has_failures(
+    fm_executor_env: Path,
+    workflow_log: Callable[[Path], Path],
+    capsys: CaptureFixture[str],
+) -> None:
+    """Only the first logfile with failures should be printed.
+
+    When the workflow.log file exists and has failures, only failure logs
+    from this file should be printed as those are the most relevant ones.
+
+    If workflow.log does not have any failures it will look for failures
+    in the other logfiles, printing the first failures it can find.
+    """
+    config = _create_config(
+        iens=0,
+        project="project",
+        workflow="workflow",
+        run_path="run_path",
+        allow_no_env=True,
+        config_file="runrms.yml",
+    )
+
+    run_path = fm_executor_env / "run_path"
+    workflow_log_path: Path = workflow_log(run_path)
+
+    other_log_path = run_path / "other_log.log"
+    with open(other_log_path, "w", encoding="utf-8") as f:
+        f.write(
+            dedent(
+                """
+                
+                Job 1 - Some other job that failed - for project realization 0
+                

+            """
+            )
+        )
+
+    rms = ForwardModelExecutor(config)
+    action = {"exit_status": 1}
+    with open(run_path / "action.json", "w") as f:
+        f.write(json.dumps(action))
+
+    with patch.object(
+        rms, "_find_job_failures", return_value="Failed job"
+    ) as mocked_find_job_failures:
+        rms.run()
+        captured = capsys.readouterr()
+
+        assert "workflow.log" in captured.err
+        assert "other_log" in captured.err
+
+        assert (
+            "The following logs regarding failed jobs were found in the log file"
+            in captured.err
+        )
+        mocked_find_job_failures.assert_called_once_with(workflow_log_path)
+
+        # Verify that deleting workflow.log will print failure from other file
+        mocked_find_job_failures.reset_mock()
+        workflow_log_path.unlink()
+        rms.run()
+
+        assert (
+            "The following logs regarding failed jobs were found in the log file"
+            in captured.err
+        )
+        mocked_find_job_failures.assert_called_once_with(other_log_path)
+
+
+def test_print_failure_no_job_failure_printed_when_no_job_failure_in_logs(
+    capsys: CaptureFixture[str], fm_executor_env: Path
+) -> None:
+    """No job failure logs should be printed when no job failures in the logfiles."""
+    config = _create_config(
+        iens=0,
+        project="project",
+        workflow="workflow",
+        run_path="run_path",
+        allow_no_env=True,
+        config_file="runrms.yml",
+    )
+
+    run_path = fm_executor_env / "run_path"
+    workflow_path = run_path / "workflow.log"
+    other_log_path = run_path / "other_log.log"
+    workflow_path.touch()
+    other_log_path.touch()
+
+    rms = ForwardModelExecutor(config)
+    action = {"exit_status": 1}
+    with open(run_path / "action.json", "w") as f:
+        f.write(json.dumps(action))
+
+    with patch.object(
+        rms, "_find_job_failures", return_value=""
+    ) as mocked_find_job_failures:
+        rms.run()
+        captured = capsys.readouterr()
+
+        mocked_find_job_failures.assert_has_calls(
+            [call(workflow_path), call(other_log_path)]
+        )
+        assert (
+            "The following logs regarding failed jobs were found in the log file"
+            not in captured.err
+        )
+
+
+def test_print_failure_handles_find_job_failures_exceptions(
+    capsys: CaptureFixture[str], fm_executor_env: Path
+) -> None:
+    """Exceptions occurring while looking for job failures should be handled gracefully.
+
+    Any exceptions occurring while searching through the log files for job failures,
+    should be handled gracefully and not affect any of the other printing operations.
+    """
+    config = _create_config(
+        iens=0,
+        project="project",
+        workflow="workflow",
+        run_path="run_path",
+        allow_no_env=True,
+        config_file="runrms.yml",
+    )
+
+    run_path = fm_executor_env / "run_path"
+    workflow_path = run_path / "workflow.log"
+    other_log_path = run_path / "other_log.log"
+    workflow_path.touch()
+    other_log_path.touch()
+
+    rms = ForwardModelExecutor(config)
+    action = {"exit_status": 1}
+    with open(run_path / "action.json", "w") as f:
+        f.write(json.dumps(action))
+
+    with patch.object(
+        rms, "_find_job_failures", side_effect=FileNotFoundError("Log file not found.")
+    ) as mocked_find_job_failures:
+        rms.run()
+        captured = capsys.readouterr()
+
+        mocked_find_job_failures.assert_has_calls(
+            [call(workflow_path), call(other_log_path)]
+        )
+        assert (
+            "The following logs regarding failed jobs were found in the log file"
+            not in captured.err
+        )
+
+
+def test_find_job_failures_returns_failures(
+    fm_executor_env: Path, workflow_log: Callable[[Path], Path]
+) -> None:
+    """Test that job failures are found and returned in expected format."""
+    config = _create_config(
+        iens=0,
+        project="project",
+        workflow="workflow",
+        run_path="run_path",
+        allow_no_env=True,
+        config_file="runrms.yml",
+    )
+
+    run_path = fm_executor_env / "run_path"
+    workflow_log(run_path)
+
+    rms = ForwardModelExecutor(config)
+    job_failed_msg = rms._find_job_failures(run_path / "workflow.log")
+
+    assert "Job: failing_python_job.py failed." in job_failed_msg
+    assert (
+        "ModuleNotFoundError: No module named 'non_existing_package'" in job_failed_msg
+    )
+    assert "Failed job" in job_failed_msg
+
+
+def test_find_job_failures_removes_html_tags(
+    fm_executor_env: Path, workflow_log: Callable[[Path], Path]
+) -> None:
+    """Test that html tags are removed from the returned job failure string."""
+    config = _create_config(
+        iens=0,
+        project="project",
+        workflow="workflow",
+        run_path="run_path",
+        allow_no_env=True,
+        config_file="runrms.yml",
+    )
+
+    run_path = fm_executor_env / "run_path"
+    workflow_log(run_path)
+
+    rms = ForwardModelExecutor(config)
+    logfile_path = run_path / "workflow.log"
+    job_failed_msg = rms._find_job_failures(logfile_path)
+
+    failed_msg_lines = [x for x in job_failed_msg.splitlines() if x != ""]
+    assert "
" not in failed_msg_lines
+
+    # Check that each message line matches a log line when html tags are removed
+    with open(logfile_path) as logfile:
+        for line in logfile:
+            if any(fail_line in line for fail_line in failed_msg_lines):
+                if "<" in line and ">" in line:
+                    assert line not in failed_msg_lines
+                    line_without_html_tag = re.sub(r"<[^>]*>", "", line)
+                    assert line_without_html_tag.strip() in failed_msg_lines
+                else:
+                    assert line.strip() in failed_msg_lines
+
+
+def test_find_job_failures_empty_string_when_no_job_failures(
+    fm_executor_env: Path,
+) -> None:
+    """Test that the failing message string is empty when no job failures in logs."""
+    config = _create_config(
+        iens=0,
+        project="project",
+        workflow="workflow",
+        run_path="run_path",
+        allow_no_env=True,
+        config_file="runrms.yml",
+    )
+
+    logfile_path = fm_executor_env / "run_path/workflow.log"
+    with open(logfile_path, "w", encoding="utf-8") as f:
+        f.write(
+            dedent(
+                """
+                
+                Job 1 - Some job that runs OK - for project realization 0
+                

+            """
+            )
+        )
+
+    rms = ForwardModelExecutor(config)
+    job_failed_msg = rms._find_job_failures(logfile_path)
+
+    assert job_failed_msg == ""
+
+
 def test_lm_license_server_overwritten_during_batch(fm_executor_env: Path) -> None:
     action = {"exit_status": 0}
     with open("run_path/action.json", "w") as f: