We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
gRPC server instrumentation creates multiple traces on streaming requests
**Environment**
Current `master` code, basically the sample code in the documentation, testing with a unary request vs. a streaming request.

**Steps to reproduce**
Create a simple gRPC servicer with two RPCs, one which returns a single message (the unary response), and one which yields items in a list for a streaming response.

The key here is to make an instrumented request within the primary request handler (I'm using a simple HTTP get with the Requests instrumentation), so you get an _additional_ span which should be attached to the same trace.

**What is the expected behavior?**
A single trace with the main span, and a second child span for the HTTP request.

**What is the actual behavior?**
Two separate traces, each containing a single span.

**Additional context**
The problem _only_ occurs on streaming requests - I'm sure the reworking I did as part of https://github.com/open-telemetry/opentelemetry-python/pull/1171 is where the problem started, I didn't take into account the streaming case specifically with multiple spans, and naturally, there are no tests for anything streaming, only unary responses.

So as part of this, we'll need some useful tests as well.  I'll see if I can write up my test case as an actual test case.

And again, I've got a vested interest in this working, so I'll have a PR up soon.

--- 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/_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 # pylint:disable=relative-beyond-top-level
16 # pylint:disable=arguments-differ
17 # pylint:disable=no-member
18 # pylint:disable=signature-differs
19 
20 """
21 Implementation of the service-side open-telemetry interceptor.
22 """
23 
24 import logging
25 from contextlib import contextmanager
26 
27 import grpc
28 
29 from opentelemetry import propagators, trace
30 from opentelemetry.context import attach, detach
31 from opentelemetry.trace.propagation.textmap import DictGetter
32 from opentelemetry.trace.status import Status, StatusCode
33 
34 logger = logging.getLogger(__name__)
35 
36 
37 # wrap an RPC call
38 # see https://github.com/grpc/grpc/issues/18191
39 def _wrap_rpc_behavior(handler, continuation):
40     if handler is None:
41         return None
42 
43     if handler.request_streaming and handler.response_streaming:
44         behavior_fn = handler.stream_stream
45         handler_factory = grpc.stream_stream_rpc_method_handler
46     elif handler.request_streaming and not handler.response_streaming:
47         behavior_fn = handler.stream_unary
48         handler_factory = grpc.stream_unary_rpc_method_handler
49     elif not handler.request_streaming and handler.response_streaming:
50         behavior_fn = handler.unary_stream
51         handler_factory = grpc.unary_stream_rpc_method_handler
52     else:
53         behavior_fn = handler.unary_unary
54         handler_factory = grpc.unary_unary_rpc_method_handler
55 
56     return handler_factory(
57         continuation(
58             behavior_fn, handler.request_streaming, handler.response_streaming
59         ),
60         request_deserializer=handler.request_deserializer,
61         response_serializer=handler.response_serializer,
62     )
63 
64 
65 # pylint:disable=abstract-method
66 class _OpenTelemetryServicerContext(grpc.ServicerContext):
67     def __init__(self, servicer_context, active_span):
68         self._servicer_context = servicer_context
69         self._active_span = active_span
70         self.code = grpc.StatusCode.OK
71         self.details = None
72         super().__init__()
73 
74     def is_active(self, *args, **kwargs):
75         return self._servicer_context.is_active(*args, **kwargs)
76 
77     def time_remaining(self, *args, **kwargs):
78         return self._servicer_context.time_remaining(*args, **kwargs)
79 
80     def cancel(self, *args, **kwargs):
81         return self._servicer_context.cancel(*args, **kwargs)
82 
83     def add_callback(self, *args, **kwargs):
84         return self._servicer_context.add_callback(*args, **kwargs)
85 
86     def disable_next_message_compression(self):
87         return self._service_context.disable_next_message_compression()
88 
89     def invocation_metadata(self, *args, **kwargs):
90         return self._servicer_context.invocation_metadata(*args, **kwargs)
91 
92     def peer(self):
93         return self._servicer_context.peer()
94 
95     def peer_identities(self):
96         return self._servicer_context.peer_identities()
97 
98     def peer_identity_key(self):
99         return self._servicer_context.peer_identity_key()
100 
101     def auth_context(self):
102         return self._servicer_context.auth_context()
103 
104     def set_compression(self, compression):
105         return self._servicer_context.set_compression(compression)
106 
107     def send_initial_metadata(self, *args, **kwargs):
108         return self._servicer_context.send_initial_metadata(*args, **kwargs)
109 
110     def set_trailing_metadata(self, *args, **kwargs):
111         return self._servicer_context.set_trailing_metadata(*args, **kwargs)
112 
113     def abort(self, code, details):
114         self.code = code
115         self.details = details
116         self._active_span.set_attribute("rpc.grpc.status_code", code.value[0])
117         self._active_span.set_status(
118             Status(
119                 status_code=StatusCode.ERROR,
120                 description="{}:{}".format(code, details),
121             )
122         )
123         return self._servicer_context.abort(code, details)
124 
125     def abort_with_status(self, status):
126         return self._servicer_context.abort_with_status(status)
127 
128     def set_code(self, code):
129         self.code = code
130         # use details if we already have it, otherwise the status description
131         details = self.details or code.value[1]
132         self._active_span.set_attribute("rpc.grpc.status_code", code.value[0])
133         if code != grpc.StatusCode.OK:
134             self._active_span.set_status(
135                 Status(
136                     status_code=StatusCode.ERROR,
137                     description="{}:{}".format(code, details),
138                 )
139             )
140         return self._servicer_context.set_code(code)
141 
142     def set_details(self, details):
143         self.details = details
144         if self.code != grpc.StatusCode.OK:
145             self._active_span.set_status(
146                 Status(
147                     status_code=StatusCode.ERROR,
148                     description="{}:{}".format(self.code, details),
149                 )
150             )
151         return self._servicer_context.set_details(details)
152 
153 
154 # pylint:disable=abstract-method
155 # pylint:disable=no-self-use
156 # pylint:disable=unused-argument
157 class OpenTelemetryServerInterceptor(grpc.ServerInterceptor):
158     """
159     A gRPC server interceptor, to add OpenTelemetry.
160 
161     Usage::
162 
163         tracer = some OpenTelemetry tracer
164 
165         interceptors = [
166             OpenTelemetryServerInterceptor(tracer),
167         ]
168 
169         server = grpc.server(
170             futures.ThreadPoolExecutor(max_workers=concurrency),
171             interceptors = interceptors)
172 
173     """
174 
175     def __init__(self, tracer):
176         self._tracer = tracer
177         self._carrier_getter = DictGetter()
178 
179     @contextmanager
180     def _set_remote_context(self, servicer_context):
181         metadata = servicer_context.invocation_metadata()
182         if metadata:
183             md_dict = {md.key: md.value for md in metadata}
184             ctx = propagators.extract(self._carrier_getter, md_dict)
185             token = attach(ctx)
186             try:
187                 yield
188             finally:
189                 detach(token)
190         else:
191             yield
192 
193     def _start_span(self, handler_call_details, context):
194 
195         # standard attributes
196         attributes = {
197             "rpc.system": "grpc",
198             "rpc.grpc.status_code": grpc.StatusCode.OK.value[0],
199         }
200 
201         # if we have details about the call, split into service and method
202         if handler_call_details.method:
203             service, method = handler_call_details.method.lstrip("/").split(
204                 "/", 1
205             )
206             attributes.update({"rpc.method": method, "rpc.service": service})
207 
208         # add some attributes from the metadata
209         metadata = dict(context.invocation_metadata())
210         if "user-agent" in metadata:
211             attributes["rpc.user_agent"] = metadata["user-agent"]
212 
213         # Split up the peer to keep with how other telemetry sources
214         # do it.  This looks like:
215         # * ipv6:[::1]:57284
216         # * ipv4:127.0.0.1:57284
217         # * ipv4:10.2.1.1:57284,127.0.0.1:57284
218         #
219         try:
220             ip, port = (
221                 context.peer().split(",")[0].split(":", 1)[1].rsplit(":", 1)
222             )
223             attributes.update({"net.peer.ip": ip, "net.peer.port": port})
224 
225             # other telemetry sources add this, so we will too
226             if ip in ("[::1]", "127.0.0.1"):
227                 attributes["net.peer.name"] = "localhost"
228 
229         except IndexError:
230             logger.warning("Failed to parse peer address '%s'", context.peer())
231 
232         return self._tracer.start_as_current_span(
233             name=handler_call_details.method,
234             kind=trace.SpanKind.SERVER,
235             attributes=attributes,
236         )
237 
238     def intercept_service(self, continuation, handler_call_details):
239         def telemetry_wrapper(behavior, request_streaming, response_streaming):
240             def telemetry_interceptor(request_or_iterator, context):
241 
242                 with self._set_remote_context(context):
243                     with self._start_span(
244                         handler_call_details, context
245                     ) as span:
246                         # wrap the context
247                         context = _OpenTelemetryServicerContext(context, span)
248 
249                         # And now we run the actual RPC.
250                         try:
251                             return behavior(request_or_iterator, context)
252                         except Exception as error:
253                             # Bare exceptions are likely to be gRPC aborts, which
254                             # we handle in our context wrapper.
255                             # Here, we're interested in uncaught exceptions.
256                             # pylint:disable=unidiomatic-typecheck
257                             if type(error) != Exception:
258                                 span.record_exception(error)
259                             raise error
260 
261             return telemetry_interceptor
262 
263         return _wrap_rpc_behavior(
264             continuation(handler_call_details), telemetry_wrapper
265         )
266
```
--- 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/_server.py b/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py
--- a/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py
+++ b/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py
@@ -239,6 +239,15 @@
         def telemetry_wrapper(behavior, request_streaming, response_streaming):
             def telemetry_interceptor(request_or_iterator, context):
 
+                # handle streaming responses specially
+                if response_streaming:
+                    return self._intercept_server_stream(
+                        behavior,
+                        handler_call_details,
+                        request_or_iterator,
+                        context,
+                    )
+
                 with self._set_remote_context(context):
                     with self._start_span(
                         handler_call_details, context
@@ -249,6 +258,7 @@
                         # And now we run the actual RPC.
                         try:
                             return behavior(request_or_iterator, context)
+
                         except Exception as error:
                             # Bare exceptions are likely to be gRPC aborts, which
                             # we handle in our context wrapper.
@@ -263,3 +273,23 @@
         return _wrap_rpc_behavior(
             continuation(handler_call_details), telemetry_wrapper
         )
+
+    # Handle streaming responses separately - we have to do this
+    # to return a *new* generator or various upstream things
+    # get confused, or we'll lose the consistent trace
+    def _intercept_server_stream(
+        self, behavior, handler_call_details, request_or_iterator, context
+    ):
+
+        with self._set_remote_context(context):
+            with self._start_span(handler_call_details, context) as span:
+                context = _OpenTelemetryServicerContext(context, span)
+
+                try:
+                    yield from behavior(request_or_iterator, context)
+
+                except Exception as error:
+                    # pylint:disable=unidiomatic-typecheck
+                    if type(error) != Exception:
+                        span.record_exception(error)
+                    raise error

{"golden_diff": "diff --git a/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py b/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py\n--- a/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py\n+++ b/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py\n@@ -239,6 +239,15 @@\n         def telemetry_wrapper(behavior, request_streaming, response_streaming):\n             def telemetry_interceptor(request_or_iterator, context):\n \n+                # handle streaming responses specially\n+                if response_streaming:\n+                    return self._intercept_server_stream(\n+                        behavior,\n+                        handler_call_details,\n+                        request_or_iterator,\n+                        context,\n+                    )\n+\n                 with self._set_remote_context(context):\n                     with self._start_span(\n                         handler_call_details, context\n@@ -249,6 +258,7 @@\n                         # And now we run the actual RPC.\n                         try:\n                             return 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@@ -263,3 +273,23 @@\n         return _wrap_rpc_behavior(\n             continuation(handler_call_details), telemetry_wrapper\n         )\n+\n+    # Handle streaming responses separately - we have to do this\n+    # to return a *new* generator or various upstream things\n+    # get confused, or we'll lose the consistent trace\n+    def _intercept_server_stream(\n+        self, behavior, handler_call_details, request_or_iterator, context\n+    ):\n+\n+        with self._set_remote_context(context):\n+            with self._start_span(handler_call_details, context) as span:\n+                context = _OpenTelemetryServicerContext(context, span)\n+\n+                try:\n+                    yield from behavior(request_or_iterator, context)\n+\n+                except Exception as error:\n+                    # pylint:disable=unidiomatic-typecheck\n+                    if type(error) != Exception:\n+                        span.record_exception(error)\n+                    raise error\n", "issue": "gRPC server instrumentation creates multiple traces on streaming requests\n**Environment**\r\nCurrent `master` code, basically the sample code in the documentation, testing with a unary request vs. a streaming request.\r\n\r\n**Steps to reproduce**\r\nCreate a simple gRPC servicer with two RPCs, one which returns a single message (the unary response), and one which yields items in a list for a streaming response.\r\n\r\nThe key here is to make an instrumented request within the primary request handler (I'm using a simple HTTP get with the Requests instrumentation), so you get an _additional_ span which should be attached to the same trace.\r\n\r\n**What is the expected behavior?**\r\nA single trace with the main span, and a second child span for the HTTP request.\r\n\r\n**What is the actual behavior?**\r\nTwo separate traces, each containing a single span.\r\n\r\n**Additional context**\r\nThe problem _only_ occurs on streaming requests - I'm sure the reworking I did as part of https://github.com/open-telemetry/opentelemetry-python/pull/1171 is where the problem started, I didn't take into account the streaming case specifically with multiple spans, and naturally, there are no tests for anything streaming, only unary responses.\r\n\r\nSo as part of this, we'll need some useful tests as well.  I'll see if I can write up my test case as an actual test case.\r\n\r\nAnd again, I've got a vested interest in this working, so I'll have a PR up soon.\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\n# pylint:disable=relative-beyond-top-level\n# pylint:disable=arguments-differ\n# pylint:disable=no-member\n# pylint:disable=signature-differs\n\n\"\"\"\nImplementation of the service-side open-telemetry interceptor.\n\"\"\"\n\nimport logging\nfrom contextlib import contextmanager\n\nimport grpc\n\nfrom opentelemetry import propagators, trace\nfrom opentelemetry.context import attach, detach\nfrom opentelemetry.trace.propagation.textmap import DictGetter\nfrom opentelemetry.trace.status import Status, StatusCode\n\nlogger = logging.getLogger(__name__)\n\n\n# wrap an RPC call\n# see https://github.com/grpc/grpc/issues/18191\ndef _wrap_rpc_behavior(handler, continuation):\n    if handler is None:\n        return None\n\n    if handler.request_streaming and handler.response_streaming:\n        behavior_fn = handler.stream_stream\n        handler_factory = grpc.stream_stream_rpc_method_handler\n    elif handler.request_streaming and not handler.response_streaming:\n        behavior_fn = handler.stream_unary\n        handler_factory = grpc.stream_unary_rpc_method_handler\n    elif not handler.request_streaming and handler.response_streaming:\n        behavior_fn = handler.unary_stream\n        handler_factory = grpc.unary_stream_rpc_method_handler\n    else:\n        behavior_fn = handler.unary_unary\n        handler_factory = grpc.unary_unary_rpc_method_handler\n\n    return handler_factory(\n        continuation(\n            behavior_fn, handler.request_streaming, handler.response_streaming\n        ),\n        request_deserializer=handler.request_deserializer,\n        response_serializer=handler.response_serializer,\n    )\n\n\n# pylint:disable=abstract-method\nclass _OpenTelemetryServicerContext(grpc.ServicerContext):\n    def __init__(self, servicer_context, active_span):\n        self._servicer_context = servicer_context\n        self._active_span = active_span\n        self.code = grpc.StatusCode.OK\n        self.details = None\n        super().__init__()\n\n    def is_active(self, *args, **kwargs):\n        return self._servicer_context.is_active(*args, **kwargs)\n\n    def time_remaining(self, *args, **kwargs):\n        return self._servicer_context.time_remaining(*args, **kwargs)\n\n    def cancel(self, *args, **kwargs):\n        return self._servicer_context.cancel(*args, **kwargs)\n\n    def add_callback(self, *args, **kwargs):\n        return self._servicer_context.add_callback(*args, **kwargs)\n\n    def disable_next_message_compression(self):\n        return self._service_context.disable_next_message_compression()\n\n    def invocation_metadata(self, *args, **kwargs):\n        return self._servicer_context.invocation_metadata(*args, **kwargs)\n\n    def peer(self):\n        return self._servicer_context.peer()\n\n    def peer_identities(self):\n        return self._servicer_context.peer_identities()\n\n    def peer_identity_key(self):\n        return self._servicer_context.peer_identity_key()\n\n    def auth_context(self):\n        return self._servicer_context.auth_context()\n\n    def set_compression(self, compression):\n        return self._servicer_context.set_compression(compression)\n\n    def send_initial_metadata(self, *args, **kwargs):\n        return self._servicer_context.send_initial_metadata(*args, **kwargs)\n\n    def set_trailing_metadata(self, *args, **kwargs):\n        return self._servicer_context.set_trailing_metadata(*args, **kwargs)\n\n    def abort(self, code, details):\n        self.code = code\n        self.details = details\n        self._active_span.set_attribute(\"rpc.grpc.status_code\", code.value[0])\n        self._active_span.set_status(\n            Status(\n                status_code=StatusCode.ERROR,\n                description=\"{}:{}\".format(code, details),\n            )\n        )\n        return self._servicer_context.abort(code, details)\n\n    def abort_with_status(self, status):\n        return self._servicer_context.abort_with_status(status)\n\n    def set_code(self, code):\n        self.code = code\n        # use details if we already have it, otherwise the status description\n        details = self.details or code.value[1]\n        self._active_span.set_attribute(\"rpc.grpc.status_code\", code.value[0])\n        if code != grpc.StatusCode.OK:\n            self._active_span.set_status(\n                Status(\n                    status_code=StatusCode.ERROR,\n                    description=\"{}:{}\".format(code, details),\n                )\n            )\n        return self._servicer_context.set_code(code)\n\n    def set_details(self, details):\n        self.details = details\n        if self.code != grpc.StatusCode.OK:\n            self._active_span.set_status(\n                Status(\n                    status_code=StatusCode.ERROR,\n                    description=\"{}:{}\".format(self.code, details),\n                )\n            )\n        return self._servicer_context.set_details(details)\n\n\n# pylint:disable=abstract-method\n# pylint:disable=no-self-use\n# pylint:disable=unused-argument\nclass OpenTelemetryServerInterceptor(grpc.ServerInterceptor):\n    \"\"\"\n    A gRPC server interceptor, to add OpenTelemetry.\n\n    Usage::\n\n        tracer = some OpenTelemetry tracer\n\n        interceptors = [\n            OpenTelemetryServerInterceptor(tracer),\n        ]\n\n        server = grpc.server(\n            futures.ThreadPoolExecutor(max_workers=concurrency),\n            interceptors = interceptors)\n\n    \"\"\"\n\n    def __init__(self, tracer):\n        self._tracer = tracer\n        self._carrier_getter = DictGetter()\n\n    @contextmanager\n    def _set_remote_context(self, servicer_context):\n        metadata = servicer_context.invocation_metadata()\n        if metadata:\n            md_dict = {md.key: md.value for md in metadata}\n            ctx = propagators.extract(self._carrier_getter, md_dict)\n            token = attach(ctx)\n            try:\n                yield\n            finally:\n                detach(token)\n        else:\n            yield\n\n    def _start_span(self, handler_call_details, context):\n\n        # standard attributes\n        attributes = {\n            \"rpc.system\": \"grpc\",\n            \"rpc.grpc.status_code\": grpc.StatusCode.OK.value[0],\n        }\n\n        # if we have details about the call, split into service and method\n        if handler_call_details.method:\n            service, method = handler_call_details.method.lstrip(\"/\").split(\n                \"/\", 1\n            )\n            attributes.update({\"rpc.method\": method, \"rpc.service\": service})\n\n        # add some attributes from the metadata\n        metadata = dict(context.invocation_metadata())\n        if \"user-agent\" in metadata:\n            attributes[\"rpc.user_agent\"] = metadata[\"user-agent\"]\n\n        # Split up the peer to keep with how other telemetry sources\n        # do it.  This looks like:\n        # * ipv6:[::1]:57284\n        # * ipv4:127.0.0.1:57284\n        # * ipv4:10.2.1.1:57284,127.0.0.1:57284\n        #\n        try:\n            ip, port = (\n                context.peer().split(\",\")[0].split(\":\", 1)[1].rsplit(\":\", 1)\n            )\n            attributes.update({\"net.peer.ip\": ip, \"net.peer.port\": port})\n\n            # other telemetry sources add this, so we will too\n            if ip in (\"[::1]\", \"127.0.0.1\"):\n                attributes[\"net.peer.name\"] = \"localhost\"\n\n        except IndexError:\n            logger.warning(\"Failed to parse peer address '%s'\", context.peer())\n\n        return self._tracer.start_as_current_span(\n            name=handler_call_details.method,\n            kind=trace.SpanKind.SERVER,\n            attributes=attributes,\n        )\n\n    def intercept_service(self, continuation, handler_call_details):\n        def telemetry_wrapper(behavior, request_streaming, response_streaming):\n            def telemetry_interceptor(request_or_iterator, context):\n\n                with self._set_remote_context(context):\n                    with self._start_span(\n                        handler_call_details, context\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 behavior(request_or_iterator, context)\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:\n                                span.record_exception(error)\n                            raise error\n\n            return telemetry_interceptor\n\n        return _wrap_rpc_behavior(\n            continuation(handler_call_details), telemetry_wrapper\n        )\n", "path": "instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_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\n# pylint:disable=relative-beyond-top-level\n# pylint:disable=arguments-differ\n# pylint:disable=no-member\n# pylint:disable=signature-differs\n\n\"\"\"\nImplementation of the service-side open-telemetry interceptor.\n\"\"\"\n\nimport logging\nfrom contextlib import contextmanager\n\nimport grpc\n\nfrom opentelemetry import propagators, trace\nfrom opentelemetry.context import attach, detach\nfrom opentelemetry.trace.propagation.textmap import DictGetter\nfrom opentelemetry.trace.status import Status, StatusCode\n\nlogger = logging.getLogger(__name__)\n\n\n# wrap an RPC call\n# see https://github.com/grpc/grpc/issues/18191\ndef _wrap_rpc_behavior(handler, continuation):\n    if handler is None:\n        return None\n\n    if handler.request_streaming and handler.response_streaming:\n        behavior_fn = handler.stream_stream\n        handler_factory = grpc.stream_stream_rpc_method_handler\n    elif handler.request_streaming and not handler.response_streaming:\n        behavior_fn = handler.stream_unary\n        handler_factory = grpc.stream_unary_rpc_method_handler\n    elif not handler.request_streaming and handler.response_streaming:\n        behavior_fn = handler.unary_stream\n        handler_factory = grpc.unary_stream_rpc_method_handler\n    else:\n        behavior_fn = handler.unary_unary\n        handler_factory = grpc.unary_unary_rpc_method_handler\n\n    return handler_factory(\n        continuation(\n            behavior_fn, handler.request_streaming, handler.response_streaming\n        ),\n        request_deserializer=handler.request_deserializer,\n        response_serializer=handler.response_serializer,\n    )\n\n\n# pylint:disable=abstract-method\nclass _OpenTelemetryServicerContext(grpc.ServicerContext):\n    def __init__(self, servicer_context, active_span):\n        self._servicer_context = servicer_context\n        self._active_span = active_span\n        self.code = grpc.StatusCode.OK\n        self.details = None\n        super().__init__()\n\n    def is_active(self, *args, **kwargs):\n        return self._servicer_context.is_active(*args, **kwargs)\n\n    def time_remaining(self, *args, **kwargs):\n        return self._servicer_context.time_remaining(*args, **kwargs)\n\n    def cancel(self, *args, **kwargs):\n        return self._servicer_context.cancel(*args, **kwargs)\n\n    def add_callback(self, *args, **kwargs):\n        return self._servicer_context.add_callback(*args, **kwargs)\n\n    def disable_next_message_compression(self):\n        return self._service_context.disable_next_message_compression()\n\n    def invocation_metadata(self, *args, **kwargs):\n        return self._servicer_context.invocation_metadata(*args, **kwargs)\n\n    def peer(self):\n        return self._servicer_context.peer()\n\n    def peer_identities(self):\n        return self._servicer_context.peer_identities()\n\n    def peer_identity_key(self):\n        return self._servicer_context.peer_identity_key()\n\n    def auth_context(self):\n        return self._servicer_context.auth_context()\n\n    def set_compression(self, compression):\n        return self._servicer_context.set_compression(compression)\n\n    def send_initial_metadata(self, *args, **kwargs):\n        return self._servicer_context.send_initial_metadata(*args, **kwargs)\n\n    def set_trailing_metadata(self, *args, **kwargs):\n        return self._servicer_context.set_trailing_metadata(*args, **kwargs)\n\n    def abort(self, code, details):\n        self.code = code\n        self.details = details\n        self._active_span.set_attribute(\"rpc.grpc.status_code\", code.value[0])\n        self._active_span.set_status(\n            Status(\n                status_code=StatusCode.ERROR,\n                description=\"{}:{}\".format(code, details),\n            )\n        )\n        return self._servicer_context.abort(code, details)\n\n    def abort_with_status(self, status):\n        return self._servicer_context.abort_with_status(status)\n\n    def set_code(self, code):\n        self.code = code\n        # use details if we already have it, otherwise the status description\n        details = self.details or code.value[1]\n        self._active_span.set_attribute(\"rpc.grpc.status_code\", code.value[0])\n        if code != grpc.StatusCode.OK:\n            self._active_span.set_status(\n                Status(\n                    status_code=StatusCode.ERROR,\n                    description=\"{}:{}\".format(code, details),\n                )\n            )\n        return self._servicer_context.set_code(code)\n\n    def set_details(self, details):\n        self.details = details\n        if self.code != grpc.StatusCode.OK:\n            self._active_span.set_status(\n                Status(\n                    status_code=StatusCode.ERROR,\n                    description=\"{}:{}\".format(self.code, details),\n                )\n            )\n        return self._servicer_context.set_details(details)\n\n\n# pylint:disable=abstract-method\n# pylint:disable=no-self-use\n# pylint:disable=unused-argument\nclass OpenTelemetryServerInterceptor(grpc.ServerInterceptor):\n    \"\"\"\n    A gRPC server interceptor, to add OpenTelemetry.\n\n    Usage::\n\n        tracer = some OpenTelemetry tracer\n\n        interceptors = [\n            OpenTelemetryServerInterceptor(tracer),\n        ]\n\n        server = grpc.server(\n            futures.ThreadPoolExecutor(max_workers=concurrency),\n            interceptors = interceptors)\n\n    \"\"\"\n\n    def __init__(self, tracer):\n        self._tracer = tracer\n        self._carrier_getter = DictGetter()\n\n    @contextmanager\n    def _set_remote_context(self, servicer_context):\n        metadata = servicer_context.invocation_metadata()\n        if metadata:\n            md_dict = {md.key: md.value for md in metadata}\n            ctx = propagators.extract(self._carrier_getter, md_dict)\n            token = attach(ctx)\n            try:\n                yield\n            finally:\n                detach(token)\n        else:\n            yield\n\n    def _start_span(self, handler_call_details, context):\n\n        # standard attributes\n        attributes = {\n            \"rpc.system\": \"grpc\",\n            \"rpc.grpc.status_code\": grpc.StatusCode.OK.value[0],\n        }\n\n        # if we have details about the call, split into service and method\n        if handler_call_details.method:\n            service, method = handler_call_details.method.lstrip(\"/\").split(\n                \"/\", 1\n            )\n            attributes.update({\"rpc.method\": method, \"rpc.service\": service})\n\n        # add some attributes from the metadata\n        metadata = dict(context.invocation_metadata())\n        if \"user-agent\" in metadata:\n            attributes[\"rpc.user_agent\"] = metadata[\"user-agent\"]\n\n        # Split up the peer to keep with how other telemetry sources\n        # do it.  This looks like:\n        # * ipv6:[::1]:57284\n        # * ipv4:127.0.0.1:57284\n        # * ipv4:10.2.1.1:57284,127.0.0.1:57284\n        #\n        try:\n            ip, port = (\n                context.peer().split(\",\")[0].split(\":\", 1)[1].rsplit(\":\", 1)\n            )\n            attributes.update({\"net.peer.ip\": ip, \"net.peer.port\": port})\n\n            # other telemetry sources add this, so we will too\n            if ip in (\"[::1]\", \"127.0.0.1\"):\n                attributes[\"net.peer.name\"] = \"localhost\"\n\n        except IndexError:\n            logger.warning(\"Failed to parse peer address '%s'\", context.peer())\n\n        return self._tracer.start_as_current_span(\n            name=handler_call_details.method,\n            kind=trace.SpanKind.SERVER,\n            attributes=attributes,\n        )\n\n    def intercept_service(self, continuation, handler_call_details):\n        def telemetry_wrapper(behavior, request_streaming, response_streaming):\n            def telemetry_interceptor(request_or_iterator, context):\n\n                # handle streaming responses specially\n                if response_streaming:\n                    return self._intercept_server_stream(\n                        behavior,\n                        handler_call_details,\n                        request_or_iterator,\n                        context,\n                    )\n\n                with self._set_remote_context(context):\n                    with self._start_span(\n                        handler_call_details, context\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 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:\n                                span.record_exception(error)\n                            raise error\n\n            return telemetry_interceptor\n\n        return _wrap_rpc_behavior(\n            continuation(handler_call_details), telemetry_wrapper\n        )\n\n    # Handle streaming responses separately - we have to do this\n    # to return a *new* generator or various upstream things\n    # get confused, or we'll lose the consistent trace\n    def _intercept_server_stream(\n        self, behavior, handler_call_details, request_or_iterator, context\n    ):\n\n        with self._set_remote_context(context):\n            with self._start_span(handler_call_details, context) as span:\n                context = _OpenTelemetryServicerContext(context, span)\n\n                try:\n                    yield from behavior(request_or_iterator, context)\n\n                except Exception as error:\n                    # pylint:disable=unidiomatic-typecheck\n                    if type(error) != Exception:\n                        span.record_exception(error)\n                    raise error\n", "path": "instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py"}]}