We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[Redis] RedisCluster support
Redis instrumentation does not support RedisCluster ([redis.asyncio.cluster.RedisCluster](https://github.com/redis/redis-py/blob/master/redis/asyncio/cluster.py#L102)).

I tried to instrument RedisCluster using `opentelemetry-instrumentation-redis` as below, but the application did not export any traces related redis.
```python
from opentelemetry.instrumentation.redis import RedisInstrumentor
from redis.asyncio.cluster import RedisCluster as Redis

RedisInstrumentor().instrument()

client = await Redis(host=host, port=6379)

client.get("my-key")
```

It would be very helpful if redis instrumentation supports RedisCluster.


--- 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-redis/src/opentelemetry/instrumentation/redis/__init__.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 """
16 Instrument `redis`_ to report Redis queries.
17 
18 There are two options for instrumenting code. The first option is to use the
19 ``opentelemetry-instrumentation`` executable which will automatically
20 instrument your Redis client. The second is to programmatically enable
21 instrumentation via the following code:
22 
23 .. _redis: https://pypi.org/project/redis/
24 
25 Usage
26 -----
27 
28 .. code:: python
29 
30     from opentelemetry.instrumentation.redis import RedisInstrumentor
31     import redis
32 
33 
34     # Instrument redis
35     RedisInstrumentor().instrument()
36 
37     # This will report a span with the default settings
38     client = redis.StrictRedis(host="localhost", port=6379)
39     client.get("my-key")
40 
41 Async Redis clients (i.e. redis.asyncio.Redis) are also instrumented in the same way:
42 
43 .. code:: python
44 
45     from opentelemetry.instrumentation.redis import RedisInstrumentor
46     import redis.asyncio
47 
48 
49     # Instrument redis
50     RedisInstrumentor().instrument()
51 
52     # This will report a span with the default settings
53     async def redis_get():
54         client = redis.asyncio.Redis(host="localhost", port=6379)
55         await client.get("my-key")
56 
57 The `instrument` method accepts the following keyword args:
58 
59 tracer_provider (TracerProvider) - an optional tracer provider
60 
61 request_hook (Callable) - a function with extra user-defined logic to be performed before performing the request
62 this function signature is:  def request_hook(span: Span, instance: redis.connection.Connection, args, kwargs) -> None
63 
64 response_hook (Callable) - a function with extra user-defined logic to be performed after performing the request
65 this function signature is: def response_hook(span: Span, instance: redis.connection.Connection, response) -> None
66 
67 for example:
68 
69 .. code: python
70 
71     from opentelemetry.instrumentation.redis import RedisInstrumentor
72     import redis
73 
74     def request_hook(span, instance, args, kwargs):
75         if span and span.is_recording():
76             span.set_attribute("custom_user_attribute_from_request_hook", "some-value")
77 
78     def response_hook(span, instance, response):
79         if span and span.is_recording():
80             span.set_attribute("custom_user_attribute_from_response_hook", "some-value")
81 
82     # Instrument redis with hooks
83     RedisInstrumentor().instrument(request_hook=request_hook, response_hook=response_hook)
84 
85     # This will report a span with the default settings and the custom attributes added from the hooks
86     client = redis.StrictRedis(host="localhost", port=6379)
87     client.get("my-key")
88 
89 API
90 ---
91 """
92 import typing
93 from typing import Any, Collection
94 
95 import redis
96 from wrapt import wrap_function_wrapper
97 
98 from opentelemetry import trace
99 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
100 from opentelemetry.instrumentation.redis.package import _instruments
101 from opentelemetry.instrumentation.redis.util import (
102     _extract_conn_attributes,
103     _format_command_args,
104 )
105 from opentelemetry.instrumentation.redis.version import __version__
106 from opentelemetry.instrumentation.utils import unwrap
107 from opentelemetry.semconv.trace import SpanAttributes
108 from opentelemetry.trace import Span
109 
110 _DEFAULT_SERVICE = "redis"
111 
112 _RequestHookT = typing.Optional[
113     typing.Callable[
114         [Span, redis.connection.Connection, typing.List, typing.Dict], None
115     ]
116 ]
117 _ResponseHookT = typing.Optional[
118     typing.Callable[[Span, redis.connection.Connection, Any], None]
119 ]
120 
121 _REDIS_ASYNCIO_VERSION = (4, 2, 0)
122 if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
123     import redis.asyncio
124 
125 
126 def _set_connection_attributes(span, conn):
127     if not span.is_recording():
128         return
129     for key, value in _extract_conn_attributes(
130         conn.connection_pool.connection_kwargs
131     ).items():
132         span.set_attribute(key, value)
133 
134 
135 def _instrument(
136     tracer,
137     request_hook: _RequestHookT = None,
138     response_hook: _ResponseHookT = None,
139 ):
140     def _traced_execute_command(func, instance, args, kwargs):
141         query = _format_command_args(args)
142         name = ""
143         if len(args) > 0 and args[0]:
144             name = args[0]
145         else:
146             name = instance.connection_pool.connection_kwargs.get("db", 0)
147         with tracer.start_as_current_span(
148             name, kind=trace.SpanKind.CLIENT
149         ) as span:
150             if span.is_recording():
151                 span.set_attribute(SpanAttributes.DB_STATEMENT, query)
152                 _set_connection_attributes(span, instance)
153                 span.set_attribute("db.redis.args_length", len(args))
154             if callable(request_hook):
155                 request_hook(span, instance, args, kwargs)
156             response = func(*args, **kwargs)
157             if callable(response_hook):
158                 response_hook(span, instance, response)
159             return response
160 
161     def _traced_execute_pipeline(func, instance, args, kwargs):
162         cmds = [_format_command_args(c) for c, _ in instance.command_stack]
163         resource = "\n".join(cmds)
164 
165         span_name = " ".join([args[0] for args, _ in instance.command_stack])
166 
167         with tracer.start_as_current_span(
168             span_name, kind=trace.SpanKind.CLIENT
169         ) as span:
170             if span.is_recording():
171                 span.set_attribute(SpanAttributes.DB_STATEMENT, resource)
172                 _set_connection_attributes(span, instance)
173                 span.set_attribute(
174                     "db.redis.pipeline_length", len(instance.command_stack)
175                 )
176             response = func(*args, **kwargs)
177             if callable(response_hook):
178                 response_hook(span, instance, response)
179             return response
180 
181     pipeline_class = (
182         "BasePipeline" if redis.VERSION < (3, 0, 0) else "Pipeline"
183     )
184     redis_class = "StrictRedis" if redis.VERSION < (3, 0, 0) else "Redis"
185 
186     wrap_function_wrapper(
187         "redis", f"{redis_class}.execute_command", _traced_execute_command
188     )
189     wrap_function_wrapper(
190         "redis.client",
191         f"{pipeline_class}.execute",
192         _traced_execute_pipeline,
193     )
194     wrap_function_wrapper(
195         "redis.client",
196         f"{pipeline_class}.immediate_execute_command",
197         _traced_execute_command,
198     )
199     if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
200         wrap_function_wrapper(
201             "redis.asyncio",
202             f"{redis_class}.execute_command",
203             _traced_execute_command,
204         )
205         wrap_function_wrapper(
206             "redis.asyncio.client",
207             f"{pipeline_class}.execute",
208             _traced_execute_pipeline,
209         )
210         wrap_function_wrapper(
211             "redis.asyncio.client",
212             f"{pipeline_class}.immediate_execute_command",
213             _traced_execute_command,
214         )
215 
216 
217 class RedisInstrumentor(BaseInstrumentor):
218     """An instrumentor for Redis
219     See `BaseInstrumentor`
220     """
221 
222     def instrumentation_dependencies(self) -> Collection[str]:
223         return _instruments
224 
225     def _instrument(self, **kwargs):
226         """Instruments the redis module
227 
228         Args:
229             **kwargs: Optional arguments
230                 ``tracer_provider``: a TracerProvider, defaults to global.
231                 ``response_hook``: An optional callback which is invoked right before the span is finished processing a response.
232         """
233         tracer_provider = kwargs.get("tracer_provider")
234         tracer = trace.get_tracer(
235             __name__, __version__, tracer_provider=tracer_provider
236         )
237         _instrument(
238             tracer,
239             request_hook=kwargs.get("request_hook"),
240             response_hook=kwargs.get("response_hook"),
241         )
242 
243     def _uninstrument(self, **kwargs):
244         if redis.VERSION < (3, 0, 0):
245             unwrap(redis.StrictRedis, "execute_command")
246             unwrap(redis.StrictRedis, "pipeline")
247             unwrap(redis.Redis, "pipeline")
248             unwrap(
249                 redis.client.BasePipeline,  # pylint:disable=no-member
250                 "execute",
251             )
252             unwrap(
253                 redis.client.BasePipeline,  # pylint:disable=no-member
254                 "immediate_execute_command",
255             )
256         else:
257             unwrap(redis.Redis, "execute_command")
258             unwrap(redis.Redis, "pipeline")
259             unwrap(redis.client.Pipeline, "execute")
260             unwrap(redis.client.Pipeline, "immediate_execute_command")
261         if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
262             unwrap(redis.asyncio.Redis, "execute_command")
263             unwrap(redis.asyncio.Redis, "pipeline")
264             unwrap(redis.asyncio.client.Pipeline, "execute")
265             unwrap(redis.asyncio.client.Pipeline, "immediate_execute_command")
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-redis/src/opentelemetry/instrumentation/redis/__init__.py b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py
--- a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py
@@ -122,9 +122,12 @@
 if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
     import redis.asyncio
 
+_REDIS_CLUSTER_VERSION = (4, 1, 0)
+_REDIS_ASYNCIO_CLUSTER_VERSION = (4, 3, 0)
+
 
 def _set_connection_attributes(span, conn):
-    if not span.is_recording():
+    if not span.is_recording() or not hasattr(conn, "connection_pool"):
         return
     for key, value in _extract_conn_attributes(
         conn.connection_pool.connection_kwargs
@@ -159,10 +162,29 @@
             return response
 
     def _traced_execute_pipeline(func, instance, args, kwargs):
-        cmds = [_format_command_args(c) for c, _ in instance.command_stack]
-        resource = "\n".join(cmds)
+        try:
+            command_stack = (
+                instance.command_stack
+                if hasattr(instance, "command_stack")
+                else instance._command_stack
+            )
 
-        span_name = " ".join([args[0] for args, _ in instance.command_stack])
+            cmds = [
+                _format_command_args(c.args if hasattr(c, "args") else c[0])
+                for c in command_stack
+            ]
+            resource = "\n".join(cmds)
+
+            span_name = " ".join(
+                [
+                    (c.args[0] if hasattr(c, "args") else c[0][0])
+                    for c in command_stack
+                ]
+            )
+        except (AttributeError, IndexError):
+            command_stack = []
+            resource = ""
+            span_name = ""
 
         with tracer.start_as_current_span(
             span_name, kind=trace.SpanKind.CLIENT
@@ -171,7 +193,7 @@
                 span.set_attribute(SpanAttributes.DB_STATEMENT, resource)
                 _set_connection_attributes(span, instance)
                 span.set_attribute(
-                    "db.redis.pipeline_length", len(instance.command_stack)
+                    "db.redis.pipeline_length", len(command_stack)
                 )
             response = func(*args, **kwargs)
             if callable(response_hook):
@@ -196,6 +218,17 @@
         f"{pipeline_class}.immediate_execute_command",
         _traced_execute_command,
     )
+    if redis.VERSION >= _REDIS_CLUSTER_VERSION:
+        wrap_function_wrapper(
+            "redis.cluster",
+            "RedisCluster.execute_command",
+            _traced_execute_command,
+        )
+        wrap_function_wrapper(
+            "redis.cluster",
+            "ClusterPipeline.execute",
+            _traced_execute_pipeline,
+        )
     if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
         wrap_function_wrapper(
             "redis.asyncio",
@@ -212,6 +245,17 @@
             f"{pipeline_class}.immediate_execute_command",
             _traced_execute_command,
         )
+    if redis.VERSION >= _REDIS_ASYNCIO_CLUSTER_VERSION:
+        wrap_function_wrapper(
+            "redis.asyncio.cluster",
+            "RedisCluster.execute_command",
+            _traced_execute_command,
+        )
+        wrap_function_wrapper(
+            "redis.asyncio.cluster",
+            "ClusterPipeline.execute",
+            _traced_execute_pipeline,
+        )
 
 
 class RedisInstrumentor(BaseInstrumentor):
@@ -258,8 +302,14 @@
             unwrap(redis.Redis, "pipeline")
             unwrap(redis.client.Pipeline, "execute")
             unwrap(redis.client.Pipeline, "immediate_execute_command")
+        if redis.VERSION >= _REDIS_CLUSTER_VERSION:
+            unwrap(redis.cluster.RedisCluster, "execute_command")
+            unwrap(redis.cluster.ClusterPipeline, "execute")
         if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
             unwrap(redis.asyncio.Redis, "execute_command")
             unwrap(redis.asyncio.Redis, "pipeline")
             unwrap(redis.asyncio.client.Pipeline, "execute")
             unwrap(redis.asyncio.client.Pipeline, "immediate_execute_command")
+        if redis.VERSION >= _REDIS_ASYNCIO_CLUSTER_VERSION:
+            unwrap(redis.asyncio.cluster.RedisCluster, "execute_command")
+            unwrap(redis.asyncio.cluster.ClusterPipeline, "execute")

{"golden_diff": "diff --git a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py\n--- a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py\n+++ b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py\n@@ -122,9 +122,12 @@\n if redis.VERSION >= _REDIS_ASYNCIO_VERSION:\n     import redis.asyncio\n \n+_REDIS_CLUSTER_VERSION = (4, 1, 0)\n+_REDIS_ASYNCIO_CLUSTER_VERSION = (4, 3, 0)\n+\n \n def _set_connection_attributes(span, conn):\n-    if not span.is_recording():\n+    if not span.is_recording() or not hasattr(conn, \"connection_pool\"):\n         return\n     for key, value in _extract_conn_attributes(\n         conn.connection_pool.connection_kwargs\n@@ -159,10 +162,29 @@\n             return response\n \n     def _traced_execute_pipeline(func, instance, args, kwargs):\n-        cmds = [_format_command_args(c) for c, _ in instance.command_stack]\n-        resource = \"\\n\".join(cmds)\n+        try:\n+            command_stack = (\n+                instance.command_stack\n+                if hasattr(instance, \"command_stack\")\n+                else instance._command_stack\n+            )\n \n-        span_name = \" \".join([args[0] for args, _ in instance.command_stack])\n+            cmds = [\n+                _format_command_args(c.args if hasattr(c, \"args\") else c[0])\n+                for c in command_stack\n+            ]\n+            resource = \"\\n\".join(cmds)\n+\n+            span_name = \" \".join(\n+                [\n+                    (c.args[0] if hasattr(c, \"args\") else c[0][0])\n+                    for c in command_stack\n+                ]\n+            )\n+        except (AttributeError, IndexError):\n+            command_stack = []\n+            resource = \"\"\n+            span_name = \"\"\n \n         with tracer.start_as_current_span(\n             span_name, kind=trace.SpanKind.CLIENT\n@@ -171,7 +193,7 @@\n                 span.set_attribute(SpanAttributes.DB_STATEMENT, resource)\n                 _set_connection_attributes(span, instance)\n                 span.set_attribute(\n-                    \"db.redis.pipeline_length\", len(instance.command_stack)\n+                    \"db.redis.pipeline_length\", len(command_stack)\n                 )\n             response = func(*args, **kwargs)\n             if callable(response_hook):\n@@ -196,6 +218,17 @@\n         f\"{pipeline_class}.immediate_execute_command\",\n         _traced_execute_command,\n     )\n+    if redis.VERSION >= _REDIS_CLUSTER_VERSION:\n+        wrap_function_wrapper(\n+            \"redis.cluster\",\n+            \"RedisCluster.execute_command\",\n+            _traced_execute_command,\n+        )\n+        wrap_function_wrapper(\n+            \"redis.cluster\",\n+            \"ClusterPipeline.execute\",\n+            _traced_execute_pipeline,\n+        )\n     if redis.VERSION >= _REDIS_ASYNCIO_VERSION:\n         wrap_function_wrapper(\n             \"redis.asyncio\",\n@@ -212,6 +245,17 @@\n             f\"{pipeline_class}.immediate_execute_command\",\n             _traced_execute_command,\n         )\n+    if redis.VERSION >= _REDIS_ASYNCIO_CLUSTER_VERSION:\n+        wrap_function_wrapper(\n+            \"redis.asyncio.cluster\",\n+            \"RedisCluster.execute_command\",\n+            _traced_execute_command,\n+        )\n+        wrap_function_wrapper(\n+            \"redis.asyncio.cluster\",\n+            \"ClusterPipeline.execute\",\n+            _traced_execute_pipeline,\n+        )\n \n \n class RedisInstrumentor(BaseInstrumentor):\n@@ -258,8 +302,14 @@\n             unwrap(redis.Redis, \"pipeline\")\n             unwrap(redis.client.Pipeline, \"execute\")\n             unwrap(redis.client.Pipeline, \"immediate_execute_command\")\n+        if redis.VERSION >= _REDIS_CLUSTER_VERSION:\n+            unwrap(redis.cluster.RedisCluster, \"execute_command\")\n+            unwrap(redis.cluster.ClusterPipeline, \"execute\")\n         if redis.VERSION >= _REDIS_ASYNCIO_VERSION:\n             unwrap(redis.asyncio.Redis, \"execute_command\")\n             unwrap(redis.asyncio.Redis, \"pipeline\")\n             unwrap(redis.asyncio.client.Pipeline, \"execute\")\n             unwrap(redis.asyncio.client.Pipeline, \"immediate_execute_command\")\n+        if redis.VERSION >= _REDIS_ASYNCIO_CLUSTER_VERSION:\n+            unwrap(redis.asyncio.cluster.RedisCluster, \"execute_command\")\n+            unwrap(redis.asyncio.cluster.ClusterPipeline, \"execute\")\n", "issue": "[Redis] RedisCluster support\nRedis instrumentation does not support RedisCluster ([redis.asyncio.cluster.RedisCluster](https://github.com/redis/redis-py/blob/master/redis/asyncio/cluster.py#L102)).\r\n\r\nI tried to instrument RedisCluster using `opentelemetry-instrumentation-redis` as below, but the application did not export any traces related redis.\r\n```python\r\nfrom opentelemetry.instrumentation.redis import RedisInstrumentor\r\nfrom redis.asyncio.cluster import RedisCluster as Redis\r\n\r\nRedisInstrumentor().instrument()\r\n\r\nclient = await Redis(host=host, port=6379)\r\n\r\nclient.get(\"my-key\")\r\n```\r\n\r\nIt would be very helpful if redis instrumentation supports RedisCluster.\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\"\"\"\nInstrument `redis`_ to report Redis queries.\n\nThere are two options for instrumenting code. The first option is to use the\n``opentelemetry-instrumentation`` executable which will automatically\ninstrument your Redis client. The second is to programmatically enable\ninstrumentation via the following code:\n\n.. _redis: https://pypi.org/project/redis/\n\nUsage\n-----\n\n.. code:: python\n\n    from opentelemetry.instrumentation.redis import RedisInstrumentor\n    import redis\n\n\n    # Instrument redis\n    RedisInstrumentor().instrument()\n\n    # This will report a span with the default settings\n    client = redis.StrictRedis(host=\"localhost\", port=6379)\n    client.get(\"my-key\")\n\nAsync Redis clients (i.e. redis.asyncio.Redis) are also instrumented in the same way:\n\n.. code:: python\n\n    from opentelemetry.instrumentation.redis import RedisInstrumentor\n    import redis.asyncio\n\n\n    # Instrument redis\n    RedisInstrumentor().instrument()\n\n    # This will report a span with the default settings\n    async def redis_get():\n        client = redis.asyncio.Redis(host=\"localhost\", port=6379)\n        await client.get(\"my-key\")\n\nThe `instrument` method accepts the following keyword args:\n\ntracer_provider (TracerProvider) - an optional tracer provider\n\nrequest_hook (Callable) - a function with extra user-defined logic to be performed before performing the request\nthis function signature is:  def request_hook(span: Span, instance: redis.connection.Connection, args, kwargs) -> None\n\nresponse_hook (Callable) - a function with extra user-defined logic to be performed after performing the request\nthis function signature is: def response_hook(span: Span, instance: redis.connection.Connection, response) -> None\n\nfor example:\n\n.. code: python\n\n    from opentelemetry.instrumentation.redis import RedisInstrumentor\n    import redis\n\n    def request_hook(span, instance, args, kwargs):\n        if span and span.is_recording():\n            span.set_attribute(\"custom_user_attribute_from_request_hook\", \"some-value\")\n\n    def response_hook(span, instance, response):\n        if span and span.is_recording():\n            span.set_attribute(\"custom_user_attribute_from_response_hook\", \"some-value\")\n\n    # Instrument redis with hooks\n    RedisInstrumentor().instrument(request_hook=request_hook, response_hook=response_hook)\n\n    # This will report a span with the default settings and the custom attributes added from the hooks\n    client = redis.StrictRedis(host=\"localhost\", port=6379)\n    client.get(\"my-key\")\n\nAPI\n---\n\"\"\"\nimport typing\nfrom typing import Any, Collection\n\nimport redis\nfrom wrapt import wrap_function_wrapper\n\nfrom opentelemetry import trace\nfrom opentelemetry.instrumentation.instrumentor import BaseInstrumentor\nfrom opentelemetry.instrumentation.redis.package import _instruments\nfrom opentelemetry.instrumentation.redis.util import (\n    _extract_conn_attributes,\n    _format_command_args,\n)\nfrom opentelemetry.instrumentation.redis.version import __version__\nfrom opentelemetry.instrumentation.utils import unwrap\nfrom opentelemetry.semconv.trace import SpanAttributes\nfrom opentelemetry.trace import Span\n\n_DEFAULT_SERVICE = \"redis\"\n\n_RequestHookT = typing.Optional[\n    typing.Callable[\n        [Span, redis.connection.Connection, typing.List, typing.Dict], None\n    ]\n]\n_ResponseHookT = typing.Optional[\n    typing.Callable[[Span, redis.connection.Connection, Any], None]\n]\n\n_REDIS_ASYNCIO_VERSION = (4, 2, 0)\nif redis.VERSION >= _REDIS_ASYNCIO_VERSION:\n    import redis.asyncio\n\n\ndef _set_connection_attributes(span, conn):\n    if not span.is_recording():\n        return\n    for key, value in _extract_conn_attributes(\n        conn.connection_pool.connection_kwargs\n    ).items():\n        span.set_attribute(key, value)\n\n\ndef _instrument(\n    tracer,\n    request_hook: _RequestHookT = None,\n    response_hook: _ResponseHookT = None,\n):\n    def _traced_execute_command(func, instance, args, kwargs):\n        query = _format_command_args(args)\n        name = \"\"\n        if len(args) > 0 and args[0]:\n            name = args[0]\n        else:\n            name = instance.connection_pool.connection_kwargs.get(\"db\", 0)\n        with tracer.start_as_current_span(\n            name, kind=trace.SpanKind.CLIENT\n        ) as span:\n            if span.is_recording():\n                span.set_attribute(SpanAttributes.DB_STATEMENT, query)\n                _set_connection_attributes(span, instance)\n                span.set_attribute(\"db.redis.args_length\", len(args))\n            if callable(request_hook):\n                request_hook(span, instance, args, kwargs)\n            response = func(*args, **kwargs)\n            if callable(response_hook):\n                response_hook(span, instance, response)\n            return response\n\n    def _traced_execute_pipeline(func, instance, args, kwargs):\n        cmds = [_format_command_args(c) for c, _ in instance.command_stack]\n        resource = \"\\n\".join(cmds)\n\n        span_name = \" \".join([args[0] for args, _ in instance.command_stack])\n\n        with tracer.start_as_current_span(\n            span_name, kind=trace.SpanKind.CLIENT\n        ) as span:\n            if span.is_recording():\n                span.set_attribute(SpanAttributes.DB_STATEMENT, resource)\n                _set_connection_attributes(span, instance)\n                span.set_attribute(\n                    \"db.redis.pipeline_length\", len(instance.command_stack)\n                )\n            response = func(*args, **kwargs)\n            if callable(response_hook):\n                response_hook(span, instance, response)\n            return response\n\n    pipeline_class = (\n        \"BasePipeline\" if redis.VERSION < (3, 0, 0) else \"Pipeline\"\n    )\n    redis_class = \"StrictRedis\" if redis.VERSION < (3, 0, 0) else \"Redis\"\n\n    wrap_function_wrapper(\n        \"redis\", f\"{redis_class}.execute_command\", _traced_execute_command\n    )\n    wrap_function_wrapper(\n        \"redis.client\",\n        f\"{pipeline_class}.execute\",\n        _traced_execute_pipeline,\n    )\n    wrap_function_wrapper(\n        \"redis.client\",\n        f\"{pipeline_class}.immediate_execute_command\",\n        _traced_execute_command,\n    )\n    if redis.VERSION >= _REDIS_ASYNCIO_VERSION:\n        wrap_function_wrapper(\n            \"redis.asyncio\",\n            f\"{redis_class}.execute_command\",\n            _traced_execute_command,\n        )\n        wrap_function_wrapper(\n            \"redis.asyncio.client\",\n            f\"{pipeline_class}.execute\",\n            _traced_execute_pipeline,\n        )\n        wrap_function_wrapper(\n            \"redis.asyncio.client\",\n            f\"{pipeline_class}.immediate_execute_command\",\n            _traced_execute_command,\n        )\n\n\nclass RedisInstrumentor(BaseInstrumentor):\n    \"\"\"An instrumentor for Redis\n    See `BaseInstrumentor`\n    \"\"\"\n\n    def instrumentation_dependencies(self) -> Collection[str]:\n        return _instruments\n\n    def _instrument(self, **kwargs):\n        \"\"\"Instruments the redis module\n\n        Args:\n            **kwargs: Optional arguments\n                ``tracer_provider``: a TracerProvider, defaults to global.\n                ``response_hook``: An optional callback which is invoked right before the span is finished processing a response.\n        \"\"\"\n        tracer_provider = kwargs.get(\"tracer_provider\")\n        tracer = trace.get_tracer(\n            __name__, __version__, tracer_provider=tracer_provider\n        )\n        _instrument(\n            tracer,\n            request_hook=kwargs.get(\"request_hook\"),\n            response_hook=kwargs.get(\"response_hook\"),\n        )\n\n    def _uninstrument(self, **kwargs):\n        if redis.VERSION < (3, 0, 0):\n            unwrap(redis.StrictRedis, \"execute_command\")\n            unwrap(redis.StrictRedis, \"pipeline\")\n            unwrap(redis.Redis, \"pipeline\")\n            unwrap(\n                redis.client.BasePipeline,  # pylint:disable=no-member\n                \"execute\",\n            )\n            unwrap(\n                redis.client.BasePipeline,  # pylint:disable=no-member\n                \"immediate_execute_command\",\n            )\n        else:\n            unwrap(redis.Redis, \"execute_command\")\n            unwrap(redis.Redis, \"pipeline\")\n            unwrap(redis.client.Pipeline, \"execute\")\n            unwrap(redis.client.Pipeline, \"immediate_execute_command\")\n        if redis.VERSION >= _REDIS_ASYNCIO_VERSION:\n            unwrap(redis.asyncio.Redis, \"execute_command\")\n            unwrap(redis.asyncio.Redis, \"pipeline\")\n            unwrap(redis.asyncio.client.Pipeline, \"execute\")\n            unwrap(redis.asyncio.client.Pipeline, \"immediate_execute_command\")\n", "path": "instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.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\"\"\"\nInstrument `redis`_ to report Redis queries.\n\nThere are two options for instrumenting code. The first option is to use the\n``opentelemetry-instrumentation`` executable which will automatically\ninstrument your Redis client. The second is to programmatically enable\ninstrumentation via the following code:\n\n.. _redis: https://pypi.org/project/redis/\n\nUsage\n-----\n\n.. code:: python\n\n    from opentelemetry.instrumentation.redis import RedisInstrumentor\n    import redis\n\n\n    # Instrument redis\n    RedisInstrumentor().instrument()\n\n    # This will report a span with the default settings\n    client = redis.StrictRedis(host=\"localhost\", port=6379)\n    client.get(\"my-key\")\n\nAsync Redis clients (i.e. redis.asyncio.Redis) are also instrumented in the same way:\n\n.. code:: python\n\n    from opentelemetry.instrumentation.redis import RedisInstrumentor\n    import redis.asyncio\n\n\n    # Instrument redis\n    RedisInstrumentor().instrument()\n\n    # This will report a span with the default settings\n    async def redis_get():\n        client = redis.asyncio.Redis(host=\"localhost\", port=6379)\n        await client.get(\"my-key\")\n\nThe `instrument` method accepts the following keyword args:\n\ntracer_provider (TracerProvider) - an optional tracer provider\n\nrequest_hook (Callable) - a function with extra user-defined logic to be performed before performing the request\nthis function signature is:  def request_hook(span: Span, instance: redis.connection.Connection, args, kwargs) -> None\n\nresponse_hook (Callable) - a function with extra user-defined logic to be performed after performing the request\nthis function signature is: def response_hook(span: Span, instance: redis.connection.Connection, response) -> None\n\nfor example:\n\n.. code: python\n\n    from opentelemetry.instrumentation.redis import RedisInstrumentor\n    import redis\n\n    def request_hook(span, instance, args, kwargs):\n        if span and span.is_recording():\n            span.set_attribute(\"custom_user_attribute_from_request_hook\", \"some-value\")\n\n    def response_hook(span, instance, response):\n        if span and span.is_recording():\n            span.set_attribute(\"custom_user_attribute_from_response_hook\", \"some-value\")\n\n    # Instrument redis with hooks\n    RedisInstrumentor().instrument(request_hook=request_hook, response_hook=response_hook)\n\n    # This will report a span with the default settings and the custom attributes added from the hooks\n    client = redis.StrictRedis(host=\"localhost\", port=6379)\n    client.get(\"my-key\")\n\nAPI\n---\n\"\"\"\nimport typing\nfrom typing import Any, Collection\n\nimport redis\nfrom wrapt import wrap_function_wrapper\n\nfrom opentelemetry import trace\nfrom opentelemetry.instrumentation.instrumentor import BaseInstrumentor\nfrom opentelemetry.instrumentation.redis.package import _instruments\nfrom opentelemetry.instrumentation.redis.util import (\n    _extract_conn_attributes,\n    _format_command_args,\n)\nfrom opentelemetry.instrumentation.redis.version import __version__\nfrom opentelemetry.instrumentation.utils import unwrap\nfrom opentelemetry.semconv.trace import SpanAttributes\nfrom opentelemetry.trace import Span\n\n_DEFAULT_SERVICE = \"redis\"\n\n_RequestHookT = typing.Optional[\n    typing.Callable[\n        [Span, redis.connection.Connection, typing.List, typing.Dict], None\n    ]\n]\n_ResponseHookT = typing.Optional[\n    typing.Callable[[Span, redis.connection.Connection, Any], None]\n]\n\n_REDIS_ASYNCIO_VERSION = (4, 2, 0)\nif redis.VERSION >= _REDIS_ASYNCIO_VERSION:\n    import redis.asyncio\n\n_REDIS_CLUSTER_VERSION = (4, 1, 0)\n_REDIS_ASYNCIO_CLUSTER_VERSION = (4, 3, 0)\n\n\ndef _set_connection_attributes(span, conn):\n    if not span.is_recording() or not hasattr(conn, \"connection_pool\"):\n        return\n    for key, value in _extract_conn_attributes(\n        conn.connection_pool.connection_kwargs\n    ).items():\n        span.set_attribute(key, value)\n\n\ndef _instrument(\n    tracer,\n    request_hook: _RequestHookT = None,\n    response_hook: _ResponseHookT = None,\n):\n    def _traced_execute_command(func, instance, args, kwargs):\n        query = _format_command_args(args)\n        name = \"\"\n        if len(args) > 0 and args[0]:\n            name = args[0]\n        else:\n            name = instance.connection_pool.connection_kwargs.get(\"db\", 0)\n        with tracer.start_as_current_span(\n            name, kind=trace.SpanKind.CLIENT\n        ) as span:\n            if span.is_recording():\n                span.set_attribute(SpanAttributes.DB_STATEMENT, query)\n                _set_connection_attributes(span, instance)\n                span.set_attribute(\"db.redis.args_length\", len(args))\n            if callable(request_hook):\n                request_hook(span, instance, args, kwargs)\n            response = func(*args, **kwargs)\n            if callable(response_hook):\n                response_hook(span, instance, response)\n            return response\n\n    def _traced_execute_pipeline(func, instance, args, kwargs):\n        try:\n            command_stack = (\n                instance.command_stack\n                if hasattr(instance, \"command_stack\")\n                else instance._command_stack\n            )\n\n            cmds = [\n                _format_command_args(c.args if hasattr(c, \"args\") else c[0])\n                for c in command_stack\n            ]\n            resource = \"\\n\".join(cmds)\n\n            span_name = \" \".join(\n                [\n                    (c.args[0] if hasattr(c, \"args\") else c[0][0])\n                    for c in command_stack\n                ]\n            )\n        except (AttributeError, IndexError):\n            command_stack = []\n            resource = \"\"\n            span_name = \"\"\n\n        with tracer.start_as_current_span(\n            span_name, kind=trace.SpanKind.CLIENT\n        ) as span:\n            if span.is_recording():\n                span.set_attribute(SpanAttributes.DB_STATEMENT, resource)\n                _set_connection_attributes(span, instance)\n                span.set_attribute(\n                    \"db.redis.pipeline_length\", len(command_stack)\n                )\n            response = func(*args, **kwargs)\n            if callable(response_hook):\n                response_hook(span, instance, response)\n            return response\n\n    pipeline_class = (\n        \"BasePipeline\" if redis.VERSION < (3, 0, 0) else \"Pipeline\"\n    )\n    redis_class = \"StrictRedis\" if redis.VERSION < (3, 0, 0) else \"Redis\"\n\n    wrap_function_wrapper(\n        \"redis\", f\"{redis_class}.execute_command\", _traced_execute_command\n    )\n    wrap_function_wrapper(\n        \"redis.client\",\n        f\"{pipeline_class}.execute\",\n        _traced_execute_pipeline,\n    )\n    wrap_function_wrapper(\n        \"redis.client\",\n        f\"{pipeline_class}.immediate_execute_command\",\n        _traced_execute_command,\n    )\n    if redis.VERSION >= _REDIS_CLUSTER_VERSION:\n        wrap_function_wrapper(\n            \"redis.cluster\",\n            \"RedisCluster.execute_command\",\n            _traced_execute_command,\n        )\n        wrap_function_wrapper(\n            \"redis.cluster\",\n            \"ClusterPipeline.execute\",\n            _traced_execute_pipeline,\n        )\n    if redis.VERSION >= _REDIS_ASYNCIO_VERSION:\n        wrap_function_wrapper(\n            \"redis.asyncio\",\n            f\"{redis_class}.execute_command\",\n            _traced_execute_command,\n        )\n        wrap_function_wrapper(\n            \"redis.asyncio.client\",\n            f\"{pipeline_class}.execute\",\n            _traced_execute_pipeline,\n        )\n        wrap_function_wrapper(\n            \"redis.asyncio.client\",\n            f\"{pipeline_class}.immediate_execute_command\",\n            _traced_execute_command,\n        )\n    if redis.VERSION >= _REDIS_ASYNCIO_CLUSTER_VERSION:\n        wrap_function_wrapper(\n            \"redis.asyncio.cluster\",\n            \"RedisCluster.execute_command\",\n            _traced_execute_command,\n        )\n        wrap_function_wrapper(\n            \"redis.asyncio.cluster\",\n            \"ClusterPipeline.execute\",\n            _traced_execute_pipeline,\n        )\n\n\nclass RedisInstrumentor(BaseInstrumentor):\n    \"\"\"An instrumentor for Redis\n    See `BaseInstrumentor`\n    \"\"\"\n\n    def instrumentation_dependencies(self) -> Collection[str]:\n        return _instruments\n\n    def _instrument(self, **kwargs):\n        \"\"\"Instruments the redis module\n\n        Args:\n            **kwargs: Optional arguments\n                ``tracer_provider``: a TracerProvider, defaults to global.\n                ``response_hook``: An optional callback which is invoked right before the span is finished processing a response.\n        \"\"\"\n        tracer_provider = kwargs.get(\"tracer_provider\")\n        tracer = trace.get_tracer(\n            __name__, __version__, tracer_provider=tracer_provider\n        )\n        _instrument(\n            tracer,\n            request_hook=kwargs.get(\"request_hook\"),\n            response_hook=kwargs.get(\"response_hook\"),\n        )\n\n    def _uninstrument(self, **kwargs):\n        if redis.VERSION < (3, 0, 0):\n            unwrap(redis.StrictRedis, \"execute_command\")\n            unwrap(redis.StrictRedis, \"pipeline\")\n            unwrap(redis.Redis, \"pipeline\")\n            unwrap(\n                redis.client.BasePipeline,  # pylint:disable=no-member\n                \"execute\",\n            )\n            unwrap(\n                redis.client.BasePipeline,  # pylint:disable=no-member\n                \"immediate_execute_command\",\n            )\n        else:\n            unwrap(redis.Redis, \"execute_command\")\n            unwrap(redis.Redis, \"pipeline\")\n            unwrap(redis.client.Pipeline, \"execute\")\n            unwrap(redis.client.Pipeline, \"immediate_execute_command\")\n        if redis.VERSION >= _REDIS_CLUSTER_VERSION:\n            unwrap(redis.cluster.RedisCluster, \"execute_command\")\n            unwrap(redis.cluster.ClusterPipeline, \"execute\")\n        if redis.VERSION >= _REDIS_ASYNCIO_VERSION:\n            unwrap(redis.asyncio.Redis, \"execute_command\")\n            unwrap(redis.asyncio.Redis, \"pipeline\")\n            unwrap(redis.asyncio.client.Pipeline, \"execute\")\n            unwrap(redis.asyncio.client.Pipeline, \"immediate_execute_command\")\n        if redis.VERSION >= _REDIS_ASYNCIO_CLUSTER_VERSION:\n            unwrap(redis.asyncio.cluster.RedisCluster, \"execute_command\")\n            unwrap(redis.asyncio.cluster.ClusterPipeline, \"execute\")\n", "path": "instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py"}]}