feat: add grpc trailing metadata when logging grpc error (#5512)

jina/clients/base/grpc.py

@@ -10,6 +10,7 @@
 from jina.excepts import BadClientInput, BadServerFlow, InternalNetworkError
 from jina.logging.profile import ProgressBar
 from jina.proto import jina_pb2, jina_pb2_grpc
+from jina.serve.helper import extract_trailing_metadata
 from jina.serve.networking import GrpcConnectionPool
 from jina.serve.stream.helper import AsyncRequestsIterator
 
@@ -226,7 +227,10 @@ async def _get_results(
                     except (grpc.aio._call.AioRpcError, InternalNetworkError) as err:
                         my_code = err.code()
                         my_details = err.details()
+                        trailing_metadata = extract_trailing_metadata(err)
                         msg = f'gRPC error: {my_code} {my_details}'
+                        if trailing_metadata:
+                            msg = f'gRPC error: {my_code} {my_details}\n{trailing_metadata}'
 
                         if my_code == grpc.StatusCode.UNAVAILABLE:
                             self.logger.error(

jina/excepts.py

@@ -3,6 +3,8 @@
 
 import grpc.aio
 
+from jina.serve.helper import extract_trailing_metadata
+
 
 class BaseJinaException(BaseException):
     """A base class for all exceptions raised by Jina"""
@@ -135,4 +137,11 @@ def details(self):
         """
         :return: details of this exception
         """
-        return self._details if self._details else self.og_exception.details()
+        if self._details:
+            trailing_metadata = extract_trailing_metadata(self.og_exception)
+            if trailing_metadata:
+                return f'{self._details}\n{trailing_metadata}'
+            else:
+                return self._details
+
+        return self.og_exception.details()

jina/serve/helper.py

@@ -3,6 +3,8 @@
 import typing
 from typing import Optional, Union
 
+import grpc
+
 from jina.helper import convert_tuple_to_list
 
 if typing.TYPE_CHECKING:
@@ -72,3 +74,29 @@ def arg_wrapper(self, *args, **kwargs):
         return f
 
     return arg_wrapper
+
+
+def extract_trailing_metadata(error: grpc.aio.AioRpcError) -> Optional[str]:
+    '''Return formatted string of the trailing metadata if exists otherwise return None
+    :param error: AioRpcError
+    :return: string of Metadata or None
+    '''
+    if type(error) == grpc.aio.AioRpcError:
+        trailing_metadata = error.trailing_metadata()
+        if trailing_metadata and len(trailing_metadata):
+            return f'trailing_metadata={trailing_metadata}'
+
+    return None
+
+
+def format_grpc_error(error: grpc.aio.AioRpcError) -> str:
+    '''Adds grpc context trainling metadata if available
+    :param error: AioRpcError
+    :return: formatted error
+    '''
+    default_string = str(error)
+    trailing_metadata = extract_trailing_metadata(error)
+    if trailing_metadata:
+        return f'{default_string}\n{trailing_metadata}'
+
+    return default_string

jina/serve/networking.py

@@ -18,6 +18,7 @@
 from jina.importer import ImportExtensions
 from jina.logging.logger import JinaLogger
 from jina.proto import jina_pb2, jina_pb2_grpc
+from jina.serve.helper import format_grpc_error
 from jina.serve.instrumentation import MetricsTimer
 from jina.types.request import Request
 from jina.types.request.data import DataRequest
@@ -969,7 +970,7 @@ async def _handle_aiorpcerror(
         # requests usually gets cancelled when the server shuts down
         # retries for cancelled requests will hit another replica in K8s
         self._logger.debug(
-            f'GRPC call to {current_deployment} errored, getting error {error} for the {retry_i + 1}th time.'
+            f'GRPC call to {current_deployment} errored, with error {format_grpc_error(error)} and for the {retry_i + 1}th time.'
         )
         if (
             error.code() != grpc.StatusCode.UNAVAILABLE
@@ -1003,7 +1004,7 @@ async def _handle_aiorpcerror(
             )
         else:
             self._logger.debug(
-                f'GRPC call to deployment {current_deployment} failed with code {error.code()}, retry attempt {retry_i + 1}/{total_num_tries - 1}.'
+                f'GRPC call to deployment {current_deployment} failed with error {format_grpc_error(error)}, for retry attempt {retry_i + 1}/{total_num_tries - 1}.'
                 f' Trying next replica, if available.'
             )
             return None

jina/serve/stream/__init__.py

@@ -87,6 +87,7 @@ async def stream(
             ):  # inside GrpcGateway we can handle the error directly here through the grpc context
                 context.set_details(err.details())
                 context.set_code(err.code())
+                context.set_trailing_metadata(err.trailing_metadata())
                 self.logger.error(
                     f'Error while getting responses from deployments: {err.details()}'
                 )

tests/unit/exceptions/test_exceptions.py

@@ -1,6 +1,7 @@
 import grpc.aio
 import pytest
 from grpc import StatusCode
+from grpc.aio import Metadata
 
 from jina.excepts import BaseJinaException, InternalNetworkError
 
@@ -30,3 +31,21 @@ def test_ine_details(aio_rpc_error):
     err = InternalNetworkError(aio_rpc_error, details='I am not a normal grpc error!')
     assert err.details() == 'I am not a normal grpc error!'
     assert str(err) == 'I am not a normal grpc error!'
+
+
+@pytest.mark.parametrize('metadata', [None, Metadata(('content-length', '0'))])
+def test_ine_trailing_metadata(metadata):
+    aio_rpc_error = grpc.aio.AioRpcError(
+        StatusCode.OK,
+        None,
+        trailing_metadata=metadata,
+        details='I am a grpc error',
+    )
+    err = InternalNetworkError(aio_rpc_error)
+    if metadata:
+        assert (
+            str(err)
+            == 'I am a grpc error\ntrailing_metadata=Metadata(((\'content-length\', \'0\'),))'
+        )
+    else:
+        assert str(err) == 'I am a grpc error'