We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support older versions of grpcio
Some attributes, like `trailing_metadata()` don't seem to exist in older versions of grpcio (e.g. `1.34.0`). Ideally we should also provide support for these ones to avoid clashes with older versions of TF (e.g. `2.5.x`). 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mlserver/grpc/utils.py`
Content:
```
1 import grpc
2 
3 from typing import Callable, Dict, Tuple
4 from fastapi import status
5 
6 from grpc import ServicerContext
7 
8 from .logging import logger
9 from ..errors import MLServerError
10 
11 
12 STATUS_CODE_MAPPING = {
13     status.HTTP_400_BAD_REQUEST: grpc.StatusCode.INVALID_ARGUMENT,
14     status.HTTP_404_NOT_FOUND: grpc.StatusCode.NOT_FOUND,
15     status.HTTP_422_UNPROCESSABLE_ENTITY: grpc.StatusCode.FAILED_PRECONDITION,
16     status.HTTP_500_INTERNAL_SERVER_ERROR: grpc.StatusCode.INTERNAL,
17 }
18 
19 
20 def to_headers(context: ServicerContext) -> Dict[str, str]:
21     metadata = context.invocation_metadata() + context.trailing_metadata()
22     headers = {}
23     for metadatum in metadata:
24         headers[metadatum.key] = metadatum.value
25 
26     return headers
27 
28 
29 def to_metadata(headers: Dict[str, str]) -> Tuple[Tuple[str, str], ...]:
30     return tuple(headers.items())
31 
32 
33 def _grpc_status_code(err: MLServerError):
34     return STATUS_CODE_MAPPING.get(err.status_code, grpc.StatusCode.UNKNOWN)
35 
36 
37 def handle_mlserver_error(f: Callable):
38     async def _inner(self, request, context):
39         try:
40             return await f(self, request, context)
41         except MLServerError as err:
42             logger.error(err)
43             await context.abort(code=_grpc_status_code(err), details=str(err))
44 
45     return _inner
46
```

Path: `mlserver/grpc/interceptors.py`
Content:
```
1 from typing import Awaitable, Callable, Tuple
2 from functools import partial
3 from timeit import default_timer
4 
5 from grpc.aio import ServerInterceptor, ServicerContext
6 from grpc import HandlerCallDetails, RpcMethodHandler, RpcError, StatusCode
7 from py_grpc_prometheus.prometheus_server_interceptor import (
8     grpc_utils,
9     PromServerInterceptor as _PromServerInterceptor,
10 )
11 
12 from .logging import logger
13 
14 
15 class LoggingInterceptor(ServerInterceptor):
16     def _get_log_message(self, handler_call_details: HandlerCallDetails) -> str:
17         return handler_call_details.method
18 
19     async def intercept_service(
20         self,
21         continuation: Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler]],
22         handler_call_details: HandlerCallDetails,
23     ) -> RpcMethodHandler:
24         logger.info(self._get_log_message(handler_call_details))
25         handler = await continuation(handler_call_details)
26         return handler
27 
28 
29 class PromServerInterceptor(ServerInterceptor):
30     """
31     Simple wrapper around `py_grpc_prometheus` to support `grpc.aio`.
32 
33     TODO: Open PR to add support upstream for AsyncIO.
34     """
35 
36     def __init__(self, *args, **kwargs):
37         self._interceptor = _PromServerInterceptor(*args, **kwargs)
38         # We need a status code mapping to ensure we can convert from an int:
39         # https://groups.google.com/g/grpc-io/c/EdIXjMEaOyw/m/d3DeqmrJAAAJ
40         self._status_codes = {code.value[0]: code for code in StatusCode}
41 
42     async def intercept_service(
43         self,
44         continuation: Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler]],
45         handler_call_details: HandlerCallDetails,
46     ) -> RpcMethodHandler:
47         method_call = grpc_utils.split_method_call(handler_call_details)
48         handler = await continuation(handler_call_details)
49 
50         metrics_wrapper = partial(self._metrics_wrapper, method_call)
51         return self._interceptor._wrap_rpc_behavior(handler, metrics_wrapper)
52 
53     def _compute_status_code(self, servicer_context: ServicerContext) -> StatusCode:
54         """
55         This method is mostly copied from `py-grpc-prometheus`, with a couple
56         minor changes to avoid using private APIs from ServicerContext which
57         don't exist anymore in `grpc.aio`.
58         To see the original implementation, please check:
59 
60         https://github.com/lchenn/py-grpc-prometheus/blob/eb9dee1f0a4e57cef220193ee48021dc9a9f3d82/py_grpc_prometheus/prometheus_server_interceptor.py#L127-L134
61         """
62         # Backwards compatibility for non-aio.
63         # TODO: It's not clear yet how to check whether the context has been
64         # cancelled with aio.
65         if hasattr(servicer_context, "_state"):
66             if servicer_context._state.client == "cancelled":
67                 return StatusCode.CANCELLED
68 
69         code = servicer_context.code()
70         if code is None:
71             return StatusCode.OK
72 
73         # NOTE: With gRPC AIO, the `code` can be a plain integer that needs to
74         # be converted to an actual `StatusCode` entry
75         if isinstance(code, int):
76             if code not in self._status_codes:
77                 return StatusCode.UNKNOWN
78 
79             return self._status_codes[code]
80 
81         return code
82 
83     def _metrics_wrapper(
84         self,
85         method_call: Tuple[str, str, str],
86         old_handler: RpcMethodHandler,
87         request_streaming: bool,
88         response_streaming: bool,
89     ):
90         """
91         Port of `py-grpc-prometheus` metrics_wrapper method to work with gRPC's
92         AsyncIO support.
93         To see the original implementation, please check:
94 
95         https://github.com/lchenn/py-grpc-prometheus/blob/eb9dee1f0a4e57cef220193ee48021dc9a9f3d82/py_grpc_prometheus/prometheus_server_interceptor.py#L46-L120
96         """
97         grpc_service_name, grpc_method_name, _ = method_call
98 
99         async def _new_handler(request_or_iterator, servicer_context: ServicerContext):
100             response_or_iterator = None
101             try:
102                 start = default_timer()
103                 grpc_type = grpc_utils.get_method_type(
104                     request_streaming, response_streaming
105                 )
106                 try:
107                     if request_streaming:
108                         request_or_iterator = grpc_utils.wrap_iterator_inc_counter(
109                             request_or_iterator,
110                             self._interceptor._metrics[
111                                 "grpc_server_stream_msg_received"
112                             ],
113                             grpc_type,
114                             grpc_service_name,
115                             grpc_method_name,
116                         )
117                     else:
118                         self._interceptor._metrics[
119                             "grpc_server_started_counter"
120                         ].labels(
121                             grpc_type=grpc_type,
122                             grpc_service=grpc_service_name,
123                             grpc_method=grpc_method_name,
124                         ).inc()
125 
126                     # Invoke the original rpc behavior.
127                     # NOTE: This is the main change required with respect to
128                     # the original implementation in `py-grpc-prometheus`.
129                     response_or_iterator = await old_handler(
130                         request_or_iterator, servicer_context
131                     )
132 
133                     if response_streaming:
134                         sent_metric = self._interceptor._metrics[
135                             "grpc_server_stream_msg_sent"
136                         ]
137                         response_or_iterator = grpc_utils.wrap_iterator_inc_counter(
138                             response_or_iterator,
139                             sent_metric,
140                             grpc_type,
141                             grpc_service_name,
142                             grpc_method_name,
143                         )
144 
145                     else:
146                         self._interceptor.increase_grpc_server_handled_total_counter(
147                             grpc_type,
148                             grpc_service_name,
149                             grpc_method_name,
150                             self._compute_status_code(servicer_context).name,
151                         )
152                     return response_or_iterator
153                 except RpcError as e:
154                     self._interceptor.increase_grpc_server_handled_total_counter(
155                         grpc_type,
156                         grpc_service_name,
157                         grpc_method_name,
158                         self._interceptor._compute_error_code(e).name,
159                     )
160                     raise e
161 
162                 finally:
163 
164                     if not response_streaming:
165                         if self._interceptor._legacy:
166                             self._interceptor._metrics[
167                                 "legacy_grpc_server_handled_latency_seconds"
168                             ].labels(
169                                 grpc_type=grpc_type,
170                                 grpc_service=grpc_service_name,
171                                 grpc_method=grpc_method_name,
172                             ).observe(
173                                 max(default_timer() - start, 0)
174                             )
175                         elif self._interceptor._enable_handling_time_histogram:
176                             self._interceptor._metrics[
177                                 "grpc_server_handled_histogram"
178                             ].labels(
179                                 grpc_type=grpc_type,
180                                 grpc_service=grpc_service_name,
181                                 grpc_method=grpc_method_name,
182                             ).observe(
183                                 max(default_timer() - start, 0)
184                             )
185             except Exception as e:  # pylint: disable=broad-except
186                 # Allow user to skip the exceptions in order to maintain
187                 # the basic functionality in the server
188                 # The logging function in exception can be toggled with log_exceptions
189                 # in order to suppress the noise in logging
190                 if self._interceptor._skip_exceptions:
191                     if self._interceptor._log_exceptions:
192                         logger.error(e)
193                     if response_or_iterator is None:
194                         return response_or_iterator
195                     return old_handler(request_or_iterator, servicer_context)
196                 raise e
197 
198         return _new_handler
199
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/mlserver/grpc/interceptors.py b/mlserver/grpc/interceptors.py
--- a/mlserver/grpc/interceptors.py
+++ b/mlserver/grpc/interceptors.py
@@ -66,6 +66,9 @@
             if servicer_context._state.client == "cancelled":
                 return StatusCode.CANCELLED
 
+        if not hasattr(servicer_context, "code"):
+            return StatusCode.OK
+
         code = servicer_context.code()
         if code is None:
             return StatusCode.OK
diff --git a/mlserver/grpc/utils.py b/mlserver/grpc/utils.py
--- a/mlserver/grpc/utils.py
+++ b/mlserver/grpc/utils.py
@@ -18,7 +18,11 @@
 
 
 def to_headers(context: ServicerContext) -> Dict[str, str]:
-    metadata = context.invocation_metadata() + context.trailing_metadata()
+    metadata = context.invocation_metadata()
+    if hasattr(context, "trailing_metadata"):
+        # NOTE: Older versions of `grpcio` (e.g. `grpcio==1.34.0`) don't expose
+        # access to the trailing metadata on the service side
+        metadata += context.trailing_metadata()
     headers = {}
     for metadatum in metadata:
         headers[metadatum.key] = metadatum.value

{"golden_diff": "diff --git a/mlserver/grpc/interceptors.py b/mlserver/grpc/interceptors.py\n--- a/mlserver/grpc/interceptors.py\n+++ b/mlserver/grpc/interceptors.py\n@@ -66,6 +66,9 @@\n             if servicer_context._state.client == \"cancelled\":\n                 return StatusCode.CANCELLED\n \n+        if not hasattr(servicer_context, \"code\"):\n+            return StatusCode.OK\n+\n         code = servicer_context.code()\n         if code is None:\n             return StatusCode.OK\ndiff --git a/mlserver/grpc/utils.py b/mlserver/grpc/utils.py\n--- a/mlserver/grpc/utils.py\n+++ b/mlserver/grpc/utils.py\n@@ -18,7 +18,11 @@\n \n \n def to_headers(context: ServicerContext) -> Dict[str, str]:\n-    metadata = context.invocation_metadata() + context.trailing_metadata()\n+    metadata = context.invocation_metadata()\n+    if hasattr(context, \"trailing_metadata\"):\n+        # NOTE: Older versions of `grpcio` (e.g. `grpcio==1.34.0`) don't expose\n+        # access to the trailing metadata on the service side\n+        metadata += context.trailing_metadata()\n     headers = {}\n     for metadatum in metadata:\n         headers[metadatum.key] = metadatum.value\n", "issue": "Support older versions of grpcio\nSome attributes, like `trailing_metadata()` don't seem to exist in older versions of grpcio (e.g. `1.34.0`). Ideally we should also provide support for these ones to avoid clashes with older versions of TF (e.g. `2.5.x`). \n", "before_files": [{"content": "import grpc\n\nfrom typing import Callable, Dict, Tuple\nfrom fastapi import status\n\nfrom grpc import ServicerContext\n\nfrom .logging import logger\nfrom ..errors import MLServerError\n\n\nSTATUS_CODE_MAPPING = {\n    status.HTTP_400_BAD_REQUEST: grpc.StatusCode.INVALID_ARGUMENT,\n    status.HTTP_404_NOT_FOUND: grpc.StatusCode.NOT_FOUND,\n    status.HTTP_422_UNPROCESSABLE_ENTITY: grpc.StatusCode.FAILED_PRECONDITION,\n    status.HTTP_500_INTERNAL_SERVER_ERROR: grpc.StatusCode.INTERNAL,\n}\n\n\ndef to_headers(context: ServicerContext) -> Dict[str, str]:\n    metadata = context.invocation_metadata() + context.trailing_metadata()\n    headers = {}\n    for metadatum in metadata:\n        headers[metadatum.key] = metadatum.value\n\n    return headers\n\n\ndef to_metadata(headers: Dict[str, str]) -> Tuple[Tuple[str, str], ...]:\n    return tuple(headers.items())\n\n\ndef _grpc_status_code(err: MLServerError):\n    return STATUS_CODE_MAPPING.get(err.status_code, grpc.StatusCode.UNKNOWN)\n\n\ndef handle_mlserver_error(f: Callable):\n    async def _inner(self, request, context):\n        try:\n            return await f(self, request, context)\n        except MLServerError as err:\n            logger.error(err)\n            await context.abort(code=_grpc_status_code(err), details=str(err))\n\n    return _inner\n", "path": "mlserver/grpc/utils.py"}, {"content": "from typing import Awaitable, Callable, Tuple\nfrom functools import partial\nfrom timeit import default_timer\n\nfrom grpc.aio import ServerInterceptor, ServicerContext\nfrom grpc import HandlerCallDetails, RpcMethodHandler, RpcError, StatusCode\nfrom py_grpc_prometheus.prometheus_server_interceptor import (\n    grpc_utils,\n    PromServerInterceptor as _PromServerInterceptor,\n)\n\nfrom .logging import logger\n\n\nclass LoggingInterceptor(ServerInterceptor):\n    def _get_log_message(self, handler_call_details: HandlerCallDetails) -> str:\n        return handler_call_details.method\n\n    async def intercept_service(\n        self,\n        continuation: Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler]],\n        handler_call_details: HandlerCallDetails,\n    ) -> RpcMethodHandler:\n        logger.info(self._get_log_message(handler_call_details))\n        handler = await continuation(handler_call_details)\n        return handler\n\n\nclass PromServerInterceptor(ServerInterceptor):\n    \"\"\"\n    Simple wrapper around `py_grpc_prometheus` to support `grpc.aio`.\n\n    TODO: Open PR to add support upstream for AsyncIO.\n    \"\"\"\n\n    def __init__(self, *args, **kwargs):\n        self._interceptor = _PromServerInterceptor(*args, **kwargs)\n        # We need a status code mapping to ensure we can convert from an int:\n        # https://groups.google.com/g/grpc-io/c/EdIXjMEaOyw/m/d3DeqmrJAAAJ\n        self._status_codes = {code.value[0]: code for code in StatusCode}\n\n    async def intercept_service(\n        self,\n        continuation: Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler]],\n        handler_call_details: HandlerCallDetails,\n    ) -> RpcMethodHandler:\n        method_call = grpc_utils.split_method_call(handler_call_details)\n        handler = await continuation(handler_call_details)\n\n        metrics_wrapper = partial(self._metrics_wrapper, method_call)\n        return self._interceptor._wrap_rpc_behavior(handler, metrics_wrapper)\n\n    def _compute_status_code(self, servicer_context: ServicerContext) -> StatusCode:\n        \"\"\"\n        This method is mostly copied from `py-grpc-prometheus`, with a couple\n        minor changes to avoid using private APIs from ServicerContext which\n        don't exist anymore in `grpc.aio`.\n        To see the original implementation, please check:\n\n        https://github.com/lchenn/py-grpc-prometheus/blob/eb9dee1f0a4e57cef220193ee48021dc9a9f3d82/py_grpc_prometheus/prometheus_server_interceptor.py#L127-L134\n        \"\"\"\n        # Backwards compatibility for non-aio.\n        # TODO: It's not clear yet how to check whether the context has been\n        # cancelled with aio.\n        if hasattr(servicer_context, \"_state\"):\n            if servicer_context._state.client == \"cancelled\":\n                return StatusCode.CANCELLED\n\n        code = servicer_context.code()\n        if code is None:\n            return StatusCode.OK\n\n        # NOTE: With gRPC AIO, the `code` can be a plain integer that needs to\n        # be converted to an actual `StatusCode` entry\n        if isinstance(code, int):\n            if code not in self._status_codes:\n                return StatusCode.UNKNOWN\n\n            return self._status_codes[code]\n\n        return code\n\n    def _metrics_wrapper(\n        self,\n        method_call: Tuple[str, str, str],\n        old_handler: RpcMethodHandler,\n        request_streaming: bool,\n        response_streaming: bool,\n    ):\n        \"\"\"\n        Port of `py-grpc-prometheus` metrics_wrapper method to work with gRPC's\n        AsyncIO support.\n        To see the original implementation, please check:\n\n        https://github.com/lchenn/py-grpc-prometheus/blob/eb9dee1f0a4e57cef220193ee48021dc9a9f3d82/py_grpc_prometheus/prometheus_server_interceptor.py#L46-L120\n        \"\"\"\n        grpc_service_name, grpc_method_name, _ = method_call\n\n        async def _new_handler(request_or_iterator, servicer_context: ServicerContext):\n            response_or_iterator = None\n            try:\n                start = default_timer()\n                grpc_type = grpc_utils.get_method_type(\n                    request_streaming, response_streaming\n                )\n                try:\n                    if request_streaming:\n                        request_or_iterator = grpc_utils.wrap_iterator_inc_counter(\n                            request_or_iterator,\n                            self._interceptor._metrics[\n                                \"grpc_server_stream_msg_received\"\n                            ],\n                            grpc_type,\n                            grpc_service_name,\n                            grpc_method_name,\n                        )\n                    else:\n                        self._interceptor._metrics[\n                            \"grpc_server_started_counter\"\n                        ].labels(\n                            grpc_type=grpc_type,\n                            grpc_service=grpc_service_name,\n                            grpc_method=grpc_method_name,\n                        ).inc()\n\n                    # Invoke the original rpc behavior.\n                    # NOTE: This is the main change required with respect to\n                    # the original implementation in `py-grpc-prometheus`.\n                    response_or_iterator = await old_handler(\n                        request_or_iterator, servicer_context\n                    )\n\n                    if response_streaming:\n                        sent_metric = self._interceptor._metrics[\n                            \"grpc_server_stream_msg_sent\"\n                        ]\n                        response_or_iterator = grpc_utils.wrap_iterator_inc_counter(\n                            response_or_iterator,\n                            sent_metric,\n                            grpc_type,\n                            grpc_service_name,\n                            grpc_method_name,\n                        )\n\n                    else:\n                        self._interceptor.increase_grpc_server_handled_total_counter(\n                            grpc_type,\n                            grpc_service_name,\n                            grpc_method_name,\n                            self._compute_status_code(servicer_context).name,\n                        )\n                    return response_or_iterator\n                except RpcError as e:\n                    self._interceptor.increase_grpc_server_handled_total_counter(\n                        grpc_type,\n                        grpc_service_name,\n                        grpc_method_name,\n                        self._interceptor._compute_error_code(e).name,\n                    )\n                    raise e\n\n                finally:\n\n                    if not response_streaming:\n                        if self._interceptor._legacy:\n                            self._interceptor._metrics[\n                                \"legacy_grpc_server_handled_latency_seconds\"\n                            ].labels(\n                                grpc_type=grpc_type,\n                                grpc_service=grpc_service_name,\n                                grpc_method=grpc_method_name,\n                            ).observe(\n                                max(default_timer() - start, 0)\n                            )\n                        elif self._interceptor._enable_handling_time_histogram:\n                            self._interceptor._metrics[\n                                \"grpc_server_handled_histogram\"\n                            ].labels(\n                                grpc_type=grpc_type,\n                                grpc_service=grpc_service_name,\n                                grpc_method=grpc_method_name,\n                            ).observe(\n                                max(default_timer() - start, 0)\n                            )\n            except Exception as e:  # pylint: disable=broad-except\n                # Allow user to skip the exceptions in order to maintain\n                # the basic functionality in the server\n                # The logging function in exception can be toggled with log_exceptions\n                # in order to suppress the noise in logging\n                if self._interceptor._skip_exceptions:\n                    if self._interceptor._log_exceptions:\n                        logger.error(e)\n                    if response_or_iterator is None:\n                        return response_or_iterator\n                    return old_handler(request_or_iterator, servicer_context)\n                raise e\n\n        return _new_handler\n", "path": "mlserver/grpc/interceptors.py"}], "after_files": [{"content": "import grpc\n\nfrom typing import Callable, Dict, Tuple\nfrom fastapi import status\n\nfrom grpc import ServicerContext\n\nfrom .logging import logger\nfrom ..errors import MLServerError\n\n\nSTATUS_CODE_MAPPING = {\n    status.HTTP_400_BAD_REQUEST: grpc.StatusCode.INVALID_ARGUMENT,\n    status.HTTP_404_NOT_FOUND: grpc.StatusCode.NOT_FOUND,\n    status.HTTP_422_UNPROCESSABLE_ENTITY: grpc.StatusCode.FAILED_PRECONDITION,\n    status.HTTP_500_INTERNAL_SERVER_ERROR: grpc.StatusCode.INTERNAL,\n}\n\n\ndef to_headers(context: ServicerContext) -> Dict[str, str]:\n    metadata = context.invocation_metadata()\n    if hasattr(context, \"trailing_metadata\"):\n        # NOTE: Older versions of `grpcio` (e.g. `grpcio==1.34.0`) don't expose\n        # access to the trailing metadata on the service side\n        metadata += context.trailing_metadata()\n    headers = {}\n    for metadatum in metadata:\n        headers[metadatum.key] = metadatum.value\n\n    return headers\n\n\ndef to_metadata(headers: Dict[str, str]) -> Tuple[Tuple[str, str], ...]:\n    return tuple(headers.items())\n\n\ndef _grpc_status_code(err: MLServerError):\n    return STATUS_CODE_MAPPING.get(err.status_code, grpc.StatusCode.UNKNOWN)\n\n\ndef handle_mlserver_error(f: Callable):\n    async def _inner(self, request, context):\n        try:\n            return await f(self, request, context)\n        except MLServerError as err:\n            logger.error(err)\n            await context.abort(code=_grpc_status_code(err), details=str(err))\n\n    return _inner\n", "path": "mlserver/grpc/utils.py"}, {"content": "from typing import Awaitable, Callable, Tuple\nfrom functools import partial\nfrom timeit import default_timer\n\nfrom grpc.aio import ServerInterceptor, ServicerContext\nfrom grpc import HandlerCallDetails, RpcMethodHandler, RpcError, StatusCode\nfrom py_grpc_prometheus.prometheus_server_interceptor import (\n    grpc_utils,\n    PromServerInterceptor as _PromServerInterceptor,\n)\n\nfrom .logging import logger\n\n\nclass LoggingInterceptor(ServerInterceptor):\n    def _get_log_message(self, handler_call_details: HandlerCallDetails) -> str:\n        return handler_call_details.method\n\n    async def intercept_service(\n        self,\n        continuation: Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler]],\n        handler_call_details: HandlerCallDetails,\n    ) -> RpcMethodHandler:\n        logger.info(self._get_log_message(handler_call_details))\n        handler = await continuation(handler_call_details)\n        return handler\n\n\nclass PromServerInterceptor(ServerInterceptor):\n    \"\"\"\n    Simple wrapper around `py_grpc_prometheus` to support `grpc.aio`.\n\n    TODO: Open PR to add support upstream for AsyncIO.\n    \"\"\"\n\n    def __init__(self, *args, **kwargs):\n        self._interceptor = _PromServerInterceptor(*args, **kwargs)\n        # We need a status code mapping to ensure we can convert from an int:\n        # https://groups.google.com/g/grpc-io/c/EdIXjMEaOyw/m/d3DeqmrJAAAJ\n        self._status_codes = {code.value[0]: code for code in StatusCode}\n\n    async def intercept_service(\n        self,\n        continuation: Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler]],\n        handler_call_details: HandlerCallDetails,\n    ) -> RpcMethodHandler:\n        method_call = grpc_utils.split_method_call(handler_call_details)\n        handler = await continuation(handler_call_details)\n\n        metrics_wrapper = partial(self._metrics_wrapper, method_call)\n        return self._interceptor._wrap_rpc_behavior(handler, metrics_wrapper)\n\n    def _compute_status_code(self, servicer_context: ServicerContext) -> StatusCode:\n        \"\"\"\n        This method is mostly copied from `py-grpc-prometheus`, with a couple\n        minor changes to avoid using private APIs from ServicerContext which\n        don't exist anymore in `grpc.aio`.\n        To see the original implementation, please check:\n\n        https://github.com/lchenn/py-grpc-prometheus/blob/eb9dee1f0a4e57cef220193ee48021dc9a9f3d82/py_grpc_prometheus/prometheus_server_interceptor.py#L127-L134\n        \"\"\"\n        # Backwards compatibility for non-aio.\n        # TODO: It's not clear yet how to check whether the context has been\n        # cancelled with aio.\n        if hasattr(servicer_context, \"_state\"):\n            if servicer_context._state.client == \"cancelled\":\n                return StatusCode.CANCELLED\n\n        if not hasattr(servicer_context, \"code\"):\n            return StatusCode.OK\n\n        code = servicer_context.code()\n        if code is None:\n            return StatusCode.OK\n\n        # NOTE: With gRPC AIO, the `code` can be a plain integer that needs to\n        # be converted to an actual `StatusCode` entry\n        if isinstance(code, int):\n            if code not in self._status_codes:\n                return StatusCode.UNKNOWN\n\n            return self._status_codes[code]\n\n        return code\n\n    def _metrics_wrapper(\n        self,\n        method_call: Tuple[str, str, str],\n        old_handler: RpcMethodHandler,\n        request_streaming: bool,\n        response_streaming: bool,\n    ):\n        \"\"\"\n        Port of `py-grpc-prometheus` metrics_wrapper method to work with gRPC's\n        AsyncIO support.\n        To see the original implementation, please check:\n\n        https://github.com/lchenn/py-grpc-prometheus/blob/eb9dee1f0a4e57cef220193ee48021dc9a9f3d82/py_grpc_prometheus/prometheus_server_interceptor.py#L46-L120\n        \"\"\"\n        grpc_service_name, grpc_method_name, _ = method_call\n\n        async def _new_handler(request_or_iterator, servicer_context: ServicerContext):\n            response_or_iterator = None\n            try:\n                start = default_timer()\n                grpc_type = grpc_utils.get_method_type(\n                    request_streaming, response_streaming\n                )\n                try:\n                    if request_streaming:\n                        request_or_iterator = grpc_utils.wrap_iterator_inc_counter(\n                            request_or_iterator,\n                            self._interceptor._metrics[\n                                \"grpc_server_stream_msg_received\"\n                            ],\n                            grpc_type,\n                            grpc_service_name,\n                            grpc_method_name,\n                        )\n                    else:\n                        self._interceptor._metrics[\n                            \"grpc_server_started_counter\"\n                        ].labels(\n                            grpc_type=grpc_type,\n                            grpc_service=grpc_service_name,\n                            grpc_method=grpc_method_name,\n                        ).inc()\n\n                    # Invoke the original rpc behavior.\n                    # NOTE: This is the main change required with respect to\n                    # the original implementation in `py-grpc-prometheus`.\n                    response_or_iterator = await old_handler(\n                        request_or_iterator, servicer_context\n                    )\n\n                    if response_streaming:\n                        sent_metric = self._interceptor._metrics[\n                            \"grpc_server_stream_msg_sent\"\n                        ]\n                        response_or_iterator = grpc_utils.wrap_iterator_inc_counter(\n                            response_or_iterator,\n                            sent_metric,\n                            grpc_type,\n                            grpc_service_name,\n                            grpc_method_name,\n                        )\n\n                    else:\n                        self._interceptor.increase_grpc_server_handled_total_counter(\n                            grpc_type,\n                            grpc_service_name,\n                            grpc_method_name,\n                            self._compute_status_code(servicer_context).name,\n                        )\n                    return response_or_iterator\n                except RpcError as e:\n                    self._interceptor.increase_grpc_server_handled_total_counter(\n                        grpc_type,\n                        grpc_service_name,\n                        grpc_method_name,\n                        self._interceptor._compute_error_code(e).name,\n                    )\n                    raise e\n\n                finally:\n\n                    if not response_streaming:\n                        if self._interceptor._legacy:\n                            self._interceptor._metrics[\n                                \"legacy_grpc_server_handled_latency_seconds\"\n                            ].labels(\n                                grpc_type=grpc_type,\n                                grpc_service=grpc_service_name,\n                                grpc_method=grpc_method_name,\n                            ).observe(\n                                max(default_timer() - start, 0)\n                            )\n                        elif self._interceptor._enable_handling_time_histogram:\n                            self._interceptor._metrics[\n                                \"grpc_server_handled_histogram\"\n                            ].labels(\n                                grpc_type=grpc_type,\n                                grpc_service=grpc_service_name,\n                                grpc_method=grpc_method_name,\n                            ).observe(\n                                max(default_timer() - start, 0)\n                            )\n            except Exception as e:  # pylint: disable=broad-except\n                # Allow user to skip the exceptions in order to maintain\n                # the basic functionality in the server\n                # The logging function in exception can be toggled with log_exceptions\n                # in order to suppress the noise in logging\n                if self._interceptor._skip_exceptions:\n                    if self._interceptor._log_exceptions:\n                        logger.error(e)\n                    if response_or_iterator is None:\n                        return response_or_iterator\n                    return old_handler(request_or_iterator, servicer_context)\n                raise e\n\n        return _new_handler\n", "path": "mlserver/grpc/interceptors.py"}]}