Add client-side rich error interceptor

src/rich_error_handling_overview.md → docs/rich_error_handling_overview.md

@@ -16,17 +16,18 @@ Every custom `drunc` exception (`DruncException`) is built to automatically gene
 * **`reason`:** A machine-readable string to categorise the error, such as `NOT_IMPLEMENTED` or `COMMAND_ERROR`.
 * **`domain`:** The component or part that generated the error (e.g., `drunc`, `ProcessManager.boot`).
 * **`details`:** Human-readable context (e.g., "DUNEDAQ_DB_PATH is not set").
+
 ### ii. Specialised Error Details
 
-Sometimes, the standard `ErrorInfo` isn't structured enough. gRPC provides specialised Protobuf messages for specific failure states, such as `PreconditionFailure`, `BadRequest`, or `ResourceInfo` and [more](https://github.com/googleapis/googleapis/blob/master/google/rpc/error_details.proto). 
+Sometimes, the standard `ErrorInfo` isn't structured enough. gRPC provides specialised Protobuf messages for specific failure states, such as `PreconditionFailure`, `BadRequest`, or `ResourceInfo` and [more](https://github.com/googleapis/googleapis/blob/master/google/rpc/error_details.proto).
 
 You can append these specialised Protobuf objects to the error by overriding the `@property def specialised_details(self)` method in the custom exception.
 
 #### Example:  PreconditionFailure
+
 If a service fails to start you could use `PreconditionFailure.
 
 ```python
-
 class DruncSetupException(DruncException):
     grpc_error_code = code_pb2.FAILED_PRECONDITION
     reason = "SETUP_FAILED"
@@ -44,45 +45,69 @@ class DruncSetupException(DruncException):
             ]
         )
         return [precond]
-
 ```
+
 ## 2. Server-Side - Throwing Errors
 
 You do not need to manually pack Protobuf objects. Simply raise a subclass of `DruncException`.
 
 ### Example: Raising an existing exception
+
 ```python
 from drunc.exceptions import DruncSetupException
 
 # The exception automatically handles formatting the reason, domain, and details.
 if not config_files:
     raise DruncSetupException(
         message="Config files missing",
-        details=f"No configuration files found in {search_paths}"
+        details=f"No configuration files found in {search_paths}",
     )
 ```
+
 You can overwrite any of the `message`, `grpc_error_code`,`details`,`reason` and `domain`fields. If not explicitly provided, they will automatically fall back to the standard default values defined in the base error.
 
-## 3. Client-Side -  Catching Errors
+## 3. Server-Side - Server Interceptor
 
-**Provisional: this will change with the implementation of a `ClientInterceptor`**
+`RichErrorServerInterceptor` is a `grpc.ServerInterceptor` (defined in `src/drunc/utils/grpc_utils.py`) that wraps every unary-unary handler in a try/except block. When a `DruncException` is caught, it calls `abort_with_rich_details()` to pack the `ErrorInfo` and any specialised details into the gRPC trailing metadata before aborting the call. Non-`DruncException` errors and streaming calls are passed through unchanged.
 
-Use `extract_grpc_rich_error` to unpack the rich metadata:
+Implementing the interceptor:
 
 ```python
-try:
-    response = stub.boot(request, timeout=60)
-except grpc.RpcError as e:
-    try:
-        error_details = extract_grpc_rich_error(e)
-        log.error(error_details)
-    except Exception as extraction_error:
-        log.debug(f"Could not extract rich error: {extraction_error}")
-
-    handle_grpc_error(e)
+import grpc
+from drunc.utils.grpc_utils import RichErrorServerInterceptor
+
+server = grpc.server(
+    futures.ThreadPoolExecutor(max_workers=10),
+    interceptors=[RichErrorServerInterceptor()],
+)
 ```
 
-## 4. Testing gRPC errors
+## 4. Client-Side - Catching Errors with a Client Interceptor
+
+Client-side rich error handling is done automatically by the `RichErrorClientInterceptor` (located in `src/drunc/utils/grpc_utils.py`).
+
+Because gRPC network calls are asynchronous, a failed request doesn't throw an error the moment it is sent. Instead, the `RpcError` is raised when the application actually tries to read the data by calling `.result()` on the response `Future`. `_GRPCCallWrapper` wraps the `Future` returned by `continuation` and delays error handling until `.result()` or `.exception()` is actually called by the application.
+
+If the interceptor can't find any rich metadata or if extraction fails, it simply logs a debug warning and passes the standard gRPC error along unchanged.
+
+### Setting up a client with the interceptor
+
+```python
+self.log = get_logger("process_manager_driver", rich_handler=True)
+self.address = address
+options = [
+    ("grpc.keepalive_time_ms", 60000)  # pings the server every 60 seconds
+]
+raw_channel = grpc.insecure_channel(self.address, options=options)
+rich_interceptor = RichErrorClientInterceptor(logger=self.log)
+self.channel = grpc.intercept_channel(raw_channel, rich_interceptor)
+self.stub = ProcessManagerStub(self.channel)
+self.token = copy_token(token)
+```
+
+Once the channel is set up with the interceptor, **no extra try/except blocks are needed** to extract the rich error. Any `grpc.RpcError` that contains rich errors will be logged automatically before being re-raised.
+
+## 5. Testing gRPC errors
 
 Because gRPC status.details is a dynamic list, you need to iterate and check the type using `.Is()`.
 

src/drunc/process_manager/process_manager_driver.py

@@ -37,9 +37,8 @@
 from drunc.process_manager.oks_parser import get_full_db_path
 from drunc.process_manager.utils import format_hostname, get_log_path, get_rte_script
 from drunc.utils.grpc_utils import (
+    RichErrorClientInterceptor,
     copy_token,
-    extract_grpc_rich_error,
-    handle_grpc_error,
 )
 from drunc.utils.utils import (
     file_is_read_only,
@@ -63,7 +62,9 @@ def __init__(self, address: str, token: Token):
         options = [
             ("grpc.keepalive_time_ms", 60000)  # pings the server every 60 seconds
         ]
-        self.channel = grpc.insecure_channel(self.address, options=options)
+        raw_channel = grpc.insecure_channel(self.address, options=options)
+        rich_interceptor = RichErrorClientInterceptor(logger=self.log)
+        self.channel = grpc.intercept_channel(raw_channel, rich_interceptor)
         self.stub = ProcessManagerStub(self.channel)
         self.token = copy_token(token)
 
@@ -207,25 +208,13 @@ def boot(
                 )
                 touch_and_chmod(opmon_file)
 
-            try:
-                response = self.stub.boot(request, timeout=timeout)
-                self.log.info(
-                    f"Booted '{request.process_description.metadata.name}' "
-                    f"from session '{request.process_description.metadata.session}' "
-                    f"with UUID {response.values[0].uuid.uuid} on host {request.process_description.metadata.hostname}"
-                )
-                yield response
-
-            except grpc.RpcError as e:
-                try:
-                    error_details = extract_grpc_rich_error(e)
-                    self.log.error(error_details)
-                except Exception as extraction_error:
-                    self.log.debug(
-                        f"Could not extract rich error details from gRPC error: {extraction_error}",
-                        exc_info=True,
-                    )
-                handle_grpc_error(e)
+            response = self.stub.boot(request, timeout=timeout)
+            self.log.info(
+                f"Booted '{request.process_description.metadata.name}' "
+                f"from session '{request.process_description.metadata.session}' "
+                f"with UUID {response.values[0].uuid.uuid} on host {request.process_description.metadata.hostname}"
+            )
+            yield response
 
         # Step 7: discover segment root controller
         self._discover_controller(
@@ -833,20 +822,8 @@ def dummy_boot(
             )
             self.log.debug(f"{request=}\n\n")
 
-            try:
-                response = self.stub.boot(request, timeout=timeout)
-                yield response
-
-            except grpc.RpcError as e:
-                try:
-                    error_details = extract_grpc_rich_error(e)
-                    self.log.error(error_details)
-                except Exception as extraction_error:
-                    self.log.debug(
-                        f"Could not extract rich error details from gRPC error: {extraction_error}",
-                        exc_info=True,
-                    )
-                handle_grpc_error(e)
+            response = self.stub.boot(request, timeout=timeout)
+            yield response
 
     def _prepare_exec_and_args_dummy_boot(self, sleep: int, n_sleeps: int) -> list:
         args = [
@@ -890,18 +867,7 @@ def terminate(
         request = Request(token=copy_token(self.token))
         msg = f"[green]{request.token.user_name}[/green] sent terminate"
         self.log.info(msg)
-        try:
-            response = self.stub.terminate(request, timeout=timeout)
-        except grpc.RpcError as e:
-            try:
-                error_details = extract_grpc_rich_error(e)
-                self.log.error(error_details)
-            except Exception as extraction_error:
-                self.log.debug(
-                    f"Could not extract rich error details from gRPC error: {extraction_error}",
-                    exc_info=True,
-                )
-            handle_grpc_error(e)
+        response = self.stub.terminate(request, timeout=timeout)
 
         return response
 
@@ -919,72 +885,36 @@ def kill(
             + log_msg_session_name_extension
         )
         self.log.info(msg)
-        try:
-            response = self.stub.kill(request, timeout=timeout)
-        except grpc.RpcError as e:
-            try:
-                error_details = extract_grpc_rich_error(e)
-                self.log.error(error_details)
-            except Exception as extraction_error:
-                self.log.debug(
-                    f"Could not extract rich error details from gRPC error: {extraction_error}",
-                    exc_info=True,
-                )
-            handle_grpc_error(e)
+        response = self.stub.kill(request, timeout=timeout)
 
         return response
 
     def logs(self, request: LogRequest, timeout: int | float = 60) -> LogLines | None:
         request.token.CopyFrom(self.token)
 
-        try:
-            response = self.stub.logs(request, timeout=timeout)
-
-            # Check if the response indicates a BadQuery error
-            if response.flag == ResponseFlag.NOT_EXECUTED_BAD_REQUEST_FORMAT:
-                lines = response.lines
-                if len(lines) == 1:
-                    lines = lines[0]
-                self.log.warning(f"Bad query for logs: {lines}")
-                return None
-
-            # Check for other error flags
-            if response.flag == ResponseFlag.DRUNC_EXCEPTION_THROWN:
-                self.log.error(f"Exception occurred on server: {response.lines}")
-                return None
+        response = self.stub.logs(request, timeout=timeout)
 
-            return response
+        # Check if the response indicates a BadQuery error
+        if response.flag == ResponseFlag.NOT_EXECUTED_BAD_REQUEST_FORMAT:
+            lines = response.lines
+            if len(lines) == 1:
+                lines = lines[0]
+            self.log.warning(f"Bad query for logs: {lines}")
+            return None
 
-        except grpc.RpcError as e:
-            try:
-                error_details = extract_grpc_rich_error(e)
-                self.log.error(error_details)
-            except Exception as extraction_error:
-                self.log.debug(
-                    f"Could not extract rich error details from gRPC error: {extraction_error}",
-                    exc_info=True,
-                )
-            handle_grpc_error(e)
+        # Check for other error flags
+        if response.flag == ResponseFlag.DRUNC_EXCEPTION_THROWN:
+            self.log.error(f"Exception occurred on server: {response.lines}")
             return None
 
+        return response
+
     def ps(
         self, request: ProcessQuery, timeout: int | float = 60
     ) -> ProcessInstanceList:
         request.token.CopyFrom(self.token)
 
-        try:
-            response = self.stub.ps(request, timeout=timeout)
-        except grpc.RpcError as e:
-            try:
-                error_details = extract_grpc_rich_error(e)
-                self.log.error(error_details)
-            except Exception as extraction_error:
-                self.log.debug(
-                    f"Could not extract rich error details from gRPC error: {extraction_error}",
-                    exc_info=True,
-                )
-
-            handle_grpc_error(e)
+        response = self.stub.ps(request, timeout=timeout)
 
         return response
 
@@ -993,19 +923,7 @@ def flush(
     ) -> ProcessInstanceList:
         request.token.CopyFrom(self.token)
 
-        try:
-            response = self.stub.flush(request, timeout=timeout)
-        except grpc.RpcError as e:
-            try:
-                error_details = extract_grpc_rich_error(e)
-                self.log.error(error_details)
-            except Exception as extraction_error:
-                self.log.debug(
-                    f"Could not extract rich error details from gRPC error: {extraction_error}",
-                    exc_info=True,
-                )
-
-            handle_grpc_error(e)
+        response = self.stub.flush(request, timeout=timeout)
 
         return response
 
@@ -1014,43 +932,19 @@ def restart(
     ) -> ProcessInstanceList:
         request.token.CopyFrom(self.token)
 
-        try:
-            response = self.stub.restart(request, timeout=timeout)
-            self.log.info(
-                f"Restarted [green]{request.names}[/green] "
-                f"from session [green]{request.session} [/green]"
-                f"with UUID [green]{response.values[0].uuid.uuid}[/green] on host [green]{response.values[0].process_description.metadata.hostname}[/green]"
-            )
-        except grpc.RpcError as e:
-            try:
-                error_details = extract_grpc_rich_error(e)
-                self.log.error(error_details)
-            except Exception as extraction_error:
-                self.log.debug(
-                    f"Could not extract rich error details from gRPC error: {extraction_error}",
-                    exc_info=True,
-                )
-
-            handle_grpc_error(e)
+        response = self.stub.restart(request, timeout=timeout)
+        self.log.info(
+            f"Restarted [green]{request.names}[/green] "
+            f"from session [green]{request.session} [/green]"
+            f"with UUID [green]{response.values[0].uuid.uuid}[/green] on host [green]{response.values[0].process_description.metadata.hostname}[/green]"
+        )
 
         return response
 
     def describe(self, timeout: int | float = 60) -> Description:
         request = Request(token=copy_token(self.token))
 
-        try:
-            response = self.stub.describe(request, timeout=timeout)
-        except grpc.RpcError as e:
-            try:
-                error_details = extract_grpc_rich_error(e)
-                self.log.error(error_details)
-            except Exception as extraction_error:
-                self.log.debug(
-                    f"Could not extract rich error details from gRPC error: {extraction_error}",
-                    exc_info=True,
-                )
-
-            handle_grpc_error(e)
+        response = self.stub.describe(request, timeout=timeout)
 
         return response
 
@@ -1122,10 +1016,7 @@ def log_on_server(
             execute_along_path=False,
             execute_on_all_subsequent_children_in_path=False,
         )
-        request.token.CopyFrom(self.token)
-        try:
-            response = self.stub.log_on_server(request, timeout=timeout)
-        except grpc.RpcError as e:
-            handle_grpc_error(e)
-
+        response: LogOnServerResponse = self.stub.log_on_server(
+            request, timeout=timeout
+        )
         return response