opentelemetry-instrumentation-redis: implement suppression of instrumentation in Redis commands and pipelines

- Added `suppress_instrumentation` context manager to allow selective disabling of instrumentation for Redis commands and pipelines.
- Updated the Redis instrumentation to check if instrumentation is enabled before executing commands.
- Added unit tests to verify the functionality of the suppression feature for both synchronous and asynchronous Redis operations.

Author: Elyahou

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

@@ -134,7 +134,10 @@ def response_hook(span, instance, response):
     _set_connection_attributes,
 )
 from opentelemetry.instrumentation.redis.version import __version__
-from opentelemetry.instrumentation.utils import unwrap
+from opentelemetry.instrumentation.utils import (
+    is_instrumentation_enabled,
+    unwrap,
+)
 from opentelemetry.semconv._incubating.attributes.db_attributes import (
     DB_STATEMENT,
 )
@@ -196,6 +199,9 @@ def _traced_execute_command(
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
     ) -> R:
+        if not is_instrumentation_enabled():
+            return func(*args, **kwargs)
+
         query = _format_command_args(args)
         name = _build_span_name(instance, args)
         with tracer.start_as_current_span(
@@ -231,6 +237,9 @@ def _traced_execute_pipeline(
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
     ) -> R:
+        if not is_instrumentation_enabled():
+            return func(*args, **kwargs)
+
         (
             command_stack,
             resource,
@@ -276,6 +285,9 @@ async def _async_traced_execute_command(
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
     ) -> Awaitable[R]:
+        if not is_instrumentation_enabled():
+            return await func(*args, **kwargs)
+
         query = _format_command_args(args)
         name = _build_span_name(instance, args)
 
@@ -307,6 +319,9 @@ async def _async_traced_execute_pipeline(
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
     ) -> Awaitable[R]:
+        if not is_instrumentation_enabled():
+            return await func(*args, **kwargs)
+
         (
             command_stack,
             resource,

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

@@ -25,6 +25,7 @@
 
 from opentelemetry import trace
 from opentelemetry.instrumentation.redis import RedisInstrumentor
+from opentelemetry.instrumentation.utils import suppress_instrumentation
 from opentelemetry.semconv._incubating.attributes.db_attributes import (
     DB_REDIS_DATABASE_INDEX,
     DB_SYSTEM,
@@ -390,6 +391,7 @@ def redis_operations():
             self.assertEqual(span.kind, SpanKind.CLIENT)
             self.assertEqual(span.status.status_code, trace.StatusCode.UNSET)
 
+
     def test_span_name_empty_pipeline(self):
         redis_client = fakeredis.FakeStrictRedis()
         pipe = redis_client.pipeline()
@@ -401,6 +403,75 @@ def test_span_name_empty_pipeline(self):
         self.assertEqual(spans[0].kind, SpanKind.CLIENT)
         self.assertEqual(spans[0].status.status_code, trace.StatusCode.UNSET)
 
+    def test_suppress_instrumentation_command(self):
+        redis_client = redis.Redis()
+
+        with mock.patch.object(redis_client, "connection"):
+            # Execute command with suppression
+            with suppress_instrumentation():
+                redis_client.get("key")
+
+        # No spans should be created
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 0)
+
+        # Verify that instrumentation works again after exiting the context
+        with mock.patch.object(redis_client, "connection"):
+            redis_client.get("key")
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+
+    def test_suppress_instrumentation_pipeline(self):
+        redis_client = fakeredis.FakeStrictRedis()
+
+        with suppress_instrumentation():
+            pipe = redis_client.pipeline()
+            pipe.set("key1", "value1")
+            pipe.set("key2", "value2")
+            pipe.get("key1")
+            pipe.execute()
+
+        # No spans should be created
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 0)
+
+        # Verify that instrumentation works again after exiting the context
+        pipe = redis_client.pipeline()
+        pipe.set("key3", "value3")
+        pipe.execute()
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        # Pipeline span could be "SET" or "redis.pipeline" depending on implementation
+        self.assertIn(spans[0].name, ["SET", "redis.pipeline"])
+
+    def test_suppress_instrumentation_mixed(self):
+        redis_client = redis.Redis()
+
+        # Regular instrumented call
+        with mock.patch.object(redis_client, "connection"):
+            redis_client.set("key1", "value1")
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        self.memory_exporter.clear()
+
+        # Suppressed call
+        with suppress_instrumentation():
+            with mock.patch.object(redis_client, "connection"):
+                redis_client.set("key2", "value2")
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 0)
+
+        # Another regular instrumented call
+        with mock.patch.object(redis_client, "connection"):
+            redis_client.get("key1")
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+
 
 class TestRedisAsync(TestBase, IsolatedAsyncioTestCase):
     def assert_span_count(self, count: int):
@@ -570,6 +641,70 @@ async def test_span_name_empty_pipeline(self):
         self.assertEqual(spans[0].status.status_code, trace.StatusCode.UNSET)
         self.instrumentor.uninstrument_client(client=redis_client)
 
+    @pytest.mark.asyncio
+    async def test_suppress_instrumentation_async_command(self):
+        self.instrumentor.instrument(tracer_provider=self.tracer_provider)
+        redis_client = FakeRedis()
+
+        # Execute command with suppression
+        with suppress_instrumentation():
+            await redis_client.get("key")
+
+        # No spans should be created
+        self.assert_span_count(0)
+
+        # Verify that instrumentation works again after exiting the context
+        await redis_client.set("key", "value")
+        self.assert_span_count(1)
+        self.instrumentor.uninstrument()
+
+    @pytest.mark.asyncio
+    async def test_suppress_instrumentation_async_pipeline(self):
+        self.instrumentor.instrument(tracer_provider=self.tracer_provider)
+        redis_client = FakeRedis()
+
+        # Execute pipeline with suppression
+        with suppress_instrumentation():
+            async with redis_client.pipeline() as pipe:
+                await pipe.set("key1", "value1")
+                await pipe.set("key2", "value2")
+                await pipe.get("key1")
+                await pipe.execute()
+
+        # No spans should be created
+        self.assert_span_count(0)
+
+        # Verify that instrumentation works again after exiting the context
+        async with redis_client.pipeline() as pipe:
+            await pipe.set("key3", "value3")
+            await pipe.execute()
+
+        spans = self.assert_span_count(1)
+        # Pipeline span could be "SET" or "redis.pipeline" depending on implementation
+        self.assertIn(spans[0].name, ["SET", "redis.pipeline"])
+        self.instrumentor.uninstrument()
+
+    @pytest.mark.asyncio
+    async def test_suppress_instrumentation_async_mixed(self):
+        self.instrumentor.instrument(tracer_provider=self.tracer_provider)
+        redis_client = FakeRedis()
+
+        # Regular instrumented call
+        await redis_client.set("key1", "value1")
+        self.assert_span_count(1)
+        self.memory_exporter.clear()
+
+        # Suppressed call
+        with suppress_instrumentation():
+            await redis_client.set("key2", "value2")
+
+        self.assert_span_count(0)
+
+        # Another regular instrumented call
+        await redis_client.get("key1")
+        self.assert_span_count(1)
+        self.instrumentor.uninstrument()
+
 
 class TestRedisInstance(TestBase):
     def assert_span_count(self, count: int):