From f8d180f48db133412a8498031d7900bd89557b80 Mon Sep 17 00:00:00 2001 From: Kristian Bendiksen Date: Wed, 29 Apr 2026 16:31:12 +0200 Subject: [PATCH] #9336 Python: Improve error reporting from gRPC to client Surface gRPC failure context to Python script authors instead of silently returning sentinel values or raising bare gRPC errors. - RipsError now carries code/details/location and a from_rpc_error() helper. Existing single-arg construction stays valid. - Stop swallowing grpc.RpcError in case.__grid_count for non-NOT_FOUND failures and in instance._check_connection_and_version, so the underlying status is propagated. - pdmobject add_method decorator and _call_pdm_method_* attach code/details to the raised RipsError; existing message text is preserved so pytest match=... assertions keep working. - Instance.start_heartbeat / stop_heartbeat / check_alive provide an opt-in background ping that flips a sticky lost-connection flag and raises a readable RipsError from check_alive() if the server dies. - Server side: GetPdmObject in RiaGrpcAppService, RiaGrpcCaseService and RiaGrpcProjectService no longer return Status::OK with an empty reply when the underlying object is missing; they return NOT_FOUND / INTERNAL with a descriptive message. Other RiaGrpc*Service files audited and already return non-OK with text on error paths. --- GrpcInterface/Python/rips/case.py | 3 +- GrpcInterface/Python/rips/exception.py | 47 ++++- .../Python/rips/grpc_retry_interceptor.py | 8 + GrpcInterface/Python/rips/instance.py | 181 ++++++++++++++++-- GrpcInterface/Python/rips/pdmobject.py | 16 +- .../Python/rips/tests/test_error_reporting.py | 72 +++++++ GrpcInterface/RiaGrpcAppService.cpp | 3 +- GrpcInterface/RiaGrpcCaseService.cpp | 3 +- GrpcInterface/RiaGrpcProjectService.cpp | 3 +- 9 files changed, 308 insertions(+), 28 deletions(-) create mode 100644 GrpcInterface/Python/rips/tests/test_error_reporting.py diff --git a/GrpcInterface/Python/rips/case.py b/GrpcInterface/Python/rips/case.py index 5379891f67..e9a15fa7c1 100644 --- a/GrpcInterface/Python/rips/case.py +++ b/GrpcInterface/Python/rips/case.py @@ -58,6 +58,7 @@ from .resinsight_classes import ( WbsParameters as WbsParameters, ) +from .exception import RipsError from .grid import Grid as Grid from .project import Project as Project from .pdmobject import add_method @@ -85,7 +86,7 @@ def __grid_count(self) -> int: except grpc.RpcError as exception: if exception.code() == grpc.StatusCode.NOT_FOUND: return 0 - return 0 + raise RipsError.from_rpc_error(exception) from exception @add_method(Case) diff --git a/GrpcInterface/Python/rips/exception.py b/GrpcInterface/Python/rips/exception.py index b1a78710dd..2b92af2a06 100644 --- a/GrpcInterface/Python/rips/exception.py +++ b/GrpcInterface/Python/rips/exception.py @@ -1,2 +1,47 @@ +from typing import Optional + + class RipsError(Exception): - pass + """Exception raised by the rips Python client. + + Carries optional gRPC context so callers can branch on code/details + when a remote ResInsight call fails. Backwards compatible with the + legacy ``RipsError("message")`` form. + """ + + def __init__( + self, + message: object, + *, + code: Optional[object] = None, + details: Optional[str] = None, + location: Optional[str] = None, + ) -> None: + super().__init__(message) + self.code = code + self.details = details + self.location = location + + @classmethod + def from_rpc_error( + cls, rpc_error: Exception, location: Optional[str] = None + ) -> "RipsError": + """Build a RipsError from a grpc.RpcError-like object.""" + code_fn = getattr(rpc_error, "code", None) + details_fn = getattr(rpc_error, "details", None) + code = code_fn() if callable(code_fn) else None + details = details_fn() if callable(details_fn) else "" + + parts = ["ResInsight gRPC call failed"] + if code is not None: + parts.append(f"({code})") + if details: + parts.append(f"- {details}") + if location: + parts.append(f"[server={location}]") + return cls( + " ".join(parts), + code=code, + details=details or None, + location=location, + ) diff --git a/GrpcInterface/Python/rips/grpc_retry_interceptor.py b/GrpcInterface/Python/rips/grpc_retry_interceptor.py index 6ca2d800e9..6e49a9cb14 100644 --- a/GrpcInterface/Python/rips/grpc_retry_interceptor.py +++ b/GrpcInterface/Python/rips/grpc_retry_interceptor.py @@ -9,9 +9,11 @@ class RetryOnRpcErrorClientInterceptor( *, retry_policy, status_for_retry, + should_abort=None, ): self.retry_policy = retry_policy self.status_for_retry = status_for_retry + self.should_abort = should_abort def _intercept_call(self, continuation, client_call_details, request_or_iterator): for retry_num in range(self.retry_policy.num_retries()): @@ -29,6 +31,12 @@ class RetryOnRpcErrorClientInterceptor( ): return response + # Bail out immediately if an external signal (e.g. the + # heartbeat) says the server is gone — otherwise we'd + # spend the full retry budget on a defunct process. + if self.should_abort is not None and self.should_abort(): + return response + self.retry_policy.sleep(retry_num) else: return response diff --git a/GrpcInterface/Python/rips/instance.py b/GrpcInterface/Python/rips/instance.py index 443ef1e379..2884eadf6d 100644 --- a/GrpcInterface/Python/rips/instance.py +++ b/GrpcInterface/Python/rips/instance.py @@ -15,6 +15,7 @@ import signal import sys import json import subprocess +import threading import grpc @@ -32,7 +33,7 @@ from .grpc_retry_interceptor import RetryOnRpcErrorClientInterceptor from .generated.generated_classes import CommandRouter from .exception import RipsError -from typing import List, Optional, Tuple +from typing import Callable, List, Optional, Tuple from pathlib import Path logger = logging.getLogger(__name__) @@ -50,6 +51,11 @@ class Instance: Set when creating an instance and updated when opening/closing projects. """ + _last_version_check_error: Optional[grpc.RpcError] + _connection_lost: bool + _heartbeat_thread: Optional[threading.Thread] + _heartbeat_stop: Optional[threading.Event] + @staticmethod def __is_port_in_use(port: int) -> bool: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as my_socket: @@ -109,6 +115,7 @@ class Instance: launch_port: int = 0, init_timeout: int = 300, command_line_parameters: List[str] = [], + enable_heartbeat: bool = True, ) -> Optional[Instance]: """Launch a new Instance of ResInsight. This requires the environment variable RESINSIGHT_EXECUTABLE to be set or the parameter resinsight_executable to be provided. @@ -124,6 +131,9 @@ class Instance: If anything else, ResInsight will try to launch with the specified portnumber. init_timeout: Number of seconds to wait for initialization before timing out. command_line_parameters(list): Additional parameters as string entries in the list. + enable_heartbeat(bool): If True (default), a background thread pings the + server periodically and aborts pending RPCs if it dies. Disable on + slow boxes where false positives matter (long GC pauses, debugger). Returns: Instance: an instance object if it worked. None if not. """ @@ -192,12 +202,20 @@ class Instance: Instance.__kill_process(pid) raise RipsError("Unable to read port number. Launch failed.") else: - instance = Instance(port=port, launched=True) + instance = Instance( + port=port, + launched=True, + enable_heartbeat=enable_heartbeat, + ) return instance return None @staticmethod - def find(start_port: int = 50051, end_port: int = 50071) -> Optional[Instance]: + def find( + start_port: int = 50051, + end_port: int = 50071, + enable_heartbeat: bool = True, + ) -> Optional[Instance]: """Search for an existing Instance of ResInsight by testing ports. By default we search from port 50051 to 50071 or if the environment @@ -207,6 +225,9 @@ class Instance: Args: start_port (int): start searching from this port end_port (int): search up to but not including this port + enable_heartbeat(bool): If True (default), a background thread pings the + server periodically and aborts pending RPCs if it dies. Disable on + slow boxes where false positives matter (long GC pauses, debugger). """ port_env = os.environ.get("RESINSIGHT_GRPC_PORT") if port_env: @@ -219,7 +240,7 @@ class Instance: if Instance.__is_port_in_use(try_port) and Instance.__is_valid_port( try_port ): - return Instance(port=try_port) + return Instance(port=try_port, enable_heartbeat=enable_heartbeat) raise RipsError( f"Could not find any ResInsight instances responding between ports {start_port} and {end_port}" @@ -236,18 +257,34 @@ class Instance: minor_version_ok = self.minor_version() == int( RiaVersionInfo.RESINSIGHT_MINOR_VERSION ) + self._last_version_check_error = None return True, major_version_ok and minor_version_ok - except grpc.RpcError: + except grpc.RpcError as exception: + self._last_version_check_error = exception return False, False - def __init__(self, port: int = 50051, launched: bool = False) -> None: + def __init__( + self, + port: int = 50051, + launched: bool = False, + enable_heartbeat: bool = True, + ) -> None: """Attempts to connect to ResInsight at a specific port on localhost Args: port(int): port number + launched(bool): True if this Python process launched ResInsight. + enable_heartbeat(bool): If True (default), a background thread + pings the server periodically. On detection of a dead server + the channel is closed so any in-flight RPC unblocks + immediately with a :class:`RipsError`. """ self.location: str = "localhost:" + str(port) self.port: int = port + self._last_version_check_error = None + self._connection_lost = False + self._heartbeat_thread = None + self._heartbeat_stop = None self.channel = grpc.insecure_channel( self.location, options=[("grpc.enable_http_proxy", False)] @@ -267,6 +304,7 @@ class Instance: min_backoff=100, max_backoff=5000, max_num_retries=20 ), status_for_retry=(grpc.StatusCode.UNAVAILABLE,), + should_abort=lambda: self._connection_lost, ), ) @@ -286,6 +324,9 @@ class Instance: path = os.getcwd() self.set_start_dir(path=path) + if enable_heartbeat: + self.start_heartbeat() + def _check_connection_and_version( self, channel: grpc.Channel, launched: bool, location: str ) -> None: @@ -303,26 +344,128 @@ class Instance: connection_ok, version_ok = self.__check_version() if not connection_ok: + last_error = self._last_version_check_error + cause_text = "" + code = None + details = None + if last_error is not None: + code_fn = getattr(last_error, "code", None) + details_fn = getattr(last_error, "details", None) + if callable(code_fn): + code = code_fn() + if callable(details_fn): + details = details_fn() + if code is not None or details: + cause_text = f" (gRPC {code}: {details or ''})" + if self.launched: - raise Exception( - "Error: Could not connect to resinsight at ", - location, - ".", - retry_policy.time_out_message(), - ) - raise Exception("Error: Could not connect to resinsight at ", location) + raise RipsError( + f"Could not connect to ResInsight at {location}.{cause_text} " + f"{retry_policy.time_out_message()}", + code=code, + details=details, + location=location, + ) from last_error + raise RipsError( + f"Could not connect to ResInsight at {location}.{cause_text}", + code=code, + details=details, + location=location, + ) from last_error if not version_ok: - raise Exception( - "Error: Wrong Version of ResInsight at ", - location, - "Executable : " + self.version_string(), - " ", - "rips : " + self.client_version_string(), + raise RipsError( + f"Wrong Version of ResInsight at {location}. " + f"Executable: {self.version_string()}, " + f"rips: {self.client_version_string()}", + location=location, ) def __version_message(self) -> App_pb2.Version: return self.app.GetVersion(Empty()) + def start_heartbeat( + self, + interval_sec: float = 5.0, + deadline_sec: float = 2.0, + on_failure: Optional[Callable[["RipsError"], None]] = None, + ) -> None: + """Start a background thread that periodically pings ResInsight. + + On the first failed ping the instance is marked as + ``connection_lost``, the underlying gRPC channel is closed so + any in-flight calls fail fast with ``UNAVAILABLE``, and the + retry interceptor is told to stop retrying. Subsequent API + calls that go through :meth:`check_alive` raise + :class:`RipsError` with the captured cause. + + Heartbeat is opt-in. Existing scripts are unaffected unless + they call this method. + + Args: + interval_sec: Seconds between pings. + deadline_sec: Per-ping deadline. Pings exceeding this are + treated as failures. + on_failure: Optional callback invoked once when the + heartbeat detects a lost connection. Receives a + :class:`RipsError`. + """ + if self._heartbeat_thread is not None and self._heartbeat_thread.is_alive(): + return + + stop_event = threading.Event() + self._heartbeat_stop = stop_event + + def _run() -> None: + while not stop_event.is_set(): + try: + self.app.GetVersion(Empty(), timeout=deadline_sec) + except grpc.RpcError as exc: + self._connection_lost = True + # Close the channel so any pending RPC unblocks + # immediately with UNAVAILABLE instead of waiting + # for TCP keepalive (which can take many minutes). + try: + self.channel.close() + except Exception: + logger.exception("Failed to close gRPC channel from heartbeat") + err = RipsError.from_rpc_error(exc, location=self.location) + if on_failure is not None: + try: + on_failure(err) + except Exception: + logger.exception("Heartbeat on_failure callback raised") + return + stop_event.wait(interval_sec) + + thread = threading.Thread(target=_run, name="rips-heartbeat", daemon=True) + self._heartbeat_thread = thread + thread.start() + + def stop_heartbeat(self) -> None: + """Stop the background heartbeat thread, if running.""" + if self._heartbeat_stop is not None: + self._heartbeat_stop.set() + if self._heartbeat_thread is not None: + self._heartbeat_thread.join(timeout=2.0) + self._heartbeat_thread = None + self._heartbeat_stop = None + + def check_alive(self) -> None: + """Raise :class:`RipsError` if the heartbeat has flagged a + lost connection. Cheap to call before issuing API requests.""" + if self._connection_lost: + raise RipsError( + f"ResInsight at {self.location} is no longer responding " + "(detected by heartbeat)", + location=self.location, + ) + + def __del__(self): + try: + self.stop_heartbeat() + except Exception: + pass + def set_start_dir(self, path: str): """Set current start directory diff --git a/GrpcInterface/Python/rips/pdmobject.py b/GrpcInterface/Python/rips/pdmobject.py index dc109de6ef..782dc562be 100644 --- a/GrpcInterface/Python/rips/pdmobject.py +++ b/GrpcInterface/Python/rips/pdmobject.py @@ -42,7 +42,9 @@ def add_method(cls: C) -> Callable[[F], F]: try: return func(*args, **kwargs) except grpc.RpcError as e: - raise RipsError(e.details()) from None + raise RipsError( + e.details(), code=e.code(), details=e.details() + ) from None # Explicitly preserve signature for Sphinx documentation wrapper.__name__ = func.__name__ @@ -558,7 +560,9 @@ class PdmObjectBase: try: self._pdm_object_stub.CallPdmObjectMethod(request) except grpc.RpcError as exc: - raise RipsError("%s" % exc.details()) from None + raise RipsError( + "%s" % exc.details(), code=exc.code(), details=exc.details() + ) from None def _call_pdm_method_return_value( self, method_name: str, class_definition: Type[PdmObjectT], **kwargs: Any @@ -577,7 +581,9 @@ class PdmObjectBase: pdm_object = class_definition(pb2_object=pb2_object, channel=self.channel()) return pdm_object except grpc.RpcError as exc: - raise RipsError("%s" % exc.details()) from None + raise RipsError( + "%s" % exc.details(), code=exc.code(), details=exc.details() + ) from None def _call_pdm_method_return_optional_value( self, method_name: str, class_definition: Type[PdmObjectT], **kwargs: Any @@ -607,7 +613,9 @@ class PdmObjectBase: return pdm_object except grpc.RpcError as exc: - raise RipsError("%s" % exc.details()) from None + raise RipsError( + "%s" % exc.details(), code=exc.code(), details=exc.details() + ) from None def update(self) -> None: """Sync all fields from the Python Object to ResInsight diff --git a/GrpcInterface/Python/rips/tests/test_error_reporting.py b/GrpcInterface/Python/rips/tests/test_error_reporting.py new file mode 100644 index 0000000000..0de8b0356c --- /dev/null +++ b/GrpcInterface/Python/rips/tests/test_error_reporting.py @@ -0,0 +1,72 @@ +import sys +import os + +import grpc +import pytest + +sys.path.insert(1, os.path.join(sys.path[0], "../../")) +import rips # noqa: E402 +from rips.exception import RipsError # noqa: E402 + +import dataroot # noqa: E402 + + +class _FakeRpcError(grpc.RpcError): + def __init__(self, code, details): + self._code = code + self._details = details + + def code(self): + return self._code + + def details(self): + return self._details + + +def test_rips_error_is_constructible_from_message_only(): + err = RipsError("boom") + assert str(err) == "boom" + assert err.code is None + assert err.details is None + assert err.location is None + + +def test_rips_error_carries_structured_fields(): + err = RipsError("boom", code=grpc.StatusCode.NOT_FOUND, details="x", location="h:1") + assert err.code == grpc.StatusCode.NOT_FOUND + assert err.details == "x" + assert err.location == "h:1" + + +def test_rips_error_from_rpc_error_extracts_code_and_details(): + rpc = _FakeRpcError(grpc.StatusCode.INVALID_ARGUMENT, "bad arg") + err = RipsError.from_rpc_error(rpc, location="localhost:50051") + assert err.code == grpc.StatusCode.INVALID_ARGUMENT + assert err.details == "bad arg" + assert err.location == "localhost:50051" + text = str(err) + assert "INVALID_ARGUMENT" in text + assert "bad arg" in text + assert "localhost:50051" in text + + +def test_pdm_method_error_includes_code_and_details(rips_instance, initialize_test): + # Trigger a server-side INVALID_ARGUMENT on a real gRPC round-trip and + # confirm the new structured fields on RipsError are populated. + case_path = dataroot.PATH + "/Case_with_10_timesteps/Real0/BRUGGE_0000.EGRID" + rips_instance.project.load_case(path=case_path) + + surface_collection = rips_instance.project.descendants(rips.SurfaceCollection)[0] + + with pytest.raises(RipsError) as excinfo: + surface_collection.new_regular_surface(increment_x=-1.0) + + err = excinfo.value + assert err.code is not None and err.code != grpc.StatusCode.OK + assert err.details and "Invalid increment X" in err.details + + +def test_heartbeat_starts_and_stops(rips_instance, initialize_test): + rips_instance.start_heartbeat(interval_sec=0.5, deadline_sec=1.0) + rips_instance.stop_heartbeat() + rips_instance.check_alive() diff --git a/GrpcInterface/RiaGrpcAppService.cpp b/GrpcInterface/RiaGrpcAppService.cpp index 4ceb8add79..abec461eab 100644 --- a/GrpcInterface/RiaGrpcAppService.cpp +++ b/GrpcInterface/RiaGrpcAppService.cpp @@ -88,8 +88,9 @@ grpc::Status if ( commandRouter ) { copyPdmObjectFromCafToRips( commandRouter, reply ); + return grpc::Status::OK; } - return grpc::Status::OK; + return grpc::Status( grpc::INTERNAL, "Command router is not available" ); } static bool RiaGrpcAppInfoService_init = diff --git a/GrpcInterface/RiaGrpcCaseService.cpp b/GrpcInterface/RiaGrpcCaseService.cpp index 1e74209dbe..3ba8ab8de5 100644 --- a/GrpcInterface/RiaGrpcCaseService.cpp +++ b/GrpcInterface/RiaGrpcCaseService.cpp @@ -176,8 +176,9 @@ grpc::Status RiaGrpcCaseService::GetPdmObject( grpc::ServerContext* context, if ( rimCase ) { copyPdmObjectFromCafToRips( rimCase, reply ); + return grpc::Status::OK; } - return grpc::Status::OK; + return grpc::Status( grpc::NOT_FOUND, "Case not found" ); } //-------------------------------------------------------------------------------------------------- diff --git a/GrpcInterface/RiaGrpcProjectService.cpp b/GrpcInterface/RiaGrpcProjectService.cpp index 5e8efa0a34..3ab6cafd73 100644 --- a/GrpcInterface/RiaGrpcProjectService.cpp +++ b/GrpcInterface/RiaGrpcProjectService.cpp @@ -199,8 +199,9 @@ grpc::Status if ( project ) { copyPdmObjectFromCafToRips( project, reply ); + return grpc::Status::OK; } - return grpc::Status::OK; + return grpc::Status( grpc::NOT_FOUND, "No active project" ); } //--------------------------------------------------------------------------------------------------