We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Implement sensitive data sanitization for redis instrumentation
Aggregated by #1543 

--- 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 _REDIS_CLUSTER_VERSION = (4, 1, 0)
126 _REDIS_ASYNCIO_CLUSTER_VERSION = (4, 3, 2)
127 
128 
129 def _set_connection_attributes(span, conn):
130     if not span.is_recording() or not hasattr(conn, "connection_pool"):
131         return
132     for key, value in _extract_conn_attributes(
133         conn.connection_pool.connection_kwargs
134     ).items():
135         span.set_attribute(key, value)
136 
137 
138 def _instrument(
139     tracer,
140     request_hook: _RequestHookT = None,
141     response_hook: _ResponseHookT = None,
142 ):
143     def _traced_execute_command(func, instance, args, kwargs):
144         query = _format_command_args(args)
145         if len(args) > 0 and args[0]:
146             name = args[0]
147         else:
148             name = instance.connection_pool.connection_kwargs.get("db", 0)
149         with tracer.start_as_current_span(
150             name, kind=trace.SpanKind.CLIENT
151         ) as span:
152             if span.is_recording():
153                 span.set_attribute(SpanAttributes.DB_STATEMENT, query)
154                 _set_connection_attributes(span, instance)
155                 span.set_attribute("db.redis.args_length", len(args))
156             if callable(request_hook):
157                 request_hook(span, instance, args, kwargs)
158             response = func(*args, **kwargs)
159             if callable(response_hook):
160                 response_hook(span, instance, response)
161             return response
162 
163     def _traced_execute_pipeline(func, instance, args, kwargs):
164         try:
165             command_stack = (
166                 instance.command_stack
167                 if hasattr(instance, "command_stack")
168                 else instance._command_stack
169             )
170 
171             cmds = [
172                 _format_command_args(c.args if hasattr(c, "args") else c[0])
173                 for c in command_stack
174             ]
175             resource = "\n".join(cmds)
176 
177             span_name = " ".join(
178                 [
179                     (c.args[0] if hasattr(c, "args") else c[0][0])
180                     for c in command_stack
181                 ]
182             )
183         except (AttributeError, IndexError):
184             command_stack = []
185             resource = ""
186             span_name = ""
187 
188         with tracer.start_as_current_span(
189             span_name, kind=trace.SpanKind.CLIENT
190         ) as span:
191             if span.is_recording():
192                 span.set_attribute(SpanAttributes.DB_STATEMENT, resource)
193                 _set_connection_attributes(span, instance)
194                 span.set_attribute(
195                     "db.redis.pipeline_length", len(command_stack)
196                 )
197             response = func(*args, **kwargs)
198             if callable(response_hook):
199                 response_hook(span, instance, response)
200             return response
201 
202     pipeline_class = (
203         "BasePipeline" if redis.VERSION < (3, 0, 0) else "Pipeline"
204     )
205     redis_class = "StrictRedis" if redis.VERSION < (3, 0, 0) else "Redis"
206 
207     wrap_function_wrapper(
208         "redis", f"{redis_class}.execute_command", _traced_execute_command
209     )
210     wrap_function_wrapper(
211         "redis.client",
212         f"{pipeline_class}.execute",
213         _traced_execute_pipeline,
214     )
215     wrap_function_wrapper(
216         "redis.client",
217         f"{pipeline_class}.immediate_execute_command",
218         _traced_execute_command,
219     )
220     if redis.VERSION >= _REDIS_CLUSTER_VERSION:
221         wrap_function_wrapper(
222             "redis.cluster",
223             "RedisCluster.execute_command",
224             _traced_execute_command,
225         )
226         wrap_function_wrapper(
227             "redis.cluster",
228             "ClusterPipeline.execute",
229             _traced_execute_pipeline,
230         )
231     if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
232         wrap_function_wrapper(
233             "redis.asyncio",
234             f"{redis_class}.execute_command",
235             _traced_execute_command,
236         )
237         wrap_function_wrapper(
238             "redis.asyncio.client",
239             f"{pipeline_class}.execute",
240             _traced_execute_pipeline,
241         )
242         wrap_function_wrapper(
243             "redis.asyncio.client",
244             f"{pipeline_class}.immediate_execute_command",
245             _traced_execute_command,
246         )
247     if redis.VERSION >= _REDIS_ASYNCIO_CLUSTER_VERSION:
248         wrap_function_wrapper(
249             "redis.asyncio.cluster",
250             "RedisCluster.execute_command",
251             _traced_execute_command,
252         )
253         wrap_function_wrapper(
254             "redis.asyncio.cluster",
255             "ClusterPipeline.execute",
256             _traced_execute_pipeline,
257         )
258 
259 
260 class RedisInstrumentor(BaseInstrumentor):
261     """An instrumentor for Redis
262     See `BaseInstrumentor`
263     """
264 
265     def instrumentation_dependencies(self) -> Collection[str]:
266         return _instruments
267 
268     def _instrument(self, **kwargs):
269         """Instruments the redis module
270 
271         Args:
272             **kwargs: Optional arguments
273                 ``tracer_provider``: a TracerProvider, defaults to global.
274                 ``response_hook``: An optional callback which is invoked right before the span is finished processing a response.
275         """
276         tracer_provider = kwargs.get("tracer_provider")
277         tracer = trace.get_tracer(
278             __name__, __version__, tracer_provider=tracer_provider
279         )
280         _instrument(
281             tracer,
282             request_hook=kwargs.get("request_hook"),
283             response_hook=kwargs.get("response_hook"),
284         )
285 
286     def _uninstrument(self, **kwargs):
287         if redis.VERSION < (3, 0, 0):
288             unwrap(redis.StrictRedis, "execute_command")
289             unwrap(redis.StrictRedis, "pipeline")
290             unwrap(redis.Redis, "pipeline")
291             unwrap(
292                 redis.client.BasePipeline,  # pylint:disable=no-member
293                 "execute",
294             )
295             unwrap(
296                 redis.client.BasePipeline,  # pylint:disable=no-member
297                 "immediate_execute_command",
298             )
299         else:
300             unwrap(redis.Redis, "execute_command")
301             unwrap(redis.Redis, "pipeline")
302             unwrap(redis.client.Pipeline, "execute")
303             unwrap(redis.client.Pipeline, "immediate_execute_command")
304         if redis.VERSION >= _REDIS_CLUSTER_VERSION:
305             unwrap(redis.cluster.RedisCluster, "execute_command")
306             unwrap(redis.cluster.ClusterPipeline, "execute")
307         if redis.VERSION >= _REDIS_ASYNCIO_VERSION:
308             unwrap(redis.asyncio.Redis, "execute_command")
309             unwrap(redis.asyncio.Redis, "pipeline")
310             unwrap(redis.asyncio.client.Pipeline, "execute")
311             unwrap(redis.asyncio.client.Pipeline, "immediate_execute_command")
312         if redis.VERSION >= _REDIS_ASYNCIO_CLUSTER_VERSION:
313             unwrap(redis.asyncio.cluster.RedisCluster, "execute_command")
314             unwrap(redis.asyncio.cluster.ClusterPipeline, "execute")
315
```

Path: `instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.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 Some utils used by the redis integration
17 """
18 from opentelemetry.semconv.trace import (
19     DbSystemValues,
20     NetTransportValues,
21     SpanAttributes,
22 )
23 
24 
25 def _extract_conn_attributes(conn_kwargs):
26     """Transform redis conn info into dict"""
27     attributes = {
28         SpanAttributes.DB_SYSTEM: DbSystemValues.REDIS.value,
29     }
30     db = conn_kwargs.get("db", 0)
31     attributes[SpanAttributes.DB_REDIS_DATABASE_INDEX] = db
32     try:
33         attributes[SpanAttributes.NET_PEER_NAME] = conn_kwargs.get(
34             "host", "localhost"
35         )
36         attributes[SpanAttributes.NET_PEER_PORT] = conn_kwargs.get(
37             "port", 6379
38         )
39         attributes[
40             SpanAttributes.NET_TRANSPORT
41         ] = NetTransportValues.IP_TCP.value
42     except KeyError:
43         attributes[SpanAttributes.NET_PEER_NAME] = conn_kwargs.get("path", "")
44         attributes[
45             SpanAttributes.NET_TRANSPORT
46         ] = NetTransportValues.UNIX.value
47 
48     return attributes
49 
50 
51 def _format_command_args(args):
52     """Format command arguments and trim them as needed"""
53     value_max_len = 100
54     value_too_long_mark = "..."
55     cmd_max_len = 1000
56     length = 0
57     out = []
58     for arg in args:
59         cmd = str(arg)
60 
61         if len(cmd) > value_max_len:
62             cmd = cmd[:value_max_len] + value_too_long_mark
63 
64         if length + len(cmd) > cmd_max_len:
65             prefix = cmd[: cmd_max_len - length]
66             out.append(f"{prefix}{value_too_long_mark}")
67             break
68 
69         out.append(cmd)
70         length += len(cmd)
71 
72     return " ".join(out)
73
```
--- 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
@@ -64,6 +64,8 @@
 response_hook (Callable) - a function with extra user-defined logic to be performed after performing the request
 this function signature is: def response_hook(span: Span, instance: redis.connection.Connection, response) -> None
 
+sanitize_query (Boolean) - default False, enable the Redis query sanitization
+
 for example:
 
 .. code: python
@@ -139,9 +141,11 @@
     tracer,
     request_hook: _RequestHookT = None,
     response_hook: _ResponseHookT = None,
+    sanitize_query: bool = False,
 ):
     def _traced_execute_command(func, instance, args, kwargs):
-        query = _format_command_args(args)
+        query = _format_command_args(args, sanitize_query)
+
         if len(args) > 0 and args[0]:
             name = args[0]
         else:
@@ -169,7 +173,9 @@
             )
 
             cmds = [
-                _format_command_args(c.args if hasattr(c, "args") else c[0])
+                _format_command_args(
+                    c.args if hasattr(c, "args") else c[0], sanitize_query
+                )
                 for c in command_stack
             ]
             resource = "\n".join(cmds)
@@ -281,6 +287,7 @@
             tracer,
             request_hook=kwargs.get("request_hook"),
             response_hook=kwargs.get("response_hook"),
+            sanitize_query=kwargs.get("sanitize_query", False),
         )
 
     def _uninstrument(self, **kwargs):
diff --git a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py
--- a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py
+++ b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py
@@ -48,11 +48,27 @@
     return attributes
 
 
-def _format_command_args(args):
-    """Format command arguments and trim them as needed"""
-    value_max_len = 100
-    value_too_long_mark = "..."
+def _format_command_args(args, sanitize_query):
+    """Format and sanitize command arguments, and trim them as needed"""
     cmd_max_len = 1000
+    value_too_long_mark = "..."
+    if sanitize_query:
+        # Sanitized query format: "COMMAND ? ?"
+        args_length = len(args)
+        if args_length > 0:
+            out = [str(args[0])] + ["?"] * (args_length - 1)
+            out_str = " ".join(out)
+
+            if len(out_str) > cmd_max_len:
+                out_str = (
+                    out_str[: cmd_max_len - len(value_too_long_mark)]
+                    + value_too_long_mark
+                )
+        else:
+            out_str = ""
+        return out_str
+
+    value_max_len = 100
     length = 0
     out = []
     for arg in args:

{"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@@ -64,6 +64,8 @@\n response_hook (Callable) - a function with extra user-defined logic to be performed after performing the request\n this function signature is: def response_hook(span: Span, instance: redis.connection.Connection, response) -> None\n \n+sanitize_query (Boolean) - default False, enable the Redis query sanitization\n+\n for example:\n \n .. code: python\n@@ -139,9 +141,11 @@\n     tracer,\n     request_hook: _RequestHookT = None,\n     response_hook: _ResponseHookT = None,\n+    sanitize_query: bool = False,\n ):\n     def _traced_execute_command(func, instance, args, kwargs):\n-        query = _format_command_args(args)\n+        query = _format_command_args(args, sanitize_query)\n+\n         if len(args) > 0 and args[0]:\n             name = args[0]\n         else:\n@@ -169,7 +173,9 @@\n             )\n \n             cmds = [\n-                _format_command_args(c.args if hasattr(c, \"args\") else c[0])\n+                _format_command_args(\n+                    c.args if hasattr(c, \"args\") else c[0], sanitize_query\n+                )\n                 for c in command_stack\n             ]\n             resource = \"\\n\".join(cmds)\n@@ -281,6 +287,7 @@\n             tracer,\n             request_hook=kwargs.get(\"request_hook\"),\n             response_hook=kwargs.get(\"response_hook\"),\n+            sanitize_query=kwargs.get(\"sanitize_query\", False),\n         )\n \n     def _uninstrument(self, **kwargs):\ndiff --git a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py\n--- a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py\n+++ b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py\n@@ -48,11 +48,27 @@\n     return attributes\n \n \n-def _format_command_args(args):\n-    \"\"\"Format command arguments and trim them as needed\"\"\"\n-    value_max_len = 100\n-    value_too_long_mark = \"...\"\n+def _format_command_args(args, sanitize_query):\n+    \"\"\"Format and sanitize command arguments, and trim them as needed\"\"\"\n     cmd_max_len = 1000\n+    value_too_long_mark = \"...\"\n+    if sanitize_query:\n+        # Sanitized query format: \"COMMAND ? ?\"\n+        args_length = len(args)\n+        if args_length > 0:\n+            out = [str(args[0])] + [\"?\"] * (args_length - 1)\n+            out_str = \" \".join(out)\n+\n+            if len(out_str) > cmd_max_len:\n+                out_str = (\n+                    out_str[: cmd_max_len - len(value_too_long_mark)]\n+                    + value_too_long_mark\n+                )\n+        else:\n+            out_str = \"\"\n+        return out_str\n+\n+    value_max_len = 100\n     length = 0\n     out = []\n     for arg in args:\n", "issue": "Implement sensitive data sanitization for redis instrumentation\nAggregated by #1543 \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_REDIS_CLUSTER_VERSION = (4, 1, 0)\n_REDIS_ASYNCIO_CLUSTER_VERSION = (4, 3, 2)\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        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"}, {"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\"\"\"\nSome utils used by the redis integration\n\"\"\"\nfrom opentelemetry.semconv.trace import (\n    DbSystemValues,\n    NetTransportValues,\n    SpanAttributes,\n)\n\n\ndef _extract_conn_attributes(conn_kwargs):\n    \"\"\"Transform redis conn info into dict\"\"\"\n    attributes = {\n        SpanAttributes.DB_SYSTEM: DbSystemValues.REDIS.value,\n    }\n    db = conn_kwargs.get(\"db\", 0)\n    attributes[SpanAttributes.DB_REDIS_DATABASE_INDEX] = db\n    try:\n        attributes[SpanAttributes.NET_PEER_NAME] = conn_kwargs.get(\n            \"host\", \"localhost\"\n        )\n        attributes[SpanAttributes.NET_PEER_PORT] = conn_kwargs.get(\n            \"port\", 6379\n        )\n        attributes[\n            SpanAttributes.NET_TRANSPORT\n        ] = NetTransportValues.IP_TCP.value\n    except KeyError:\n        attributes[SpanAttributes.NET_PEER_NAME] = conn_kwargs.get(\"path\", \"\")\n        attributes[\n            SpanAttributes.NET_TRANSPORT\n        ] = NetTransportValues.UNIX.value\n\n    return attributes\n\n\ndef _format_command_args(args):\n    \"\"\"Format command arguments and trim them as needed\"\"\"\n    value_max_len = 100\n    value_too_long_mark = \"...\"\n    cmd_max_len = 1000\n    length = 0\n    out = []\n    for arg in args:\n        cmd = str(arg)\n\n        if len(cmd) > value_max_len:\n            cmd = cmd[:value_max_len] + value_too_long_mark\n\n        if length + len(cmd) > cmd_max_len:\n            prefix = cmd[: cmd_max_len - length]\n            out.append(f\"{prefix}{value_too_long_mark}\")\n            break\n\n        out.append(cmd)\n        length += len(cmd)\n\n    return \" \".join(out)\n", "path": "instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.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\nsanitize_query (Boolean) - default False, enable the Redis query sanitization\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, 2)\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    sanitize_query: bool = False,\n):\n    def _traced_execute_command(func, instance, args, kwargs):\n        query = _format_command_args(args, sanitize_query)\n\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(\n                    c.args if hasattr(c, \"args\") else c[0], sanitize_query\n                )\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            sanitize_query=kwargs.get(\"sanitize_query\", False),\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"}, {"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\"\"\"\nSome utils used by the redis integration\n\"\"\"\nfrom opentelemetry.semconv.trace import (\n    DbSystemValues,\n    NetTransportValues,\n    SpanAttributes,\n)\n\n\ndef _extract_conn_attributes(conn_kwargs):\n    \"\"\"Transform redis conn info into dict\"\"\"\n    attributes = {\n        SpanAttributes.DB_SYSTEM: DbSystemValues.REDIS.value,\n    }\n    db = conn_kwargs.get(\"db\", 0)\n    attributes[SpanAttributes.DB_REDIS_DATABASE_INDEX] = db\n    try:\n        attributes[SpanAttributes.NET_PEER_NAME] = conn_kwargs.get(\n            \"host\", \"localhost\"\n        )\n        attributes[SpanAttributes.NET_PEER_PORT] = conn_kwargs.get(\n            \"port\", 6379\n        )\n        attributes[\n            SpanAttributes.NET_TRANSPORT\n        ] = NetTransportValues.IP_TCP.value\n    except KeyError:\n        attributes[SpanAttributes.NET_PEER_NAME] = conn_kwargs.get(\"path\", \"\")\n        attributes[\n            SpanAttributes.NET_TRANSPORT\n        ] = NetTransportValues.UNIX.value\n\n    return attributes\n\n\ndef _format_command_args(args, sanitize_query):\n    \"\"\"Format and sanitize command arguments, and trim them as needed\"\"\"\n    cmd_max_len = 1000\n    value_too_long_mark = \"...\"\n    if sanitize_query:\n        # Sanitized query format: \"COMMAND ? ?\"\n        args_length = len(args)\n        if args_length > 0:\n            out = [str(args[0])] + [\"?\"] * (args_length - 1)\n            out_str = \" \".join(out)\n\n            if len(out_str) > cmd_max_len:\n                out_str = (\n                    out_str[: cmd_max_len - len(value_too_long_mark)]\n                    + value_too_long_mark\n                )\n        else:\n            out_str = \"\"\n        return out_str\n\n    value_max_len = 100\n    length = 0\n    out = []\n    for arg in args:\n        cmd = str(arg)\n\n        if len(cmd) > value_max_len:\n            cmd = cmd[:value_max_len] + value_too_long_mark\n\n        if length + len(cmd) > cmd_max_len:\n            prefix = cmd[: cmd_max_len - length]\n            out.append(f\"{prefix}{value_too_long_mark}\")\n            break\n\n        out.append(cmd)\n        length += len(cmd)\n\n    return \" \".join(out)\n", "path": "instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py"}]}