opentelemetry-instrumentation-redis: implement suppression of instrumentation in Redis commands and pipelines (#3955)

* 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.

* Doc

* Ruff

* Disable too-many-public-methods for tests file

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

---------

Co-authored-by: Emídio Neto <[email protected]>
Co-authored-by: Riccardo Magliocchetti <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -27,6 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3884](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3884))
 - `opentelemetry-instrumentation-aiohttp-server`: add support for custom header captures via `OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST` and `OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE`
   ([#3916](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3916))
+- `opentelemetry-instrumentation-redis`: add support for `suppress_instrumentation` context manager for both sync and async Redis clients and pipelines
 
 ### Fixed
 

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

@@ -110,6 +110,35 @@ def response_hook(span, instance, response):
     client = redis.StrictRedis(host="localhost", port=6379)
     client.get("my-key")
 
+Suppress Instrumentation
+------------------------
+
+You can use the ``suppress_instrumentation`` context manager to prevent instrumentation
+from being applied to specific Redis operations. This is useful when you want to avoid
+creating spans for internal operations, health checks, or during specific code paths.
+
+.. code:: python
+
+    from opentelemetry.instrumentation.redis import RedisInstrumentor
+    from opentelemetry.instrumentation.utils import suppress_instrumentation
+    import redis
+
+    # Instrument redis
+    RedisInstrumentor().instrument()
+
+    client = redis.StrictRedis(host="localhost", port=6379)
+
+    # This will report a span
+    client.get("my-key")
+
+    # This will NOT report a span
+    with suppress_instrumentation():
+        client.get("internal-key")
+        client.set("cache-key", "value")
+
+    # This will report a span again
+    client.get("another-key")
+
 API
 ---
 """
@@ -134,7 +163,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 +228,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 +266,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 +314,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 +348,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,
@@ -40,6 +41,7 @@
 from opentelemetry.trace import SpanKind
 
 
+# pylint: disable=too-many-public-methods
 class TestRedis(TestBase):
     def assert_span_count(self, count: int):
         """
@@ -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):