#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.
This commit is contained in:
Kristian Bendiksen
2026-05-04 09:54:08 +02:00
parent 605662062c
commit db871296cd
9 changed files with 308 additions and 28 deletions
+2 -1
View File
@@ -58,6 +58,7 @@ from .resinsight_classes import (
WbsParameters as WbsParameters, WbsParameters as WbsParameters,
) )
from .exception import RipsError
from .grid import Grid as Grid from .grid import Grid as Grid
from .project import Project as Project from .project import Project as Project
from .pdmobject import add_method from .pdmobject import add_method
@@ -85,7 +86,7 @@ def __grid_count(self) -> int:
except grpc.RpcError as exception: except grpc.RpcError as exception:
if exception.code() == grpc.StatusCode.NOT_FOUND: if exception.code() == grpc.StatusCode.NOT_FOUND:
return 0 return 0
return 0 raise RipsError.from_rpc_error(exception) from exception
@add_method(Case) @add_method(Case)
+46 -1
View File
@@ -1,2 +1,47 @@
from typing import Optional
class RipsError(Exception): 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,
)
@@ -9,9 +9,11 @@ class RetryOnRpcErrorClientInterceptor(
*, *,
retry_policy, retry_policy,
status_for_retry, status_for_retry,
should_abort=None,
): ):
self.retry_policy = retry_policy self.retry_policy = retry_policy
self.status_for_retry = status_for_retry self.status_for_retry = status_for_retry
self.should_abort = should_abort
def _intercept_call(self, continuation, client_call_details, request_or_iterator): def _intercept_call(self, continuation, client_call_details, request_or_iterator):
for retry_num in range(self.retry_policy.num_retries()): for retry_num in range(self.retry_policy.num_retries()):
@@ -29,6 +31,12 @@ class RetryOnRpcErrorClientInterceptor(
): ):
return response 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) self.retry_policy.sleep(retry_num)
else: else:
return response return response
+162 -19
View File
@@ -15,6 +15,7 @@ import signal
import sys import sys
import json import json
import subprocess import subprocess
import threading
import grpc import grpc
@@ -32,7 +33,7 @@ from .grpc_retry_interceptor import RetryOnRpcErrorClientInterceptor
from .generated.generated_classes import CommandRouter from .generated.generated_classes import CommandRouter
from .exception import RipsError from .exception import RipsError
from typing import List, Optional, Tuple from typing import Callable, List, Optional, Tuple
from pathlib import Path from pathlib import Path
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -50,6 +51,11 @@ class Instance:
Set when creating an instance and updated when opening/closing projects. 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 @staticmethod
def __is_port_in_use(port: int) -> bool: def __is_port_in_use(port: int) -> bool:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as my_socket: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as my_socket:
@@ -109,6 +115,7 @@ class Instance:
launch_port: int = 0, launch_port: int = 0,
init_timeout: int = 300, init_timeout: int = 300,
command_line_parameters: List[str] = [], command_line_parameters: List[str] = [],
enable_heartbeat: bool = True,
) -> Optional[Instance]: ) -> Optional[Instance]:
"""Launch a new Instance of ResInsight. This requires the environment variable """Launch a new Instance of ResInsight. This requires the environment variable
RESINSIGHT_EXECUTABLE to be set or the parameter resinsight_executable to be provided. 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. If anything else, ResInsight will try to launch with the specified portnumber.
init_timeout: Number of seconds to wait for initialization before timing out. init_timeout: Number of seconds to wait for initialization before timing out.
command_line_parameters(list): Additional parameters as string entries in the list. 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: Returns:
Instance: an instance object if it worked. None if not. Instance: an instance object if it worked. None if not.
""" """
@@ -192,12 +202,20 @@ class Instance:
Instance.__kill_process(pid) Instance.__kill_process(pid)
raise RipsError("Unable to read port number. Launch failed.") raise RipsError("Unable to read port number. Launch failed.")
else: else:
instance = Instance(port=port, launched=True) instance = Instance(
port=port,
launched=True,
enable_heartbeat=enable_heartbeat,
)
return instance return instance
return None return None
@staticmethod @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. """Search for an existing Instance of ResInsight by testing ports.
By default we search from port 50051 to 50071 or if the environment By default we search from port 50051 to 50071 or if the environment
@@ -207,6 +225,9 @@ class Instance:
Args: Args:
start_port (int): start searching from this port start_port (int): start searching from this port
end_port (int): search up to but not including 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") port_env = os.environ.get("RESINSIGHT_GRPC_PORT")
if port_env: if port_env:
@@ -219,7 +240,7 @@ class Instance:
if Instance.__is_port_in_use(try_port) and Instance.__is_valid_port( if Instance.__is_port_in_use(try_port) and Instance.__is_valid_port(
try_port try_port
): ):
return Instance(port=try_port) return Instance(port=try_port, enable_heartbeat=enable_heartbeat)
raise RipsError( raise RipsError(
f"Could not find any ResInsight instances responding between ports {start_port} and {end_port}" 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( minor_version_ok = self.minor_version() == int(
RiaVersionInfo.RESINSIGHT_MINOR_VERSION RiaVersionInfo.RESINSIGHT_MINOR_VERSION
) )
self._last_version_check_error = None
return True, major_version_ok and minor_version_ok 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 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 """Attempts to connect to ResInsight at a specific port on localhost
Args: Args:
port(int): port number 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.location: str = "localhost:" + str(port)
self.port: int = 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.channel = grpc.insecure_channel(
self.location, options=[("grpc.enable_http_proxy", False)] self.location, options=[("grpc.enable_http_proxy", False)]
@@ -267,6 +304,7 @@ class Instance:
min_backoff=100, max_backoff=5000, max_num_retries=20 min_backoff=100, max_backoff=5000, max_num_retries=20
), ),
status_for_retry=(grpc.StatusCode.UNAVAILABLE,), status_for_retry=(grpc.StatusCode.UNAVAILABLE,),
should_abort=lambda: self._connection_lost,
), ),
) )
@@ -286,6 +324,9 @@ class Instance:
path = os.getcwd() path = os.getcwd()
self.set_start_dir(path=path) self.set_start_dir(path=path)
if enable_heartbeat:
self.start_heartbeat()
def _check_connection_and_version( def _check_connection_and_version(
self, channel: grpc.Channel, launched: bool, location: str self, channel: grpc.Channel, launched: bool, location: str
) -> None: ) -> None:
@@ -303,26 +344,128 @@ class Instance:
connection_ok, version_ok = self.__check_version() connection_ok, version_ok = self.__check_version()
if not connection_ok: 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: if self.launched:
raise Exception( raise RipsError(
"Error: Could not connect to resinsight at ", f"Could not connect to ResInsight at {location}.{cause_text} "
location, f"{retry_policy.time_out_message()}",
".", code=code,
retry_policy.time_out_message(), details=details,
) location=location,
raise Exception("Error: Could not connect to resinsight at ", 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: if not version_ok:
raise Exception( raise RipsError(
"Error: Wrong Version of ResInsight at ", f"Wrong Version of ResInsight at {location}. "
location, f"Executable: {self.version_string()}, "
"Executable : " + self.version_string(), f"rips: {self.client_version_string()}",
" ", location=location,
"rips : " + self.client_version_string(),
) )
def __version_message(self) -> App_pb2.Version: def __version_message(self) -> App_pb2.Version:
return self.app.GetVersion(Empty()) 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): def set_start_dir(self, path: str):
"""Set current start directory """Set current start directory
+12 -4
View File
@@ -42,7 +42,9 @@ def add_method(cls: C) -> Callable[[F], F]:
try: try:
return func(*args, **kwargs) return func(*args, **kwargs)
except grpc.RpcError as e: 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 # Explicitly preserve signature for Sphinx documentation
wrapper.__name__ = func.__name__ wrapper.__name__ = func.__name__
@@ -558,7 +560,9 @@ class PdmObjectBase:
try: try:
self._pdm_object_stub.CallPdmObjectMethod(request) self._pdm_object_stub.CallPdmObjectMethod(request)
except grpc.RpcError as exc: 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( def _call_pdm_method_return_value(
self, method_name: str, class_definition: Type[PdmObjectT], **kwargs: Any 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()) pdm_object = class_definition(pb2_object=pb2_object, channel=self.channel())
return pdm_object return pdm_object
except grpc.RpcError as exc: 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( def _call_pdm_method_return_optional_value(
self, method_name: str, class_definition: Type[PdmObjectT], **kwargs: Any self, method_name: str, class_definition: Type[PdmObjectT], **kwargs: Any
@@ -607,7 +613,9 @@ class PdmObjectBase:
return pdm_object return pdm_object
except grpc.RpcError as exc: 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: def update(self) -> None:
"""Sync all fields from the Python Object to ResInsight """Sync all fields from the Python Object to ResInsight
@@ -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()
+2 -1
View File
@@ -88,8 +88,9 @@ grpc::Status
if ( commandRouter ) if ( commandRouter )
{ {
copyPdmObjectFromCafToRips( commandRouter, reply ); 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 = static bool RiaGrpcAppInfoService_init =
+2 -1
View File
@@ -176,8 +176,9 @@ grpc::Status RiaGrpcCaseService::GetPdmObject( grpc::ServerContext* context,
if ( rimCase ) if ( rimCase )
{ {
copyPdmObjectFromCafToRips( rimCase, reply ); copyPdmObjectFromCafToRips( rimCase, reply );
return grpc::Status::OK;
} }
return grpc::Status::OK; return grpc::Status( grpc::NOT_FOUND, "Case not found" );
} }
//-------------------------------------------------------------------------------------------------- //--------------------------------------------------------------------------------------------------
+2 -1
View File
@@ -199,8 +199,9 @@ grpc::Status
if ( project ) if ( project )
{ {
copyPdmObjectFromCafToRips( project, reply ); copyPdmObjectFromCafToRips( project, reply );
return grpc::Status::OK;
} }
return grpc::Status::OK; return grpc::Status( grpc::NOT_FOUND, "No active project" );
} }
//-------------------------------------------------------------------------------------------------- //--------------------------------------------------------------------------------------------------