Log computer load when we report CPU misconfiguration

The current load on the machine is relevant when interpreting the actual
overspent seconds (but you will have to obtain the number of CPU-cores
on the particular compute node elsewhere).

(cherry picked from commit 7c75f0c)

Author: berland

src/_ert/events.py

@@ -82,6 +82,7 @@ class ForwardModelStepRunning(ForwardModelStepBaseEvent):
     max_memory_usage: int | None = None
     current_memory_usage: int | None = None
     cpu_seconds: float = 0.0
+    computer_load: float | None = None
 
 
 class ForwardModelStepSuccess(ForwardModelStepBaseEvent):

src/_ert/forward_model_runner/forward_model_step.py

@@ -17,7 +17,14 @@
 from subprocess import Popen, run
 from typing import TYPE_CHECKING
 
-from psutil import AccessDenied, NoSuchProcess, Process, TimeoutExpired, ZombieProcess
+from psutil import (
+    AccessDenied,
+    NoSuchProcess,
+    Process,
+    TimeoutExpired,
+    ZombieProcess,
+    getloadavg,
+)
 
 from .reporting.message import (
     Exited,
@@ -215,6 +222,8 @@ def _run(self) -> Generator[Start | Exited | Running]:
         max_memory_usage = 0
         fm_step_pids = {int(process.pid)}
         cpu_seconds_processtree: ProcesstreeTimer = ProcesstreeTimer()
+        computer_load_1min_sum = 0.0
+        computer_load_1min_samples = 0
         while True:
             try:
                 exit_code = process.wait(timeout=self.MEMORY_POLL_PERIOD)
@@ -238,6 +247,9 @@ def _run(self) -> Generator[Start | Exited | Running]:
             cpu_seconds_processtree.update(cpu_seconds_snapshot)
             fm_step_pids |= pids
             max_memory_usage = max(memory_rss, max_memory_usage)
+            with contextlib.suppress(OSError):
+                computer_load_1min_sum += getloadavg()[0]
+                computer_load_1min_samples += 1
             yield Running(
                 self,
                 ProcessTreeStatus(
@@ -246,6 +258,11 @@ def _run(self) -> Generator[Start | Exited | Running]:
                     fm_step_id=self.index,
                     fm_step_name=self.step_data.get("name"),
                     cpu_seconds=cpu_seconds_processtree.total_cpu_seconds(),
+                    computer_load=(
+                        computer_load_1min_sum / computer_load_1min_samples
+                        if computer_load_1min_samples
+                        else None
+                    ),
                     oom_score=oom_score,
                 ),
             )

src/_ert/forward_model_runner/reporting/event.py

@@ -256,6 +256,7 @@ def _step_handler(self, msg: Start | Running | Exited) -> None:
                     max_memory_usage=msg.memory_status.max_rss,
                     current_memory_usage=msg.memory_status.rss,
                     cpu_seconds=msg.memory_status.cpu_seconds,
+                    computer_load=msg.memory_status.computer_load,
                 )
             )
 

src/_ert/forward_model_runner/reporting/message.py

@@ -38,6 +38,8 @@ class ProcessTreeStatus:
 
     cpu_seconds: float = 0.0
 
+    computer_load: float | None = None  # Accumulated and averaged over time
+
     oom_score: int | None = None
 
     def __post_init__(self) -> None:

src/ert/ensemble_evaluator/evaluator.py

@@ -720,11 +720,13 @@ def detect_overspent_cpu(
             parallelization_obtained > allowed_overspending
             and self.ensemble.queue_system != QueueSystem.LOCAL
         ):
+            computer_load = fm_step.get(ids.COMPUTER_LOAD, -1.0)
             logger.warning(
                 f"Misconfigured NUM_CPU, forward model step '{fm_step.get(ids.NAME)}' "
                 f"for realization {real_id} spent {cpu_seconds} cpu seconds "
                 f"with wall clock duration {duration:.1f} seconds, a factor of "
-                f"{parallelization_obtained:.2f}, while NUM_CPU was {num_cpu}."
+                f"{parallelization_obtained:.2f}, while NUM_CPU was {num_cpu}, "
+                f"computer load {computer_load:.2f}"
             )
             if parallelization_obtained > overspending_warning_threshold:
                 warning_msg = (

src/ert/ensemble_evaluator/identifiers.py

@@ -6,6 +6,7 @@
 INDEX: Final = "index"
 MAX_MEMORY_USAGE: Final = "max_memory_usage"
 CPU_SECONDS: Final = "cpu_seconds"
+COMPUTER_LOAD: Final = "computer_load"
 NAME: Final = "name"
 START_TIME: Final = "start_time"
 STATUS: Final = "status"

src/ert/ensemble_evaluator/snapshot.py

@@ -354,6 +354,7 @@ def update_from_realization_event(
                             current_memory_usage=None,
                             max_memory_usage=None,
                             cpu_seconds=None,
+                            computer_load=None,
                             error=None,
                             stdout=None,
                             stderr=None,
@@ -385,6 +386,7 @@ def update_from_fm_event(
                 fm_data["current_memory_usage"] = event.current_memory_usage
                 fm_data["max_memory_usage"] = event.max_memory_usage
                 fm_data["cpu_seconds"] = event.cpu_seconds
+                fm_data["computer_load"] = event.computer_load
             case ForwardModelStepSuccess():
                 end_time = convert_iso8601_to_datetime(timestamp)
                 # Make sure error msg from previous failed run is replaced
@@ -459,6 +461,7 @@ class FMStepSnapshot(TypedDict, total=False):
     current_memory_usage: int | None
     max_memory_usage: int | None
     cpu_seconds: float | None
+    computer_load: float | None
     name: str | None
     error: str | None
     stdout: str | None

tests/ert/unit_tests/ensemble_evaluator/test_ensemble_evaluator.py

@@ -342,6 +342,36 @@ def test_overspent_cpu_is_logged(
     caplog.clear()
 
 
+def test_that_average_computer_load_is_included_in_overspent_cpu_message(
+    evaluator_to_use,
+    caplog,
+    monkeypatch,
+):
+    caplog.set_level(logging.WARNING)
+    evaluator, _ = evaluator_to_use
+
+    monkeypatch.setattr(TestEnsemble, "queue_system", QueueSystem.LSF)
+
+    start = datetime.datetime(2024, 1, 1, tzinfo=datetime.UTC)
+    duration = 100
+    num_cpu = 1
+    cpu_seconds = 1000.0  # Guarantees overspending is detected
+
+    evaluator.detect_overspent_cpu(
+        num_cpu,
+        "dummy",
+        FMStepSnapshot(
+            start_time=start,
+            end_time=start + datetime.timedelta(seconds=duration),
+            cpu_seconds=cpu_seconds,
+            computer_load=12.34,
+        ),
+    )
+
+    assert "Misconfigured NUM_CPU" in caplog.text
+    assert "computer load 12.34" in caplog.text
+
+
 @pytest.mark.slow
 async def test_snapshot_on_resubmit_is_cleared(evaluator_to_use):
     (evaluator, event_queue) = evaluator_to_use