fix(redis): build ClusterPipeline span metadata from _execution_strategy in redis-py 6+ (#4728)

* fix(redis): build ClusterPipeline span metadata from _execution_strategy in redis-py 6+

redis-py 6 refactored ClusterPipeline so that queued commands are no
longer appended to `command_stack`; they now live on
`_execution_strategy.command_queue`. As a result the redis instrumentor
emitted ClusterPipeline spans with an empty DB_STATEMENT and a
`db.redis.pipeline_length` of 0.

`_build_span_meta_data_for_pipeline` now reads commands from
`_execution_strategy.command_queue` when present, falling back to
`command_stack` / `_command_stack` for non-cluster pipelines and older
redis-py versions.

Fixes #4084

* chore(redis): rename changelog fragment to PR number

* test(redis): add docker regression test for ClusterPipeline span metadata

Exercises a real RedisCluster pipeline and asserts the redis-py 6+ code
path (command_stack empty, _execution_strategy.command_queue populated)
is genuinely hit before checking the emitted span statement and
db.redis.pipeline_length. Guards against regression #4084.

* Bump redis-py to 6.4.0 in docker tests so the ClusterPipeline regression test exercises the redis-py 6+ code path

* Fix RediSearch import for redis-py 6+ in docker tests

redis-py 6.x renamed redis.commands.search.indexDefinition to
index_definition (snake_case). The camelCase module was removed, so
bumping redis to 6.4.0 broke collection of the entire redis functional
test module. IndexDefinition/IndexType are unchanged in the new module.

* fix(redis): read async ClusterPipeline commands from private _command_queue

The redis-py 6+ sync cluster strategy exposes queued commands via a public
command_queue property, but the async cluster strategy only exposes the
private _command_queue attribute. Reading only command_queue left the async
ClusterPipeline span with an empty statement and a default "redis" name once
the docker tests were bumped to redis-py 6.4.0. Resolve _command_queue as a
fallback so both sync and async cluster pipelines populate span metadata.

---------

Co-authored-by: Tammy Baylis <96076570+tammy-baylis-swi@users.noreply.github.com>
Co-authored-by: Riccardo Magliocchetti <riccardo.magliocchetti@gmail.com>
Co-authored-by: Lukas Hering <40302054+herin049@users.noreply.github.com>

Author: 4 people

.changelog/4728.fixed

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-redis`: extract `ClusterPipeline` commands from `_execution_strategy` in redis-py 6+ so pipeline span attributes include the queued commands instead of being empty

instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py

@@ -207,11 +207,25 @@ def _build_span_meta_data_for_pipeline(
     instance: PipelineInstance | AsyncPipelineInstance,
 ) -> tuple[list[Any], str, str]:
     try:
-        command_stack = (
-            instance.command_stack
-            if hasattr(instance, "command_stack")
-            else instance._command_stack
-        )
+        # redis-py 6+ ClusterPipeline no longer updates ``command_stack``;
+        # queued commands are tracked on the execution strategy instead. The
+        # sync cluster strategy exposes them via a public ``command_queue``
+        # property, while the async cluster strategy only has the private
+        # ``_command_queue`` attribute. Fall back to ``command_stack`` /
+        # ``_command_stack`` for non-cluster pipelines and older redis-py.
+        execution_strategy = getattr(instance, "_execution_strategy", None)
+        if execution_strategy is not None and hasattr(
+            execution_strategy, "command_queue"
+        ):
+            command_stack = execution_strategy.command_queue
+        elif execution_strategy is not None and hasattr(
+            execution_strategy, "_command_queue"
+        ):
+            command_stack = execution_strategy._command_queue
+        elif hasattr(instance, "command_stack"):
+            command_stack = instance.command_stack
+        else:
+            command_stack = instance._command_stack
 
         cmds = [
             _format_command_args(c.args if hasattr(c, "args") else c[0])

instrumentation/opentelemetry-instrumentation-redis/tests/test_redis.py

@@ -22,6 +22,9 @@
     _OpenTelemetrySemanticConventionStability,
 )
 from opentelemetry.instrumentation.redis import RedisInstrumentor
+from opentelemetry.instrumentation.redis.util import (
+    _build_span_meta_data_for_pipeline,
+)
 from opentelemetry.instrumentation.utils import suppress_instrumentation
 from opentelemetry.semconv._incubating.attributes.db_attributes import (
     DB_REDIS_DATABASE_INDEX,
@@ -1649,3 +1652,100 @@ def test_schema_url_combined_mode(self):
                 call_args[1]["schema_url"],
                 "https://opentelemetry.io/schemas/1.25.0",
             )
+
+
+class _FakeCommand:
+    def __init__(self, *args):
+        self.args = args
+
+
+class _FakeExecutionStrategy:
+    def __init__(self, commands):
+        self.command_queue = commands
+
+
+class _FakeClusterPipeline:
+    """Mimics redis-py 6+ ClusterPipeline: queued commands live on
+    ``_execution_strategy.command_queue`` while ``command_stack`` stays empty."""
+
+    def __init__(self, commands):
+        self.command_stack = []
+        self._execution_strategy = _FakeExecutionStrategy(commands)
+
+
+class _FakeAsyncExecutionStrategy:
+    """Mimics redis-py 6+ async cluster strategy: unlike the sync strategy
+    (public ``command_queue`` property), it only exposes the private
+    ``_command_queue`` attribute."""
+
+    def __init__(self, commands):
+        self._command_queue = commands
+
+
+class _FakeAsyncClusterPipeline:
+    """Mimics redis-py 6+ async ClusterPipeline: queued commands live on
+    ``_execution_strategy._command_queue`` and there is no ``command_stack``."""
+
+    def __init__(self, commands):
+        self._execution_strategy = _FakeAsyncExecutionStrategy(commands)
+
+
+class _FakeLegacyPipeline:
+    def __init__(self, commands):
+        self.command_stack = commands
+
+
+class TestBuildSpanMetaDataForPipeline(TestBase):
+    def test_cluster_pipeline_reads_execution_strategy(self):
+        # Regression test for issue #4084: redis-py 6+ ClusterPipeline no
+        # longer populates command_stack, so commands must be read from
+        # _execution_strategy.command_queue.
+        commands = [_FakeCommand("SET", "k1", "v1"), _FakeCommand("GET", "k1")]
+        instance = _FakeClusterPipeline(commands)
+
+        command_stack, resource, span_name = (
+            _build_span_meta_data_for_pipeline(instance)
+        )
+
+        self.assertEqual(len(command_stack), 2)
+        self.assertEqual(resource, "SET ? ?\nGET ?")
+        self.assertEqual(span_name, "SET GET")
+
+    def test_async_cluster_pipeline_reads_private_command_queue(self):
+        # Regression test for issue #4084 on the async path: the redis-py 6+
+        # async cluster strategy exposes only the private ``_command_queue``
+        # (no public ``command_queue`` property), so reading only
+        # ``command_queue`` left the async ClusterPipeline span empty.
+        commands = [_FakeCommand("SET", "k1", "v1"), _FakeCommand("GET", "k1")]
+        instance = _FakeAsyncClusterPipeline(commands)
+
+        command_stack, resource, span_name = (
+            _build_span_meta_data_for_pipeline(instance)
+        )
+
+        self.assertEqual(len(command_stack), 2)
+        self.assertEqual(resource, "SET ? ?\nGET ?")
+        self.assertEqual(span_name, "SET GET")
+
+    def test_legacy_pipeline_still_reads_command_stack(self):
+        commands = [_FakeCommand("SET", "k1", "v1")]
+        instance = _FakeLegacyPipeline(commands)
+
+        command_stack, resource, span_name = (
+            _build_span_meta_data_for_pipeline(instance)
+        )
+
+        self.assertEqual(len(command_stack), 1)
+        self.assertEqual(resource, "SET ? ?")
+        self.assertEqual(span_name, "SET")
+
+    def test_empty_cluster_pipeline_falls_back_to_redis_span_name(self):
+        instance = _FakeClusterPipeline([])
+
+        command_stack, resource, span_name = (
+            _build_span_meta_data_for_pipeline(instance)
+        )
+
+        self.assertEqual(command_stack, [])
+        self.assertEqual(resource, "")
+        self.assertEqual(span_name, "redis")

tests/opentelemetry-docker-tests/tests/redis/test_redis_functional.py

@@ -10,7 +10,7 @@
     TextField,
     VectorField,
 )
-from redis.commands.search.indexDefinition import IndexDefinition, IndexType
+from redis.commands.search.index_definition import IndexDefinition, IndexType
 from redis.commands.search.query import Query
 from redis.exceptions import ResponseError
 
@@ -225,6 +225,42 @@ def test_pipeline_traced(self):
         )
         self.assertEqual(span.attributes.get("db.redis.pipeline_length"), 3)
 
+    def test_cluster_pipeline_span_metadata_regression_4084(self):
+        """Regression test for issue #4084 against a real cluster.
+
+        redis-py 6+ refactored ClusterPipeline so queued commands live on
+        ``_execution_strategy.command_queue`` and no longer populate
+        ``command_stack``. This previously produced a ClusterPipeline span
+        with an empty ``db.statement`` and a ``db.redis.pipeline_length`` of 0.
+        """
+        with self.redis_client.pipeline(transaction=False) as pipeline:
+            pipeline.set("blah", 32)
+            pipeline.rpush("foo", "éé")
+            pipeline.hgetall("xxx")
+
+            # On redis-py 6+ the queued commands are tracked on the execution
+            # strategy rather than command_stack; assert the regression path is
+            # genuinely exercised before checking the emitted span metadata.
+            if hasattr(pipeline, "_execution_strategy") and hasattr(
+                pipeline._execution_strategy, "command_queue"
+            ):
+                self.assertEqual(pipeline.command_stack, [])
+                self.assertEqual(
+                    len(pipeline._execution_strategy.command_queue), 3
+                )
+
+            pipeline.execute()
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        span = spans[0]
+        self._check_span(span, "SET RPUSH HGETALL")
+        self.assertEqual(
+            span.attributes.get(DB_STATEMENT),
+            "SET ? ?\nRPUSH ? ?\nHGETALL ?",
+        )
+        self.assertEqual(span.attributes.get("db.redis.pipeline_length"), 3)
+
     def test_parent(self):
         """Ensure OpenTelemetry works with redis."""
         ot_tracer = trace.get_tracer("redis_svc")

tests/opentelemetry-docker-tests/tests/test-requirements.txt

@@ -41,7 +41,7 @@ pytest==8.0.2
 pytest-celery==0.0.0
 python-dateutil==2.9.0.post0
 pytz==2024.1
-redis==5.0.1
+redis==6.4.0
 requests==2.33.1
 six==1.16.0
 SQLAlchemy==1.4.52