We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Make sure rpc instrumentations follow semantic conventions
Specs: https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/rpc.md

--- 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.name)
117         self._active_span.set_status(
118             Status(status_code=StatusCode.ERROR, description=details)
119         )
120         return self._servicer_context.abort(code, details)
121 
122     def abort_with_status(self, status):
123         return self._servicer_context.abort_with_status(status)
124 
125     def set_code(self, code):
126         self.code = code
127         # use details if we already have it, otherwise the status description
128         details = self.details or code.value[1]
129         self._active_span.set_attribute("rpc.grpc.status_code", code.name)
130         self._active_span.set_status(
131             Status(status_code=StatusCode.ERROR, description=details)
132         )
133         return self._servicer_context.set_code(code)
134 
135     def set_details(self, details):
136         self.details = details
137         self._active_span.set_status(
138             Status(status_code=StatusCode.ERROR, description=details)
139         )
140         return self._servicer_context.set_details(details)
141 
142 
143 # pylint:disable=abstract-method
144 # pylint:disable=no-self-use
145 # pylint:disable=unused-argument
146 class OpenTelemetryServerInterceptor(grpc.ServerInterceptor):
147     """
148     A gRPC server interceptor, to add OpenTelemetry.
149 
150     Usage::
151 
152         tracer = some OpenTelemetry tracer
153 
154         interceptors = [
155             OpenTelemetryServerInterceptor(tracer),
156         ]
157 
158         server = grpc.server(
159             futures.ThreadPoolExecutor(max_workers=concurrency),
160             interceptors = interceptors)
161 
162     """
163 
164     def __init__(self, tracer):
165         self._tracer = tracer
166         self._carrier_getter = DictGetter()
167 
168     @contextmanager
169     def _set_remote_context(self, servicer_context):
170         metadata = servicer_context.invocation_metadata()
171         if metadata:
172             md_dict = {md.key: md.value for md in metadata}
173             ctx = propagators.extract(self._carrier_getter, md_dict)
174             token = attach(ctx)
175             try:
176                 yield
177             finally:
178                 detach(token)
179         else:
180             yield
181 
182     def _start_span(self, handler_call_details, context):
183 
184         attributes = {
185             "rpc.method": handler_call_details.method,
186             "rpc.system": "grpc",
187             "rpc.grpc.status_code": grpc.StatusCode.OK,
188         }
189 
190         metadata = dict(context.invocation_metadata())
191         if "user-agent" in metadata:
192             attributes["rpc.user_agent"] = metadata["user-agent"]
193 
194         # Split up the peer to keep with how other telemetry sources
195         # do it.  This looks like:
196         # * ipv6:[::1]:57284
197         # * ipv4:127.0.0.1:57284
198         # * ipv4:10.2.1.1:57284,127.0.0.1:57284
199         #
200         try:
201             host, port = (
202                 context.peer().split(",")[0].split(":", 1)[1].rsplit(":", 1)
203             )
204 
205             # other telemetry sources convert this, so we will too
206             if host in ("[::1]", "127.0.0.1"):
207                 host = "localhost"
208 
209             attributes.update({"net.peer.name": host, "net.peer.port": port})
210         except IndexError:
211             logger.warning("Failed to parse peer address '%s'", context.peer())
212 
213         return self._tracer.start_as_current_span(
214             name=handler_call_details.method,
215             kind=trace.SpanKind.SERVER,
216             attributes=attributes,
217         )
218 
219     def intercept_service(self, continuation, handler_call_details):
220         def telemetry_wrapper(behavior, request_streaming, response_streaming):
221             def telemetry_interceptor(request_or_iterator, context):
222 
223                 with self._set_remote_context(context):
224                     with self._start_span(
225                         handler_call_details, context
226                     ) as span:
227                         # wrap the context
228                         context = _OpenTelemetryServicerContext(context, span)
229 
230                         # And now we run the actual RPC.
231                         try:
232                             return behavior(request_or_iterator, context)
233                         except Exception as error:
234                             # Bare exceptions are likely to be gRPC aborts, which
235                             # we handle in our context wrapper.
236                             # Here, we're interested in uncaught exceptions.
237                             # pylint:disable=unidiomatic-typecheck
238                             if type(error) != Exception:
239                                 span.record_exception(error)
240                             raise error
241 
242             return telemetry_interceptor
243 
244         return _wrap_rpc_behavior(
245             continuation(handler_call_details), telemetry_wrapper
246         )
247
```
--- 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
@@ -113,9 +113,12 @@
     def abort(self, code, details):
         self.code = code
         self.details = details
-        self._active_span.set_attribute("rpc.grpc.status_code", code.name)
+        self._active_span.set_attribute("rpc.grpc.status_code", code.value[0])
         self._active_span.set_status(
-            Status(status_code=StatusCode.ERROR, description=details)
+            Status(
+                status_code=StatusCode.ERROR,
+                description="{}:{}".format(code, details),
+            )
         )
         return self._servicer_context.abort(code, details)
 
@@ -126,17 +129,25 @@
         self.code = code
         # use details if we already have it, otherwise the status description
         details = self.details or code.value[1]
-        self._active_span.set_attribute("rpc.grpc.status_code", code.name)
-        self._active_span.set_status(
-            Status(status_code=StatusCode.ERROR, description=details)
-        )
+        self._active_span.set_attribute("rpc.grpc.status_code", code.value[0])
+        if code != grpc.StatusCode.OK:
+            self._active_span.set_status(
+                Status(
+                    status_code=StatusCode.ERROR,
+                    description="{}:{}".format(code, details),
+                )
+            )
         return self._servicer_context.set_code(code)
 
     def set_details(self, details):
         self.details = details
-        self._active_span.set_status(
-            Status(status_code=StatusCode.ERROR, description=details)
-        )
+        if self.code != grpc.StatusCode.OK:
+            self._active_span.set_status(
+                Status(
+                    status_code=StatusCode.ERROR,
+                    description="{}:{}".format(self.code, details),
+                )
+            )
         return self._servicer_context.set_details(details)
 
 
@@ -181,12 +192,20 @@
 
     def _start_span(self, handler_call_details, context):
 
+        # standard attributes
         attributes = {
-            "rpc.method": handler_call_details.method,
             "rpc.system": "grpc",
-            "rpc.grpc.status_code": grpc.StatusCode.OK,
+            "rpc.grpc.status_code": grpc.StatusCode.OK.value[0],
         }
 
+        # if we have details about the call, split into service and method
+        if handler_call_details.method:
+            service, method = handler_call_details.method.lstrip("/").split(
+                "/", 1
+            )
+            attributes.update({"rpc.method": method, "rpc.service": service})
+
+        # add some attributes from the metadata
         metadata = dict(context.invocation_metadata())
         if "user-agent" in metadata:
             attributes["rpc.user_agent"] = metadata["user-agent"]
@@ -198,15 +217,15 @@
         # * ipv4:10.2.1.1:57284,127.0.0.1:57284
         #
         try:
-            host, port = (
+            ip, port = (
                 context.peer().split(",")[0].split(":", 1)[1].rsplit(":", 1)
             )
+            attributes.update({"net.peer.ip": ip, "net.peer.port": port})
 
-            # other telemetry sources convert this, so we will too
-            if host in ("[::1]", "127.0.0.1"):
-                host = "localhost"
+            # other telemetry sources add this, so we will too
+            if ip in ("[::1]", "127.0.0.1"):
+                attributes["net.peer.name"] = "localhost"
 
-            attributes.update({"net.peer.name": host, "net.peer.port": port})
         except IndexError:
             logger.warning("Failed to parse peer address '%s'", context.peer())

{"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@@ -113,9 +113,12 @@\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.name)\n+        self._active_span.set_attribute(\"rpc.grpc.status_code\", code.value[0])\n         self._active_span.set_status(\n-            Status(status_code=StatusCode.ERROR, description=details)\n+            Status(\n+                status_code=StatusCode.ERROR,\n+                description=\"{}:{}\".format(code, details),\n+            )\n         )\n         return self._servicer_context.abort(code, details)\n \n@@ -126,17 +129,25 @@\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.name)\n-        self._active_span.set_status(\n-            Status(status_code=StatusCode.ERROR, description=details)\n-        )\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-        self._active_span.set_status(\n-            Status(status_code=StatusCode.ERROR, description=details)\n-        )\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@@ -181,12 +192,20 @@\n \n     def _start_span(self, handler_call_details, context):\n \n+        # standard attributes\n         attributes = {\n-            \"rpc.method\": handler_call_details.method,\n             \"rpc.system\": \"grpc\",\n-            \"rpc.grpc.status_code\": grpc.StatusCode.OK,\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@@ -198,15 +217,15 @@\n         # * ipv4:10.2.1.1:57284,127.0.0.1:57284\n         #\n         try:\n-            host, port = (\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 convert this, so we will too\n-            if host in (\"[::1]\", \"127.0.0.1\"):\n-                host = \"localhost\"\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-            attributes.update({\"net.peer.name\": host, \"net.peer.port\": port})\n         except IndexError:\n             logger.warning(\"Failed to parse peer address '%s'\", context.peer())\n", "issue": "Make sure rpc instrumentations follow semantic conventions\nSpecs: https://github.com/open-telemetry/opentelemetry-specification/blob/master/specification/trace/semantic_conventions/rpc.md\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.name)\n        self._active_span.set_status(\n            Status(status_code=StatusCode.ERROR, description=details)\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.name)\n        self._active_span.set_status(\n            Status(status_code=StatusCode.ERROR, description=details)\n        )\n        return self._servicer_context.set_code(code)\n\n    def set_details(self, details):\n        self.details = details\n        self._active_span.set_status(\n            Status(status_code=StatusCode.ERROR, description=details)\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        attributes = {\n            \"rpc.method\": handler_call_details.method,\n            \"rpc.system\": \"grpc\",\n            \"rpc.grpc.status_code\": grpc.StatusCode.OK,\n        }\n\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            host, port = (\n                context.peer().split(\",\")[0].split(\":\", 1)[1].rsplit(\":\", 1)\n            )\n\n            # other telemetry sources convert this, so we will too\n            if host in (\"[::1]\", \"127.0.0.1\"):\n                host = \"localhost\"\n\n            attributes.update({\"net.peer.name\": host, \"net.peer.port\": port})\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                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"}]}