Add Redis instrumentation query sanitization

Add a query sanitizer to the Redis instrumentation. This can be disabled
with the `sanitize_query = False` config option.

Given the query `SET key value`, the sanitized query becomes `SET ? ?`.
Both the keys and values are sanitized, as both can contain PII data.

The Redis queries are sanitized by default. This changes the default
behavior of this instrumentation. Previously it reported unsanitized
Redis queries.

This was previously discussed in the previous implementation of this PR
in PR open-telemetry#1571

Closes open-telemetry#1548

Author: tombruijn

Diff for: CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+
+- `opentelemetry-instrumentation-redis` Add `sanitize_query` config option to allow query sanitization. Enabled by default.
+  ([#1572](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1572))
+
 ## Fixed
 
 - Fix aiopg instrumentation to work with aiopg < 2.0.0

Diff for: instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py

@@ -64,6 +64,8 @@ async def redis_get():
 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 True(Default) or False - enable the Redis query sanization
+
 for example:
 
 .. code: python
@@ -139,9 +141,11 @@ def _instrument(
     tracer,
     request_hook: _RequestHookT = None,
     response_hook: _ResponseHookT = None,
+    sanitize_query: bool = True,
 ):
     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,7 @@ def _traced_execute_pipeline(func, instance, args, kwargs):
             )
 
             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 +285,7 @@ def _instrument(self, **kwargs):
             tracer,
             request_hook=kwargs.get("request_hook"),
             response_hook=kwargs.get("response_hook"),
+            sanitize_query=kwargs.get("sanitize_query", True),
         )
 
     def _uninstrument(self, **kwargs):

Diff for: instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py

@@ -48,25 +48,29 @@ def _extract_conn_attributes(conn_kwargs):
     return attributes
 
 
-def _format_command_args(args):
-    """Format command arguments and trim them as needed"""
-    value_max_len = 100
-    value_too_long_mark = "..."
-    cmd_max_len = 1000
-    length = 0
-    out = []
-    for arg in args:
-        cmd = str(arg)
+def _format_command_args(args, sanitize_query):
+    """Format and sanitize command arguments, and trim them as needed"""
+    if sanitize_query:
+        # Sanitized query format: "COMMAND ? ?"
+        out = [str(args[0])] + ["?"] * (len(args) - 1)
+    else:
+        value_max_len = 100
+        value_too_long_mark = "..."
+        cmd_max_len = 1000
+        length = 0
+        out = []
+        for arg in args:
+            cmd = str(arg)
 
-        if len(cmd) > value_max_len:
-            cmd = cmd[:value_max_len] + value_too_long_mark
+            if len(cmd) > value_max_len:
+                cmd = cmd[:value_max_len] + value_too_long_mark
 
-        if length + len(cmd) > cmd_max_len:
-            prefix = cmd[: cmd_max_len - length]
-            out.append(f"{prefix}{value_too_long_mark}")
-            break
+            if length + len(cmd) > cmd_max_len:
+                prefix = cmd[: cmd_max_len - length]
+                out.append(f"{prefix}{value_too_long_mark}")
+                break
 
-        out.append(cmd)
-        length += len(cmd)
+            out.append(cmd)
+            length += len(cmd)
 
     return " ".join(out)

Diff for: instrumentation/opentelemetry-instrumentation-redis/tests/test_redis.py

@@ -148,6 +148,45 @@ def request_hook(span, conn, args, kwargs):
         span = spans[0]
         self.assertEqual(span.attributes.get(custom_attribute_name), "GET")
 
+    def test_query_sanitizer_enabled(self):
+        redis_client = redis.Redis()
+        connection = redis.connection.Connection()
+        redis_client.connection = connection
+
+        RedisInstrumentor().uninstrument()
+        RedisInstrumentor().instrument(
+            tracer_provider=self.tracer_provider
+        )
+
+        with mock.patch.object(redis_client, "connection"):
+            redis_client.set("key", "value")
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+
+        span = spans[0]
+        self.assertEqual(span.attributes.get("db.statement"), "SET ? ?")
+
+    def test_query_sanitizer_disabled(self):
+        redis_client = redis.Redis()
+        connection = redis.connection.Connection()
+        redis_client.connection = connection
+
+        RedisInstrumentor().uninstrument()
+        RedisInstrumentor().instrument(
+            tracer_provider=self.tracer_provider,
+            sanitize_query=False,
+        )
+
+        with mock.patch.object(redis_client, "connection"):
+            redis_client.set("key", "value")
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+
+        span = spans[0]
+        self.assertEqual(span.attributes.get("db.statement"), "SET key value")
+
     def test_no_op_tracer_provider(self):
         RedisInstrumentor().uninstrument()
         tracer_provider = trace.NoOpTracerProvider