We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
grpc: trailing metadata can't be passed when aborting an AIO server RPC
**Describe your environment**

Python: 3.12.0
`opentelemetry-distro[otlp]==0.42b0`
`opentelemetry-instrumentation-grpc==0.42b0`
`grpcio==1.59.2`

**Steps to reproduce**
Full replication can be found [here](https://github.com/cookiefission/testbench/tree/main/otel/python-grpc-aio-abort-issue).

`server.py`:

```python
import asyncio
import logging

import grpc
import helloworld_pb2
import helloworld_pb2_grpc


class Greeter(helloworld_pb2_grpc.GreeterServicer):
    async def SayHello(
        self,
        request: helloworld_pb2.HelloRequest,
        context: grpc.aio.ServicerContext,
    ) -> helloworld_pb2.HelloReply:
        metadata = (
            ("this", "should"),
            ("work", "nicely")
        )
        await context.abort(code=grpc.StatusCode.ABORTED, details="This is the actual error message", trailing_metadata=metadata)


async def serve() -> None:
    server = grpc.aio.server()
    helloworld_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)
    listen_addr = "[::]:50051"
    server.add_insecure_port(listen_addr)
    logging.info("Starting server on %s", listen_addr)
    await server.start()
    await server.wait_for_termination()


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(serve())
```
This server is just adopted from [the grpc examples](https://github.com/grpc/grpc/blob/master/examples/python/helloworld/async_greeter_server.py). The helloworld protobuf artefacts can be found [there](https://github.com/grpc/grpc/blob/master/examples/python/helloworld/helloworld_pb2_grpc.py).

For a client, the [`async_greeter_client.py`](https://github.com/grpc/grpc/blob/master/examples/python/helloworld/async_greeter_client.py) from grpc examples can be used.

This only happens when `trailing_metadata` is passed as an argument to `context.abort`. 

**What is the expected behavior?**

Running the server without opentelemetry, the (expected) error from running the client is:

```
grpc.aio._call.AioRpcError: <AioRpcError of RPC that terminated with:
        status = StatusCode.ABORTED
        details = "This is the actual error message"
        debug_error_string = "UNKNOWN:Error received from peer  {grpc_message:"This is the actual error message", grpc_status:10, created_time:"2023-11-16T15:09:03.552303+00:00"}"
>
```

When running the server under `opentelemetry-instrument`, I expect the same output as when it is run without OTEL. 

**What is the actual behavior?**

Instead, an exception is still raised but the underlying code, details, and trailing_metadata are missing. This makes any nuanced error handling on the client side impossible. 

```
grpc.aio._call.AioRpcError: <AioRpcError of RPC that terminated with:
        status = StatusCode.UNKNOWN
        details = "Unexpected <class 'TypeError'>: _OpenTelemetryServicerContext.abort() got an unexpected keyword argument 'trailing_metadata'"
        debug_error_string = "UNKNOWN:Error received from peer  {created_time:"2023-11-16T15:30:02.017999+00:00", grpc_status:2, grpc_message:"Unexpected <class \'TypeError\'>: _OpenTelemetryServicerContext.abort() got an unexpected keyword argument \'trailing_metadata\'"}"
>
```

**Additional context**

The `abort` method for [`grpc.ServicerContext`](https://grpc.github.io/grpc/python/grpc.html#grpc.ServicerContext.abort) and [`grpc.aio.ServicerContext`](https://grpc.github.io/grpc/python/grpc_asyncio.html#grpc.aio.ServicerContext.abort) is different. It's unclear why. The AIO instrumentation [re-uses](https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py#L19) the [`_OpenTelemetryServicerContext`](https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py#L68) from the non-async server instrumentation which is how this bug snuck in, despite there being a test in place for the abort. 


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py`
Content:
```
1 # Copyright The OpenTelemetry Authors
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import grpc.aio
16 
17 from ._server import (
18     OpenTelemetryServerInterceptor,
19     _OpenTelemetryServicerContext,
20     _wrap_rpc_behavior,
21 )
22 
23 
24 class OpenTelemetryAioServerInterceptor(
25     grpc.aio.ServerInterceptor, OpenTelemetryServerInterceptor
26 ):
27     """
28     An AsyncIO gRPC server interceptor, to add OpenTelemetry.
29     Usage::
30         tracer = some OpenTelemetry tracer
31         interceptors = [
32             AsyncOpenTelemetryServerInterceptor(tracer),
33         ]
34         server = aio.server(
35             futures.ThreadPoolExecutor(max_workers=concurrency),
36             interceptors = (interceptors,))
37     """
38 
39     async def intercept_service(self, continuation, handler_call_details):
40         if self._filter is not None and not self._filter(handler_call_details):
41             return await continuation(handler_call_details)
42 
43         def telemetry_wrapper(behavior, request_streaming, response_streaming):
44             # handle streaming responses specially
45             if response_streaming:
46                 return self._intercept_aio_server_stream(
47                     behavior,
48                     handler_call_details,
49                 )
50 
51             return self._intercept_aio_server_unary(
52                 behavior,
53                 handler_call_details,
54             )
55 
56         next_handler = await continuation(handler_call_details)
57 
58         return _wrap_rpc_behavior(next_handler, telemetry_wrapper)
59 
60     def _intercept_aio_server_unary(self, behavior, handler_call_details):
61         async def _unary_interceptor(request_or_iterator, context):
62             with self._set_remote_context(context):
63                 with self._start_span(
64                     handler_call_details,
65                     context,
66                     set_status_on_exception=False,
67                 ) as span:
68                     # wrap the context
69                     context = _OpenTelemetryServicerContext(context, span)
70 
71                     # And now we run the actual RPC.
72                     try:
73                         return await behavior(request_or_iterator, context)
74 
75                     except Exception as error:
76                         # Bare exceptions are likely to be gRPC aborts, which
77                         # we handle in our context wrapper.
78                         # Here, we're interested in uncaught exceptions.
79                         # pylint:disable=unidiomatic-typecheck
80                         if type(error) != Exception:  # noqa: E721
81                             span.record_exception(error)
82                         raise error
83 
84         return _unary_interceptor
85 
86     def _intercept_aio_server_stream(self, behavior, handler_call_details):
87         async def _stream_interceptor(request_or_iterator, context):
88             with self._set_remote_context(context):
89                 with self._start_span(
90                     handler_call_details,
91                     context,
92                     set_status_on_exception=False,
93                 ) as span:
94                     context = _OpenTelemetryServicerContext(context, span)
95 
96                     try:
97                         async for response in behavior(
98                             request_or_iterator, context
99                         ):
100                             yield response
101 
102                     except Exception as error:
103                         # pylint:disable=unidiomatic-typecheck
104                         if type(error) != Exception:  # noqa: E721
105                             span.record_exception(error)
106                         raise error
107 
108         return _stream_interceptor
109
```
--- 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/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py b/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py
--- a/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py
+++ b/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py
@@ -12,13 +12,63 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import grpc
 import grpc.aio
-
-from ._server import (
-    OpenTelemetryServerInterceptor,
-    _OpenTelemetryServicerContext,
-    _wrap_rpc_behavior,
-)
+import wrapt
+
+from opentelemetry.semconv.trace import SpanAttributes
+from opentelemetry.trace.status import Status, StatusCode
+
+from ._server import OpenTelemetryServerInterceptor, _wrap_rpc_behavior
+
+
+# pylint:disable=abstract-method
+class _OpenTelemetryAioServicerContext(wrapt.ObjectProxy):
+    def __init__(self, servicer_context, active_span):
+        super().__init__(servicer_context)
+        self._self_active_span = active_span
+        self._self_code = grpc.StatusCode.OK
+        self._self_details = None
+
+    async def abort(self, code, details="", trailing_metadata=tuple()):
+        self._self_code = code
+        self._self_details = details
+        self._self_active_span.set_attribute(
+            SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]
+        )
+        self._self_active_span.set_status(
+            Status(
+                status_code=StatusCode.ERROR,
+                description=f"{code}:{details}",
+            )
+        )
+        return await self.__wrapped__.abort(code, details, trailing_metadata)
+
+    def set_code(self, code):
+        self._self_code = code
+        details = self._self_details or code.value[1]
+        self._self_active_span.set_attribute(
+            SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]
+        )
+        if code != grpc.StatusCode.OK:
+            self._self_active_span.set_status(
+                Status(
+                    status_code=StatusCode.ERROR,
+                    description=f"{code}:{details}",
+                )
+            )
+        return self.__wrapped__.set_code(code)
+
+    def set_details(self, details):
+        self._self_details = details
+        if self._self_code != grpc.StatusCode.OK:
+            self._self_active_span.set_status(
+                Status(
+                    status_code=StatusCode.ERROR,
+                    description=f"{self._self_code}:{details}",
+                )
+            )
+        return self.__wrapped__.set_details(details)
 
 
 class OpenTelemetryAioServerInterceptor(
@@ -66,7 +116,7 @@
                     set_status_on_exception=False,
                 ) as span:
                     # wrap the context
-                    context = _OpenTelemetryServicerContext(context, span)
+                    context = _OpenTelemetryAioServicerContext(context, span)
 
                     # And now we run the actual RPC.
                     try:
@@ -91,7 +141,7 @@
                     context,
                     set_status_on_exception=False,
                 ) as span:
-                    context = _OpenTelemetryServicerContext(context, span)
+                    context = _OpenTelemetryAioServicerContext(context, span)
 
                     try:
                         async for response in behavior(

{"golden_diff": "diff --git a/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py b/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py\n--- a/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py\n+++ b/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py\n@@ -12,13 +12,63 @@\n # See the License for the specific language governing permissions and\n # limitations under the License.\n \n+import grpc\n import grpc.aio\n-\n-from ._server import (\n-    OpenTelemetryServerInterceptor,\n-    _OpenTelemetryServicerContext,\n-    _wrap_rpc_behavior,\n-)\n+import wrapt\n+\n+from opentelemetry.semconv.trace import SpanAttributes\n+from opentelemetry.trace.status import Status, StatusCode\n+\n+from ._server import OpenTelemetryServerInterceptor, _wrap_rpc_behavior\n+\n+\n+# pylint:disable=abstract-method\n+class _OpenTelemetryAioServicerContext(wrapt.ObjectProxy):\n+    def __init__(self, servicer_context, active_span):\n+        super().__init__(servicer_context)\n+        self._self_active_span = active_span\n+        self._self_code = grpc.StatusCode.OK\n+        self._self_details = None\n+\n+    async def abort(self, code, details=\"\", trailing_metadata=tuple()):\n+        self._self_code = code\n+        self._self_details = details\n+        self._self_active_span.set_attribute(\n+            SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]\n+        )\n+        self._self_active_span.set_status(\n+            Status(\n+                status_code=StatusCode.ERROR,\n+                description=f\"{code}:{details}\",\n+            )\n+        )\n+        return await self.__wrapped__.abort(code, details, trailing_metadata)\n+\n+    def set_code(self, code):\n+        self._self_code = code\n+        details = self._self_details or code.value[1]\n+        self._self_active_span.set_attribute(\n+            SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]\n+        )\n+        if code != grpc.StatusCode.OK:\n+            self._self_active_span.set_status(\n+                Status(\n+                    status_code=StatusCode.ERROR,\n+                    description=f\"{code}:{details}\",\n+                )\n+            )\n+        return self.__wrapped__.set_code(code)\n+\n+    def set_details(self, details):\n+        self._self_details = details\n+        if self._self_code != grpc.StatusCode.OK:\n+            self._self_active_span.set_status(\n+                Status(\n+                    status_code=StatusCode.ERROR,\n+                    description=f\"{self._self_code}:{details}\",\n+                )\n+            )\n+        return self.__wrapped__.set_details(details)\n \n \n class OpenTelemetryAioServerInterceptor(\n@@ -66,7 +116,7 @@\n                     set_status_on_exception=False,\n                 ) as span:\n                     # wrap the context\n-                    context = _OpenTelemetryServicerContext(context, span)\n+                    context = _OpenTelemetryAioServicerContext(context, span)\n \n                     # And now we run the actual RPC.\n                     try:\n@@ -91,7 +141,7 @@\n                     context,\n                     set_status_on_exception=False,\n                 ) as span:\n-                    context = _OpenTelemetryServicerContext(context, span)\n+                    context = _OpenTelemetryAioServicerContext(context, span)\n \n                     try:\n                         async for response in behavior(\n", "issue": "grpc: trailing metadata can't be passed when aborting an AIO server RPC\n**Describe your environment**\r\n\r\nPython: 3.12.0\r\n`opentelemetry-distro[otlp]==0.42b0`\r\n`opentelemetry-instrumentation-grpc==0.42b0`\r\n`grpcio==1.59.2`\r\n\r\n**Steps to reproduce**\r\nFull replication can be found [here](https://github.com/cookiefission/testbench/tree/main/otel/python-grpc-aio-abort-issue).\r\n\r\n`server.py`:\r\n\r\n```python\r\nimport asyncio\r\nimport logging\r\n\r\nimport grpc\r\nimport helloworld_pb2\r\nimport helloworld_pb2_grpc\r\n\r\n\r\nclass Greeter(helloworld_pb2_grpc.GreeterServicer):\r\n    async def SayHello(\r\n        self,\r\n        request: helloworld_pb2.HelloRequest,\r\n        context: grpc.aio.ServicerContext,\r\n    ) -> helloworld_pb2.HelloReply:\r\n        metadata = (\r\n            (\"this\", \"should\"),\r\n            (\"work\", \"nicely\")\r\n        )\r\n        await context.abort(code=grpc.StatusCode.ABORTED, details=\"This is the actual error message\", trailing_metadata=metadata)\r\n\r\n\r\nasync def serve() -> None:\r\n    server = grpc.aio.server()\r\n    helloworld_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)\r\n    listen_addr = \"[::]:50051\"\r\n    server.add_insecure_port(listen_addr)\r\n    logging.info(\"Starting server on %s\", listen_addr)\r\n    await server.start()\r\n    await server.wait_for_termination()\r\n\r\n\r\nif __name__ == \"__main__\":\r\n    logging.basicConfig(level=logging.INFO)\r\n    asyncio.run(serve())\r\n```\r\nThis server is just adopted from [the grpc examples](https://github.com/grpc/grpc/blob/master/examples/python/helloworld/async_greeter_server.py). The helloworld protobuf artefacts can be found [there](https://github.com/grpc/grpc/blob/master/examples/python/helloworld/helloworld_pb2_grpc.py).\r\n\r\nFor a client, the [`async_greeter_client.py`](https://github.com/grpc/grpc/blob/master/examples/python/helloworld/async_greeter_client.py) from grpc examples can be used.\r\n\r\nThis only happens when `trailing_metadata` is passed as an argument to `context.abort`. \r\n\r\n**What is the expected behavior?**\r\n\r\nRunning the server without opentelemetry, the (expected) error from running the client is:\r\n\r\n```\r\ngrpc.aio._call.AioRpcError: <AioRpcError of RPC that terminated with:\r\n        status = StatusCode.ABORTED\r\n        details = \"This is the actual error message\"\r\n        debug_error_string = \"UNKNOWN:Error received from peer  {grpc_message:\"This is the actual error message\", grpc_status:10, created_time:\"2023-11-16T15:09:03.552303+00:00\"}\"\r\n>\r\n```\r\n\r\nWhen running the server under `opentelemetry-instrument`, I expect the same output as when it is run without OTEL. \r\n\r\n**What is the actual behavior?**\r\n\r\nInstead, an exception is still raised but the underlying code, details, and trailing_metadata are missing. This makes any nuanced error handling on the client side impossible. \r\n\r\n```\r\ngrpc.aio._call.AioRpcError: <AioRpcError of RPC that terminated with:\r\n        status = StatusCode.UNKNOWN\r\n        details = \"Unexpected <class 'TypeError'>: _OpenTelemetryServicerContext.abort() got an unexpected keyword argument 'trailing_metadata'\"\r\n        debug_error_string = \"UNKNOWN:Error received from peer  {created_time:\"2023-11-16T15:30:02.017999+00:00\", grpc_status:2, grpc_message:\"Unexpected <class \\'TypeError\\'>: _OpenTelemetryServicerContext.abort() got an unexpected keyword argument \\'trailing_metadata\\'\"}\"\r\n>\r\n```\r\n\r\n**Additional context**\r\n\r\nThe `abort` method for [`grpc.ServicerContext`](https://grpc.github.io/grpc/python/grpc.html#grpc.ServicerContext.abort) and [`grpc.aio.ServicerContext`](https://grpc.github.io/grpc/python/grpc_asyncio.html#grpc.aio.ServicerContext.abort) is different. It's unclear why. The AIO instrumentation [re-uses](https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py#L19) the [`_OpenTelemetryServicerContext`](https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py#L68) from the non-async server instrumentation which is how this bug snuck in, despite there being a test in place for the abort. \r\n\n", "before_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport grpc.aio\n\nfrom ._server import (\n    OpenTelemetryServerInterceptor,\n    _OpenTelemetryServicerContext,\n    _wrap_rpc_behavior,\n)\n\n\nclass OpenTelemetryAioServerInterceptor(\n    grpc.aio.ServerInterceptor, OpenTelemetryServerInterceptor\n):\n    \"\"\"\n    An AsyncIO gRPC server interceptor, to add OpenTelemetry.\n    Usage::\n        tracer = some OpenTelemetry tracer\n        interceptors = [\n            AsyncOpenTelemetryServerInterceptor(tracer),\n        ]\n        server = aio.server(\n            futures.ThreadPoolExecutor(max_workers=concurrency),\n            interceptors = (interceptors,))\n    \"\"\"\n\n    async def intercept_service(self, continuation, handler_call_details):\n        if self._filter is not None and not self._filter(handler_call_details):\n            return await continuation(handler_call_details)\n\n        def telemetry_wrapper(behavior, request_streaming, response_streaming):\n            # handle streaming responses specially\n            if response_streaming:\n                return self._intercept_aio_server_stream(\n                    behavior,\n                    handler_call_details,\n                )\n\n            return self._intercept_aio_server_unary(\n                behavior,\n                handler_call_details,\n            )\n\n        next_handler = await continuation(handler_call_details)\n\n        return _wrap_rpc_behavior(next_handler, telemetry_wrapper)\n\n    def _intercept_aio_server_unary(self, behavior, handler_call_details):\n        async def _unary_interceptor(request_or_iterator, context):\n            with self._set_remote_context(context):\n                with self._start_span(\n                    handler_call_details,\n                    context,\n                    set_status_on_exception=False,\n                ) as span:\n                    # wrap the context\n                    context = _OpenTelemetryServicerContext(context, span)\n\n                    # And now we run the actual RPC.\n                    try:\n                        return await behavior(request_or_iterator, context)\n\n                    except Exception as error:\n                        # Bare exceptions are likely to be gRPC aborts, which\n                        # we handle in our context wrapper.\n                        # Here, we're interested in uncaught exceptions.\n                        # pylint:disable=unidiomatic-typecheck\n                        if type(error) != Exception:  # noqa: E721\n                            span.record_exception(error)\n                        raise error\n\n        return _unary_interceptor\n\n    def _intercept_aio_server_stream(self, behavior, handler_call_details):\n        async def _stream_interceptor(request_or_iterator, context):\n            with self._set_remote_context(context):\n                with self._start_span(\n                    handler_call_details,\n                    context,\n                    set_status_on_exception=False,\n                ) as span:\n                    context = _OpenTelemetryServicerContext(context, span)\n\n                    try:\n                        async for response in behavior(\n                            request_or_iterator, context\n                        ):\n                            yield response\n\n                    except Exception as error:\n                        # pylint:disable=unidiomatic-typecheck\n                        if type(error) != Exception:  # noqa: E721\n                            span.record_exception(error)\n                        raise error\n\n        return _stream_interceptor\n", "path": "instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py"}], "after_files": [{"content": "# Copyright The OpenTelemetry Authors\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport grpc\nimport grpc.aio\nimport wrapt\n\nfrom opentelemetry.semconv.trace import SpanAttributes\nfrom opentelemetry.trace.status import Status, StatusCode\n\nfrom ._server import OpenTelemetryServerInterceptor, _wrap_rpc_behavior\n\n\n# pylint:disable=abstract-method\nclass _OpenTelemetryAioServicerContext(wrapt.ObjectProxy):\n    def __init__(self, servicer_context, active_span):\n        super().__init__(servicer_context)\n        self._self_active_span = active_span\n        self._self_code = grpc.StatusCode.OK\n        self._self_details = None\n\n    async def abort(self, code, details=\"\", trailing_metadata=tuple()):\n        self._self_code = code\n        self._self_details = details\n        self._self_active_span.set_attribute(\n            SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]\n        )\n        self._self_active_span.set_status(\n            Status(\n                status_code=StatusCode.ERROR,\n                description=f\"{code}:{details}\",\n            )\n        )\n        return await self.__wrapped__.abort(code, details, trailing_metadata)\n\n    def set_code(self, code):\n        self._self_code = code\n        details = self._self_details or code.value[1]\n        self._self_active_span.set_attribute(\n            SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]\n        )\n        if code != grpc.StatusCode.OK:\n            self._self_active_span.set_status(\n                Status(\n                    status_code=StatusCode.ERROR,\n                    description=f\"{code}:{details}\",\n                )\n            )\n        return self.__wrapped__.set_code(code)\n\n    def set_details(self, details):\n        self._self_details = details\n        if self._self_code != grpc.StatusCode.OK:\n            self._self_active_span.set_status(\n                Status(\n                    status_code=StatusCode.ERROR,\n                    description=f\"{self._self_code}:{details}\",\n                )\n            )\n        return self.__wrapped__.set_details(details)\n\n\nclass OpenTelemetryAioServerInterceptor(\n    grpc.aio.ServerInterceptor, OpenTelemetryServerInterceptor\n):\n    \"\"\"\n    An AsyncIO gRPC server interceptor, to add OpenTelemetry.\n    Usage::\n        tracer = some OpenTelemetry tracer\n        interceptors = [\n            AsyncOpenTelemetryServerInterceptor(tracer),\n        ]\n        server = aio.server(\n            futures.ThreadPoolExecutor(max_workers=concurrency),\n            interceptors = (interceptors,))\n    \"\"\"\n\n    async def intercept_service(self, continuation, handler_call_details):\n        if self._filter is not None and not self._filter(handler_call_details):\n            return await continuation(handler_call_details)\n\n        def telemetry_wrapper(behavior, request_streaming, response_streaming):\n            # handle streaming responses specially\n            if response_streaming:\n                return self._intercept_aio_server_stream(\n                    behavior,\n                    handler_call_details,\n                )\n\n            return self._intercept_aio_server_unary(\n                behavior,\n                handler_call_details,\n            )\n\n        next_handler = await continuation(handler_call_details)\n\n        return _wrap_rpc_behavior(next_handler, telemetry_wrapper)\n\n    def _intercept_aio_server_unary(self, behavior, handler_call_details):\n        async def _unary_interceptor(request_or_iterator, context):\n            with self._set_remote_context(context):\n                with self._start_span(\n                    handler_call_details,\n                    context,\n                    set_status_on_exception=False,\n                ) as span:\n                    # wrap the context\n                    context = _OpenTelemetryAioServicerContext(context, span)\n\n                    # And now we run the actual RPC.\n                    try:\n                        return await behavior(request_or_iterator, context)\n\n                    except Exception as error:\n                        # Bare exceptions are likely to be gRPC aborts, which\n                        # we handle in our context wrapper.\n                        # Here, we're interested in uncaught exceptions.\n                        # pylint:disable=unidiomatic-typecheck\n                        if type(error) != Exception:  # noqa: E721\n                            span.record_exception(error)\n                        raise error\n\n        return _unary_interceptor\n\n    def _intercept_aio_server_stream(self, behavior, handler_call_details):\n        async def _stream_interceptor(request_or_iterator, context):\n            with self._set_remote_context(context):\n                with self._start_span(\n                    handler_call_details,\n                    context,\n                    set_status_on_exception=False,\n                ) as span:\n                    context = _OpenTelemetryAioServicerContext(context, span)\n\n                    try:\n                        async for response in behavior(\n                            request_or_iterator, context\n                        ):\n                            yield response\n\n                    except Exception as error:\n                        # pylint:disable=unidiomatic-typecheck\n                        if type(error) != Exception:  # noqa: E721\n                            span.record_exception(error)\n                        raise error\n\n        return _stream_interceptor\n", "path": "instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_server.py"}]}