fix(redis): add compatibility with redis version 6.2.0 (#14475)

Github Issue: #13598
- Fix Redis asyncio cluster pipeline tracing for redis-py ≥ 6.2.0
- now supports both command_stack and _command_stack since
_command_stack was removed
- Prevents AttributeError and ensures pipelines are traced correctly
across versions

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Author: rachelyangdog

ddtrace/contrib/internal/redis/asyncio_patch.py

@@ -30,6 +30,11 @@ async def instrumented_async_execute_cluster_pipeline(func, instance, args, kwar
     if not pin or not pin.enabled():
         return await func(*args, **kwargs)
 
-    cmds = [stringify_cache_args(c.args, cmd_max_len=config.redis.cmd_max_length) for c in instance._command_stack]
+    # Try to access command_stack, fallback to _command_stack for backward compatibility
+    command_stack = getattr(instance, "command_stack", None)
+    if command_stack is None:
+        command_stack = getattr(instance, "_command_stack", [])
+
+    cmds = [stringify_cache_args(c.args, cmd_max_len=config.redis.cmd_max_length) for c in command_stack]
     with _instrument_redis_execute_pipeline(pin, config.redis, cmds, instance):
         return await func(*args, **kwargs)

releasenotes/notes/fix-redis-command-stack-2ef2be7f28c2e9ee.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    redis: Fix pipeline compatibility with redis >= 6.2.0 after ``_command_stack`` removal.

tests/contrib/redis/test_redis_cluster_asyncio.py

@@ -115,6 +115,41 @@ async def test_pipeline(traced_redis_cluster):
     assert span.get_metric("redis.pipeline_length") == 3
 
 
+@pytest.mark.snapshot(wait_for_num_traces=1)
+@pytest.mark.asyncio
+async def test_pipeline_command_stack_count_matches_metric(redis_cluster):
+    patch()
+    try:
+        async with redis_cluster.pipeline(transaction=False) as p:
+            p.set("a", 1)
+            p.get("a")
+            await p.execute()
+    finally:
+        unpatch()
+
+
+@pytest.mark.asyncio
+async def test_pipeline_command_stack_parity_when_visible(traced_redis_cluster):
+    cluster, test_spans = traced_redis_cluster
+    async with cluster.pipeline(transaction=False) as p:
+        p.set("a", 1)
+        p.get("a")
+        queued = None
+        if hasattr(p, "command_stack"):
+            queued = len(p.command_stack)
+        elif hasattr(p, "_command_stack"):
+            queued = len(p._command_stack)
+        await p.execute()
+
+    traces = test_spans.pop_traces()
+    spans = traces[0]
+    span = spans[0]
+    assert span.resource == "SET\nGET"
+    assert span.get_metric("redis.pipeline_length") == 2
+    if queued is not None:
+        assert span.get_metric("redis.pipeline_length") == queued
+
+
 @pytest.mark.skipif(redis.VERSION < (4, 3, 0), reason="redis.asyncio.cluster is not implemented in redis<4.3.0")
 @pytest.mark.asyncio
 async def test_patch_unpatch(redis_cluster):

tests/snapshots/tests.contrib.redis.test_redis_cluster_asyncio.test_pipeline_command_stack_count_matches_metric.json

@@ -0,0 +1,32 @@
+[[
+  {
+    "name": "redis.command",
+    "service": "redis",
+    "resource": "SET\nGET",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "redis",
+    "error": 0,
+    "meta": {
+      "_dd.base_service": "tests.contrib.redis",
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "68b8917800000000",
+      "component": "redis",
+      "db.system": "redis",
+      "language": "python",
+      "redis.raw_command": "SET a 1\nGET a",
+      "runtime-id": "73c800bba8084530a15a15dbcea1a8bc",
+      "span.kind": "client"
+    },
+    "metrics": {
+      "_dd.measured": 1,
+      "_dd.top_level": 1,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1,
+      "process_id": 4422,
+      "redis.pipeline_length": 2
+    },
+    "duration": 1411000,
+    "start": 1756926328130176000
+  }]]