mirror of
https://github.com/OPM/ResInsight.git
synced 2026-08-18 09:05:11 -05:00
Kristian Bendiksen
f8d180f48d
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.
592 lines
22 KiB
Python
592 lines
22 KiB
Python
# pylint: disable=no-self-use
|
|
"""
|
|
The main entry point for ResInsight connections
|
|
The Instance class contained have static methods launch and find for
|
|
creating connections to ResInsight
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import os
|
|
import socket
|
|
import logging
|
|
import time
|
|
import tempfile
|
|
import signal
|
|
import sys
|
|
import json
|
|
import subprocess
|
|
import threading
|
|
|
|
import grpc
|
|
|
|
import App_pb2
|
|
import App_pb2_grpc
|
|
import Commands_pb2
|
|
import Commands_pb2_grpc
|
|
from Definitions_pb2 import Empty
|
|
|
|
import RiaVersionInfo
|
|
|
|
from .project import Project
|
|
from .retry_policy import ExponentialBackoffRetryPolicy
|
|
from .grpc_retry_interceptor import RetryOnRpcErrorClientInterceptor
|
|
from .generated.generated_classes import CommandRouter
|
|
from .exception import RipsError
|
|
|
|
from typing import Callable, List, Optional, Tuple
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Instance:
|
|
"""The ResInsight Instance class. Use to launch or find existing ResInsight instances
|
|
|
|
Attributes:
|
|
launched (bool): Tells us whether the application was launched as a new process.
|
|
If the application was launched we may need to close it when exiting the script.
|
|
port (int): The port number used to connect to this instance.
|
|
commands (Commands): Command executor. Set when creating an instance.
|
|
project (Project): Current project in ResInsight.
|
|
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:
|
|
my_socket.settimeout(0.2)
|
|
return my_socket.connect_ex(("localhost", port)) == 0
|
|
|
|
@staticmethod
|
|
def __is_valid_port(port: int) -> bool:
|
|
location = "localhost:" + str(port)
|
|
channel = grpc.insecure_channel(
|
|
location, options=[("grpc.enable_http_proxy", False)]
|
|
)
|
|
app = App_pb2_grpc.AppStub(channel)
|
|
try:
|
|
app.GetVersion(Empty(), timeout=1)
|
|
except grpc.RpcError:
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def __read_port_number_from_file(file_path: str, max_attempts: int) -> int:
|
|
retry_count = 0
|
|
while not os.path.exists(file_path) and retry_count < max_attempts:
|
|
time.sleep(1)
|
|
retry_count = retry_count + 1
|
|
|
|
if os.path.isfile(file_path):
|
|
with open(file_path) as f:
|
|
value = f.readline()
|
|
return int(value)
|
|
|
|
if retry_count == max_attempts:
|
|
logger.warning(
|
|
"Waiting for port number file timed out after %d seconds. File: %s",
|
|
max_attempts,
|
|
file_path,
|
|
)
|
|
|
|
return -1
|
|
|
|
@staticmethod
|
|
def __kill_process(pid: int) -> None:
|
|
"""
|
|
Kill the process with a given pid.
|
|
"""
|
|
if hasattr(signal, "CTRL_C_EVENT"):
|
|
# windows - use SIGTERM for process termination
|
|
os.kill(pid, signal.SIGTERM)
|
|
else:
|
|
# linux/unix
|
|
os.kill(pid, signal.SIGKILL)
|
|
|
|
@staticmethod
|
|
def launch(
|
|
resinsight_executable: str = "",
|
|
console: bool = False,
|
|
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.
|
|
The RESINSIGHT_GRPC_PORT environment variable can be set to an alternative port number.
|
|
|
|
Args:
|
|
resinsight_executable (str): Path to a valid ResInsight executable. If set
|
|
will take precedence over what is provided in the RESINSIGHT_EXECUTABLE
|
|
environment variable.
|
|
console (bool): If True, launch as console application, without GUI.
|
|
launch_port(int): If 0, GRPC will find an available port.
|
|
If -1, use the default port 50051 or RESINSIGHT_GRPC_PORT
|
|
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.
|
|
"""
|
|
|
|
requested_port: int = 50051
|
|
port_env = os.environ.get("RESINSIGHT_GRPC_PORT")
|
|
if port_env:
|
|
requested_port = int(port_env)
|
|
if launch_port != -1:
|
|
requested_port = launch_port
|
|
|
|
if not resinsight_executable:
|
|
filename = Path(sys.prefix) / "share" / "rips" / "rips_config.json"
|
|
if filename.is_file():
|
|
f = open(filename)
|
|
data = json.load(f)
|
|
resinsight_executable = data["resinsight_executable"]
|
|
if resinsight_executable:
|
|
logger.info(
|
|
"In './share/rips/rips_config.json', found resinsight_executable: %s",
|
|
resinsight_executable,
|
|
)
|
|
|
|
if not resinsight_executable:
|
|
resinsight_executable_from_env = os.environ.get("RESINSIGHT_EXECUTABLE")
|
|
if not resinsight_executable_from_env:
|
|
raise RipsError(
|
|
"Could not launch ResInsight because the environment variable"
|
|
" RESINSIGHT_EXECUTABLE is not set"
|
|
)
|
|
else:
|
|
resinsight_executable = resinsight_executable_from_env
|
|
|
|
# Check if executable file exists
|
|
if not os.path.isfile(resinsight_executable):
|
|
raise RipsError(
|
|
f"ResInsight executable not found at path: {resinsight_executable}"
|
|
)
|
|
|
|
logger.info("Trying to launch %s", resinsight_executable)
|
|
with tempfile.TemporaryDirectory() as tmp_dir_path:
|
|
port_number_file = tmp_dir_path + "/portnumber.txt"
|
|
parameters: List[str] = [
|
|
resinsight_executable,
|
|
"--server",
|
|
str(requested_port),
|
|
"--portnumberfile",
|
|
str(port_number_file),
|
|
] + command_line_parameters
|
|
if console:
|
|
logger.info("Launching as console app")
|
|
parameters.append("--console")
|
|
|
|
# Stringify all parameters
|
|
for i in range(0, len(parameters)):
|
|
parameters[i] = str(parameters[i])
|
|
|
|
process = subprocess.Popen(parameters)
|
|
pid = process.pid
|
|
if pid:
|
|
port = Instance.__read_port_number_from_file(
|
|
port_number_file, init_timeout
|
|
)
|
|
if port == -1:
|
|
# Need to kill the process using PID since there is no GRPC connection to use.
|
|
Instance.__kill_process(pid)
|
|
raise RipsError("Unable to read port number. Launch failed.")
|
|
else:
|
|
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,
|
|
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
|
|
variable RESINSIGHT_GRPC_PORT is set we search
|
|
RESINSIGHT_GRPC_PORT to RESINSIGHT_GRPC_PORT+20
|
|
|
|
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:
|
|
logger.info("Got port %s from environment", port_env)
|
|
start_port = int(port_env)
|
|
end_port = start_port + 20
|
|
|
|
for try_port in range(start_port, end_port):
|
|
logger.debug("Trying port %d", try_port)
|
|
if Instance.__is_port_in_use(try_port) and Instance.__is_valid_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}"
|
|
)
|
|
|
|
def __execute_command(self, **command_params):
|
|
return self.commands.Execute(Commands_pb2.CommandParams(**command_params))
|
|
|
|
def __check_version(self) -> Tuple[bool, bool]:
|
|
try:
|
|
major_version_ok = self.major_version() == int(
|
|
RiaVersionInfo.RESINSIGHT_MAJOR_VERSION
|
|
)
|
|
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 as exception:
|
|
self._last_version_check_error = exception
|
|
return False, False
|
|
|
|
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)]
|
|
)
|
|
self.launched = launched
|
|
self.commands = Commands_pb2_grpc.CommandsStub(self.channel)
|
|
|
|
# Main version check package
|
|
self.app = App_pb2_grpc.AppStub(self.channel)
|
|
|
|
self._check_connection_and_version(self.channel, launched, self.location)
|
|
|
|
# Intercept UNAVAILABLE errors and retry on failures
|
|
interceptors = (
|
|
RetryOnRpcErrorClientInterceptor(
|
|
retry_policy=ExponentialBackoffRetryPolicy(
|
|
min_backoff=100, max_backoff=5000, max_num_retries=20
|
|
),
|
|
status_for_retry=(grpc.StatusCode.UNAVAILABLE,),
|
|
should_abort=lambda: self._connection_lost,
|
|
),
|
|
)
|
|
|
|
intercepted_channel = grpc.intercept_channel(self.channel, *interceptors)
|
|
|
|
# Recreate command stubs with the retry policy
|
|
self.commands = Commands_pb2_grpc.CommandsStub(intercepted_channel)
|
|
|
|
# Service packages
|
|
self.project = Project.create(intercepted_channel)
|
|
|
|
# Command Router object used as entry point for independent processing functions
|
|
self.command_router = CommandRouter(
|
|
self.app.GetPdmObject(Empty()), intercepted_channel
|
|
)
|
|
|
|
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:
|
|
connection_ok = False
|
|
version_ok = False
|
|
|
|
retry_policy = ExponentialBackoffRetryPolicy()
|
|
if self.launched:
|
|
for num_tries in range(0, retry_policy.num_retries()):
|
|
connection_ok, version_ok = self.__check_version()
|
|
if connection_ok:
|
|
break
|
|
retry_policy.sleep(num_tries)
|
|
else:
|
|
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 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 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
|
|
|
|
Arguments:
|
|
path (str): path to directory
|
|
|
|
"""
|
|
return self.__execute_command(
|
|
setStartDir=Commands_pb2.FilePathRequest(path=path)
|
|
)
|
|
|
|
def set_export_folder(
|
|
self, export_type: str, path: str, create_folder: bool = False
|
|
):
|
|
"""
|
|
Set the export folder used for all export functions
|
|
|
|
**Parameters**::
|
|
|
|
Parameter | Description | Type
|
|
---------------- | -------------------------------------------- | -----
|
|
export_type | String specifying what to export | String
|
|
path | Path to folder | String
|
|
create_folder | Create folder if it doesn't exist? | Boolean
|
|
|
|
**Enum export_type**::
|
|
|
|
Option | Description
|
|
--------------- | ------------
|
|
"COMPLETIONS" |
|
|
"SNAPSHOTS" |
|
|
"PROPERTIES" |
|
|
"STATISTICS" |
|
|
|
|
"""
|
|
return self.__execute_command(
|
|
setExportFolder=Commands_pb2.SetExportFolderRequest(
|
|
type=export_type, path=path, createFolder=create_folder
|
|
)
|
|
)
|
|
|
|
def set_main_window_size(self, width: int, height: int):
|
|
"""
|
|
Set the main window size in pixels
|
|
|
|
**Parameters**::
|
|
|
|
Parameter | Description | Type
|
|
--------- | ---------------- | -----
|
|
width | Width in pixels | Integer
|
|
height | Height in pixels | Integer
|
|
|
|
"""
|
|
return self.__execute_command(
|
|
setMainWindowSize=Commands_pb2.SetWindowSizeParams(
|
|
width=width, height=height
|
|
)
|
|
)
|
|
|
|
def set_plot_window_size(self, width: int, height: int):
|
|
"""
|
|
Set the plot window size in pixels
|
|
|
|
**Parameters**::
|
|
|
|
Parameter | Description | Type
|
|
--------- | ---------------- | -----
|
|
width | Width in pixels | Integer
|
|
height | Height in pixels | Integer
|
|
"""
|
|
return self.__execute_command(
|
|
setPlotWindowSize=Commands_pb2.SetWindowSizeParams(
|
|
width=width, height=height
|
|
)
|
|
)
|
|
|
|
def major_version(self) -> int:
|
|
"""Get an integer with the major version number"""
|
|
return int(self.__version_message().major_version)
|
|
|
|
def minor_version(self) -> int:
|
|
"""Get an integer with the minor version number"""
|
|
return int(self.__version_message().minor_version)
|
|
|
|
def patch_version(self) -> int:
|
|
"""Get an integer with the patch version number"""
|
|
return int(self.__version_message().patch_version)
|
|
|
|
def version_string(self) -> str:
|
|
"""Get a full version string, i.e. 2019.04.01"""
|
|
return (
|
|
str(self.major_version())
|
|
+ "."
|
|
+ str(self.minor_version())
|
|
+ "."
|
|
+ str(self.patch_version())
|
|
)
|
|
|
|
def client_version_string(self) -> str:
|
|
"""Get a full version string, i.e. 2019.04.01"""
|
|
version_string: str = RiaVersionInfo.RESINSIGHT_MAJOR_VERSION + "."
|
|
version_string += RiaVersionInfo.RESINSIGHT_MINOR_VERSION + "."
|
|
version_string += RiaVersionInfo.RESINSIGHT_PATCH_VERSION
|
|
return version_string
|
|
|
|
def exit(self):
|
|
"""Tell ResInsight instance to quit"""
|
|
logger.info("Telling ResInsight to Exit")
|
|
return self.app.Exit(Empty())
|
|
|
|
def is_console(self) -> bool:
|
|
"""Returns true if the connected ResInsight instance is a console app"""
|
|
return bool(
|
|
self.app.GetRuntimeInfo(Empty()).app_type
|
|
== App_pb2.ApplicationTypeEnum.Value("CONSOLE_APPLICATION")
|
|
)
|
|
|
|
def is_gui(self) -> bool:
|
|
"""Returns true if the connected ResInsight instance is a GUI app"""
|
|
return bool(
|
|
self.app.GetRuntimeInfo(Empty()).app_type
|
|
== App_pb2.ApplicationTypeEnum.Value("GUI_APPLICATION")
|
|
)
|