We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
gRPC server throws ValueError for Unix sockets
**Describe your environment**

```
Ubuntu 20.04
x86_64
Python 3.10.9
```

**Steps to reproduce**

I'll post a PR that addresses the issue with a test that reproduces the problem shortly.
This problem can be triggered by running the `grpc` instrumentation and utilizing Unix sockets for connections.

```py
server.add_insecure_port("unix:///tmp/grpc.sock")
channel = grpc.insecure_channel("unix:///tmp/grpc.sock")
```

**What is the expected behavior?**

The `grpc` instrumentation should collect traces when running on a Unix socket.

**What is the actual behavior?**

A `ValueError` is thrown here:
https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py#L254-L256

`context.peer()` is `unix:` which results in `ip, port = [""]`.


--- 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 from urllib.parse import unquote
27 
28 import grpc
29 
30 from opentelemetry import trace
31 from opentelemetry.context import attach, detach
32 from opentelemetry.propagate import extract
33 from opentelemetry.semconv.trace import SpanAttributes
34 from opentelemetry.trace.status import Status, StatusCode
35 
36 logger = logging.getLogger(__name__)
37 
38 
39 # wrap an RPC call
40 # see https://github.com/grpc/grpc/issues/18191
41 def _wrap_rpc_behavior(handler, continuation):
42     if handler is None:
43         return None
44 
45     if handler.request_streaming and handler.response_streaming:
46         behavior_fn = handler.stream_stream
47         handler_factory = grpc.stream_stream_rpc_method_handler
48     elif handler.request_streaming and not handler.response_streaming:
49         behavior_fn = handler.stream_unary
50         handler_factory = grpc.stream_unary_rpc_method_handler
51     elif not handler.request_streaming and handler.response_streaming:
52         behavior_fn = handler.unary_stream
53         handler_factory = grpc.unary_stream_rpc_method_handler
54     else:
55         behavior_fn = handler.unary_unary
56         handler_factory = grpc.unary_unary_rpc_method_handler
57 
58     return handler_factory(
59         continuation(
60             behavior_fn, handler.request_streaming, handler.response_streaming
61         ),
62         request_deserializer=handler.request_deserializer,
63         response_serializer=handler.response_serializer,
64     )
65 
66 
67 # pylint:disable=abstract-method
68 class _OpenTelemetryServicerContext(grpc.ServicerContext):
69     def __init__(self, servicer_context, active_span):
70         self._servicer_context = servicer_context
71         self._active_span = active_span
72         self._code = grpc.StatusCode.OK
73         self._details = None
74         super().__init__()
75 
76     def __getattr__(self, attr):
77         return getattr(self._servicer_context, attr)
78 
79     def is_active(self, *args, **kwargs):
80         return self._servicer_context.is_active(*args, **kwargs)
81 
82     def time_remaining(self, *args, **kwargs):
83         return self._servicer_context.time_remaining(*args, **kwargs)
84 
85     def cancel(self, *args, **kwargs):
86         return self._servicer_context.cancel(*args, **kwargs)
87 
88     def add_callback(self, *args, **kwargs):
89         return self._servicer_context.add_callback(*args, **kwargs)
90 
91     def disable_next_message_compression(self):
92         return self._service_context.disable_next_message_compression()
93 
94     def invocation_metadata(self, *args, **kwargs):
95         return self._servicer_context.invocation_metadata(*args, **kwargs)
96 
97     def peer(self):
98         return self._servicer_context.peer()
99 
100     def peer_identities(self):
101         return self._servicer_context.peer_identities()
102 
103     def peer_identity_key(self):
104         return self._servicer_context.peer_identity_key()
105 
106     def auth_context(self):
107         return self._servicer_context.auth_context()
108 
109     def set_compression(self, compression):
110         return self._servicer_context.set_compression(compression)
111 
112     def send_initial_metadata(self, *args, **kwargs):
113         return self._servicer_context.send_initial_metadata(*args, **kwargs)
114 
115     def set_trailing_metadata(self, *args, **kwargs):
116         return self._servicer_context.set_trailing_metadata(*args, **kwargs)
117 
118     def trailing_metadata(self):
119         return self._servicer_context.trailing_metadata()
120 
121     def abort(self, code, details):
122         self._code = code
123         self._details = details
124         self._active_span.set_attribute(
125             SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]
126         )
127         self._active_span.set_status(
128             Status(
129                 status_code=StatusCode.ERROR,
130                 description=f"{code}:{details}",
131             )
132         )
133         return self._servicer_context.abort(code, details)
134 
135     def abort_with_status(self, status):
136         return self._servicer_context.abort_with_status(status)
137 
138     def code(self):
139         if not hasattr(self._servicer_context, "code"):
140             raise RuntimeError(
141                 "code() is not supported with the installed version of grpcio"
142             )
143         return self._servicer_context.code()
144 
145     def details(self):
146         if not hasattr(self._servicer_context, "details"):
147             raise RuntimeError(
148                 "details() is not supported with the installed version of "
149                 "grpcio"
150             )
151         return self._servicer_context.details()
152 
153     def set_code(self, code):
154         self._code = code
155         # use details if we already have it, otherwise the status description
156         details = self._details or code.value[1]
157         self._active_span.set_attribute(
158             SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]
159         )
160         if code != grpc.StatusCode.OK:
161             self._active_span.set_status(
162                 Status(
163                     status_code=StatusCode.ERROR,
164                     description=f"{code}:{details}",
165                 )
166             )
167         return self._servicer_context.set_code(code)
168 
169     def set_details(self, details):
170         self._details = details
171         if self._code != grpc.StatusCode.OK:
172             self._active_span.set_status(
173                 Status(
174                     status_code=StatusCode.ERROR,
175                     description=f"{self._code}:{details}",
176                 )
177             )
178         return self._servicer_context.set_details(details)
179 
180 
181 # pylint:disable=abstract-method
182 # pylint:disable=no-self-use
183 # pylint:disable=unused-argument
184 class OpenTelemetryServerInterceptor(grpc.ServerInterceptor):
185     """
186     A gRPC server interceptor, to add OpenTelemetry.
187 
188     Usage::
189 
190         tracer = some OpenTelemetry tracer
191         filter = filters.negate(filters.method_name("service.Foo"))
192 
193         interceptors = [
194             OpenTelemetryServerInterceptor(tracer, filter),
195         ]
196 
197         server = grpc.server(
198             futures.ThreadPoolExecutor(max_workers=concurrency),
199             interceptors = interceptors)
200 
201     """
202 
203     def __init__(self, tracer, filter_=None):
204         self._tracer = tracer
205         self._filter = filter_
206 
207     @contextmanager
208     def _set_remote_context(self, servicer_context):
209         metadata = servicer_context.invocation_metadata()
210         if metadata:
211             md_dict = {md.key: md.value for md in metadata}
212             ctx = extract(md_dict)
213             token = attach(ctx)
214             try:
215                 yield
216             finally:
217                 detach(token)
218         else:
219             yield
220 
221     def _start_span(
222         self, handler_call_details, context, set_status_on_exception=False
223     ):
224         # standard attributes
225         attributes = {
226             SpanAttributes.RPC_SYSTEM: "grpc",
227             SpanAttributes.RPC_GRPC_STATUS_CODE: grpc.StatusCode.OK.value[0],
228         }
229 
230         # if we have details about the call, split into service and method
231         if handler_call_details.method:
232             service, method = handler_call_details.method.lstrip("/").split(
233                 "/", 1
234             )
235             attributes.update(
236                 {
237                     SpanAttributes.RPC_METHOD: method,
238                     SpanAttributes.RPC_SERVICE: service,
239                 }
240             )
241 
242         # add some attributes from the metadata
243         metadata = dict(context.invocation_metadata())
244         if "user-agent" in metadata:
245             attributes["rpc.user_agent"] = metadata["user-agent"]
246 
247         # Split up the peer to keep with how other telemetry sources
248         # do it.  This looks like:
249         # * ipv6:[::1]:57284
250         # * ipv4:127.0.0.1:57284
251         # * ipv4:10.2.1.1:57284,127.0.0.1:57284
252         #
253         try:
254             ip, port = (
255                 context.peer().split(",")[0].split(":", 1)[1].rsplit(":", 1)
256             )
257             ip = unquote(ip)
258             attributes.update(
259                 {
260                     SpanAttributes.NET_PEER_IP: ip,
261                     SpanAttributes.NET_PEER_PORT: port,
262                 }
263             )
264 
265             # other telemetry sources add this, so we will too
266             if ip in ("[::1]", "127.0.0.1"):
267                 attributes[SpanAttributes.NET_PEER_NAME] = "localhost"
268 
269         except IndexError:
270             logger.warning("Failed to parse peer address '%s'", context.peer())
271 
272         return self._tracer.start_as_current_span(
273             name=handler_call_details.method,
274             kind=trace.SpanKind.SERVER,
275             attributes=attributes,
276             set_status_on_exception=set_status_on_exception,
277         )
278 
279     def intercept_service(self, continuation, handler_call_details):
280         if self._filter is not None and not self._filter(handler_call_details):
281             return continuation(handler_call_details)
282 
283         def telemetry_wrapper(behavior, request_streaming, response_streaming):
284             def telemetry_interceptor(request_or_iterator, context):
285                 # handle streaming responses specially
286                 if response_streaming:
287                     return self._intercept_server_stream(
288                         behavior,
289                         handler_call_details,
290                         request_or_iterator,
291                         context,
292                     )
293 
294                 with self._set_remote_context(context):
295                     with self._start_span(
296                         handler_call_details,
297                         context,
298                         set_status_on_exception=False,
299                     ) as span:
300                         # wrap the context
301                         context = _OpenTelemetryServicerContext(context, span)
302 
303                         # And now we run the actual RPC.
304                         try:
305                             return behavior(request_or_iterator, context)
306 
307                         except Exception as error:
308                             # Bare exceptions are likely to be gRPC aborts, which
309                             # we handle in our context wrapper.
310                             # Here, we're interested in uncaught exceptions.
311                             # pylint:disable=unidiomatic-typecheck
312                             if type(error) != Exception:
313                                 span.record_exception(error)
314                             raise error
315 
316             return telemetry_interceptor
317 
318         return _wrap_rpc_behavior(
319             continuation(handler_call_details), telemetry_wrapper
320         )
321 
322     # Handle streaming responses separately - we have to do this
323     # to return a *new* generator or various upstream things
324     # get confused, or we'll lose the consistent trace
325     def _intercept_server_stream(
326         self, behavior, handler_call_details, request_or_iterator, context
327     ):
328         with self._set_remote_context(context):
329             with self._start_span(
330                 handler_call_details, context, set_status_on_exception=False
331             ) as span:
332                 context = _OpenTelemetryServicerContext(context, span)
333 
334                 try:
335                     yield from behavior(request_or_iterator, context)
336 
337                 except Exception as error:
338                     # pylint:disable=unidiomatic-typecheck
339                     if type(error) != Exception:
340                         span.record_exception(error)
341                     raise error
342
```
--- 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
@@ -250,24 +250,30 @@
         # * ipv4:127.0.0.1:57284
         # * ipv4:10.2.1.1:57284,127.0.0.1:57284
         #
-        try:
-            ip, port = (
-                context.peer().split(",")[0].split(":", 1)[1].rsplit(":", 1)
-            )
-            ip = unquote(ip)
-            attributes.update(
-                {
-                    SpanAttributes.NET_PEER_IP: ip,
-                    SpanAttributes.NET_PEER_PORT: port,
-                }
-            )
+        if context.peer() != "unix:":
+            try:
+                ip, port = (
+                    context.peer()
+                    .split(",")[0]
+                    .split(":", 1)[1]
+                    .rsplit(":", 1)
+                )
+                ip = unquote(ip)
+                attributes.update(
+                    {
+                        SpanAttributes.NET_PEER_IP: ip,
+                        SpanAttributes.NET_PEER_PORT: port,
+                    }
+                )
 
-            # other telemetry sources add this, so we will too
-            if ip in ("[::1]", "127.0.0.1"):
-                attributes[SpanAttributes.NET_PEER_NAME] = "localhost"
+                # other telemetry sources add this, so we will too
+                if ip in ("[::1]", "127.0.0.1"):
+                    attributes[SpanAttributes.NET_PEER_NAME] = "localhost"
 
-        except IndexError:
-            logger.warning("Failed to parse peer address '%s'", context.peer())
+            except IndexError:
+                logger.warning(
+                    "Failed to parse peer address '%s'", context.peer()
+                )
 
         return self._tracer.start_as_current_span(
             name=handler_call_details.method,

{"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@@ -250,24 +250,30 @@\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-            ip = unquote(ip)\n-            attributes.update(\n-                {\n-                    SpanAttributes.NET_PEER_IP: ip,\n-                    SpanAttributes.NET_PEER_PORT: port,\n-                }\n-            )\n+        if context.peer() != \"unix:\":\n+            try:\n+                ip, port = (\n+                    context.peer()\n+                    .split(\",\")[0]\n+                    .split(\":\", 1)[1]\n+                    .rsplit(\":\", 1)\n+                )\n+                ip = unquote(ip)\n+                attributes.update(\n+                    {\n+                        SpanAttributes.NET_PEER_IP: ip,\n+                        SpanAttributes.NET_PEER_PORT: port,\n+                    }\n+                )\n \n-            # other telemetry sources add this, so we will too\n-            if ip in (\"[::1]\", \"127.0.0.1\"):\n-                attributes[SpanAttributes.NET_PEER_NAME] = \"localhost\"\n+                # other telemetry sources add this, so we will too\n+                if ip in (\"[::1]\", \"127.0.0.1\"):\n+                    attributes[SpanAttributes.NET_PEER_NAME] = \"localhost\"\n \n-        except IndexError:\n-            logger.warning(\"Failed to parse peer address '%s'\", context.peer())\n+            except IndexError:\n+                logger.warning(\n+                    \"Failed to parse peer address '%s'\", context.peer()\n+                )\n \n         return self._tracer.start_as_current_span(\n             name=handler_call_details.method,\n", "issue": "gRPC server throws ValueError for Unix sockets\n**Describe your environment**\r\n\r\n```\r\nUbuntu 20.04\r\nx86_64\r\nPython 3.10.9\r\n```\r\n\r\n**Steps to reproduce**\r\n\r\nI'll post a PR that addresses the issue with a test that reproduces the problem shortly.\r\nThis problem can be triggered by running the `grpc` instrumentation and utilizing Unix sockets for connections.\r\n\r\n```py\r\nserver.add_insecure_port(\"unix:///tmp/grpc.sock\")\r\nchannel = grpc.insecure_channel(\"unix:///tmp/grpc.sock\")\r\n```\r\n\r\n**What is the expected behavior?**\r\n\r\nThe `grpc` instrumentation should collect traces when running on a Unix socket.\r\n\r\n**What is the actual behavior?**\r\n\r\nA `ValueError` is thrown here:\r\nhttps://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_server.py#L254-L256\r\n\r\n`context.peer()` is `unix:` which results in `ip, port = [\"\"]`.\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\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\nfrom urllib.parse import unquote\n\nimport grpc\n\nfrom opentelemetry import trace\nfrom opentelemetry.context import attach, detach\nfrom opentelemetry.propagate import extract\nfrom opentelemetry.semconv.trace import SpanAttributes\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 __getattr__(self, attr):\n        return getattr(self._servicer_context, attr)\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 trailing_metadata(self):\n        return self._servicer_context.trailing_metadata()\n\n    def abort(self, code, details):\n        self._code = code\n        self._details = details\n        self._active_span.set_attribute(\n            SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]\n        )\n        self._active_span.set_status(\n            Status(\n                status_code=StatusCode.ERROR,\n                description=f\"{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 code(self):\n        if not hasattr(self._servicer_context, \"code\"):\n            raise RuntimeError(\n                \"code() is not supported with the installed version of grpcio\"\n            )\n        return self._servicer_context.code()\n\n    def details(self):\n        if not hasattr(self._servicer_context, \"details\"):\n            raise RuntimeError(\n                \"details() is not supported with the installed version of \"\n                \"grpcio\"\n            )\n        return self._servicer_context.details()\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(\n            SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]\n        )\n        if code != grpc.StatusCode.OK:\n            self._active_span.set_status(\n                Status(\n                    status_code=StatusCode.ERROR,\n                    description=f\"{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=f\"{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        filter = filters.negate(filters.method_name(\"service.Foo\"))\n\n        interceptors = [\n            OpenTelemetryServerInterceptor(tracer, filter),\n        ]\n\n        server = grpc.server(\n            futures.ThreadPoolExecutor(max_workers=concurrency),\n            interceptors = interceptors)\n\n    \"\"\"\n\n    def __init__(self, tracer, filter_=None):\n        self._tracer = tracer\n        self._filter = filter_\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 = extract(md_dict)\n            token = attach(ctx)\n            try:\n                yield\n            finally:\n                detach(token)\n        else:\n            yield\n\n    def _start_span(\n        self, handler_call_details, context, set_status_on_exception=False\n    ):\n        # standard attributes\n        attributes = {\n            SpanAttributes.RPC_SYSTEM: \"grpc\",\n            SpanAttributes.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(\n                {\n                    SpanAttributes.RPC_METHOD: method,\n                    SpanAttributes.RPC_SERVICE: service,\n                }\n            )\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            ip = unquote(ip)\n            attributes.update(\n                {\n                    SpanAttributes.NET_PEER_IP: ip,\n                    SpanAttributes.NET_PEER_PORT: port,\n                }\n            )\n\n            # other telemetry sources add this, so we will too\n            if ip in (\"[::1]\", \"127.0.0.1\"):\n                attributes[SpanAttributes.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            set_status_on_exception=set_status_on_exception,\n        )\n\n    def intercept_service(self, continuation, handler_call_details):\n        if self._filter is not None and not self._filter(handler_call_details):\n            return continuation(handler_call_details)\n\n        def telemetry_wrapper(behavior, request_streaming, response_streaming):\n            def telemetry_interceptor(request_or_iterator, context):\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,\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 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        with self._set_remote_context(context):\n            with self._start_span(\n                handler_call_details, context, set_status_on_exception=False\n            ) 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"}], "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\nfrom urllib.parse import unquote\n\nimport grpc\n\nfrom opentelemetry import trace\nfrom opentelemetry.context import attach, detach\nfrom opentelemetry.propagate import extract\nfrom opentelemetry.semconv.trace import SpanAttributes\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 __getattr__(self, attr):\n        return getattr(self._servicer_context, attr)\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 trailing_metadata(self):\n        return self._servicer_context.trailing_metadata()\n\n    def abort(self, code, details):\n        self._code = code\n        self._details = details\n        self._active_span.set_attribute(\n            SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]\n        )\n        self._active_span.set_status(\n            Status(\n                status_code=StatusCode.ERROR,\n                description=f\"{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 code(self):\n        if not hasattr(self._servicer_context, \"code\"):\n            raise RuntimeError(\n                \"code() is not supported with the installed version of grpcio\"\n            )\n        return self._servicer_context.code()\n\n    def details(self):\n        if not hasattr(self._servicer_context, \"details\"):\n            raise RuntimeError(\n                \"details() is not supported with the installed version of \"\n                \"grpcio\"\n            )\n        return self._servicer_context.details()\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(\n            SpanAttributes.RPC_GRPC_STATUS_CODE, code.value[0]\n        )\n        if code != grpc.StatusCode.OK:\n            self._active_span.set_status(\n                Status(\n                    status_code=StatusCode.ERROR,\n                    description=f\"{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=f\"{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        filter = filters.negate(filters.method_name(\"service.Foo\"))\n\n        interceptors = [\n            OpenTelemetryServerInterceptor(tracer, filter),\n        ]\n\n        server = grpc.server(\n            futures.ThreadPoolExecutor(max_workers=concurrency),\n            interceptors = interceptors)\n\n    \"\"\"\n\n    def __init__(self, tracer, filter_=None):\n        self._tracer = tracer\n        self._filter = filter_\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 = extract(md_dict)\n            token = attach(ctx)\n            try:\n                yield\n            finally:\n                detach(token)\n        else:\n            yield\n\n    def _start_span(\n        self, handler_call_details, context, set_status_on_exception=False\n    ):\n        # standard attributes\n        attributes = {\n            SpanAttributes.RPC_SYSTEM: \"grpc\",\n            SpanAttributes.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(\n                {\n                    SpanAttributes.RPC_METHOD: method,\n                    SpanAttributes.RPC_SERVICE: service,\n                }\n            )\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        if context.peer() != \"unix:\":\n            try:\n                ip, port = (\n                    context.peer()\n                    .split(\",\")[0]\n                    .split(\":\", 1)[1]\n                    .rsplit(\":\", 1)\n                )\n                ip = unquote(ip)\n                attributes.update(\n                    {\n                        SpanAttributes.NET_PEER_IP: ip,\n                        SpanAttributes.NET_PEER_PORT: port,\n                    }\n                )\n\n                # other telemetry sources add this, so we will too\n                if ip in (\"[::1]\", \"127.0.0.1\"):\n                    attributes[SpanAttributes.NET_PEER_NAME] = \"localhost\"\n\n            except IndexError:\n                logger.warning(\n                    \"Failed to parse peer address '%s'\", context.peer()\n                )\n\n        return self._tracer.start_as_current_span(\n            name=handler_call_details.method,\n            kind=trace.SpanKind.SERVER,\n            attributes=attributes,\n            set_status_on_exception=set_status_on_exception,\n        )\n\n    def intercept_service(self, continuation, handler_call_details):\n        if self._filter is not None and not self._filter(handler_call_details):\n            return continuation(handler_call_details)\n\n        def telemetry_wrapper(behavior, request_streaming, response_streaming):\n            def telemetry_interceptor(request_or_iterator, context):\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,\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 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        with self._set_remote_context(context):\n            with self._start_span(\n                handler_call_details, context, set_status_on_exception=False\n            ) 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"}]}