From f2628b4ee2d3131e5012e52e09c6e1b901a56464 Mon Sep 17 00:00:00 2001 From: Ben Sully Date: Fri, 29 Apr 2022 10:54:02 +0100 Subject: [PATCH 1/4] Add failing tests for async Redis client instrumentation --- .../tests/redis/test_redis_functional.py | 117 ++++++++++++++++++ tox.ini | 2 +- 2 files changed, 118 insertions(+), 1 deletion(-) diff --git a/tests/opentelemetry-docker-tests/tests/redis/test_redis_functional.py b/tests/opentelemetry-docker-tests/tests/redis/test_redis_functional.py index b5a4b1702b..0b5d49ca10 100644 --- a/tests/opentelemetry-docker-tests/tests/redis/test_redis_functional.py +++ b/tests/opentelemetry-docker-tests/tests/redis/test_redis_functional.py @@ -12,7 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. +import asyncio + import redis +import redis.asyncio from opentelemetry import trace from opentelemetry.instrumentation.redis import RedisInstrumentor @@ -121,6 +124,120 @@ def test_parent(self): self.assertEqual(child_span.name, "GET") +def async_call(coro): + loop = asyncio.get_event_loop() + return loop.run_until_complete(coro) + + +class TestAsyncRedisInstrument(TestBase): + def setUp(self): + super().setUp() + self.redis_client = redis.asyncio.Redis(port=6379) + async_call(self.redis_client.flushall()) + RedisInstrumentor().instrument(tracer_provider=self.tracer_provider) + + def tearDown(self): + super().tearDown() + RedisInstrumentor().uninstrument() + + def _check_span(self, span, name): + self.assertEqual(span.name, name) + self.assertIs(span.status.status_code, trace.StatusCode.UNSET) + self.assertEqual(span.attributes.get(SpanAttributes.DB_NAME), 0) + self.assertEqual( + span.attributes[SpanAttributes.NET_PEER_NAME], "localhost" + ) + self.assertEqual(span.attributes[SpanAttributes.NET_PEER_PORT], 6379) + + def test_long_command(self): + async_call(self.redis_client.mget(*range(1000))) + + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 1) + span = spans[0] + self._check_span(span, "MGET") + self.assertTrue( + span.attributes.get(SpanAttributes.DB_STATEMENT).startswith( + "MGET 0 1 2 3" + ) + ) + self.assertTrue( + span.attributes.get(SpanAttributes.DB_STATEMENT).endswith("...") + ) + + def test_basics(self): + self.assertIsNone(async_call(self.redis_client.get("cheese"))) + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 1) + span = spans[0] + self._check_span(span, "GET") + self.assertEqual( + span.attributes.get(SpanAttributes.DB_STATEMENT), "GET cheese" + ) + self.assertEqual(span.attributes.get("db.redis.args_length"), 2) + + def test_pipeline_traced(self): + async def pipeline_simple(): + async with self.redis_client.pipeline( + transaction=False + ) as pipeline: + pipeline.set("blah", 32) + pipeline.rpush("foo", "éé") + pipeline.hgetall("xxx") + await pipeline.execute() + + async_call(pipeline_simple()) + + 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(SpanAttributes.DB_STATEMENT), + "SET blah 32\nRPUSH foo éé\nHGETALL xxx", + ) + self.assertEqual(span.attributes.get("db.redis.pipeline_length"), 3) + + def test_pipeline_immediate(self): + async def pipeline_immediate(): + async with self.redis_client.pipeline() as pipeline: + pipeline.set("a", 1) + await pipeline.immediate_execute_command("SET", "b", 2) + await pipeline.execute() + + async_call(pipeline_immediate()) + + spans = self.memory_exporter.get_finished_spans() + # expecting two separate spans here, rather than a + # single span for the whole pipeline + self.assertEqual(len(spans), 2) + span = spans[0] + self._check_span(span, "SET") + self.assertEqual( + span.attributes.get(SpanAttributes.DB_STATEMENT), "SET b 2" + ) + + def test_parent(self): + """Ensure OpenTelemetry works with redis.""" + ot_tracer = trace.get_tracer("redis_svc") + + with ot_tracer.start_as_current_span("redis_get"): + self.assertIsNone(async_call(self.redis_client.get("cheese"))) + + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 2) + child_span, parent_span = spans[0], spans[1] + + # confirm the parenting + self.assertIsNone(parent_span.parent) + self.assertIs(child_span.parent, parent_span.get_span_context()) + + self.assertEqual(parent_span.name, "redis_get") + self.assertEqual(parent_span.instrumentation_info.name, "redis_svc") + + self.assertEqual(child_span.name, "GET") + + class TestRedisDBIndexInstrument(TestBase): def setUp(self): super().setUp() diff --git a/tox.ini b/tox.ini index 87c77b50c5..ca1cea222c 100644 --- a/tox.ini +++ b/tox.ini @@ -491,7 +491,7 @@ deps = psycopg2 ~= 2.8.4 aiopg >= 0.13.0, < 1.3.0 sqlalchemy ~= 1.4 - redis ~= 3.5 + redis ~= 4.2 celery[pytest] >= 4.0, < 6.0 protobuf>=3.13.0 requests==2.25.0 From 8bd43849dad928c89d0db14ef66be24957e68f0a Mon Sep 17 00:00:00 2001 From: Ben Sully Date: Fri, 29 Apr 2022 11:05:56 +0100 Subject: [PATCH 2/4] Instrument async Redis clients similarly to sync clients --- .../instrumentation/redis/__init__.py | 25 +++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py index 080b690099..a67a255c8d 100644 --- a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py @@ -102,6 +102,10 @@ def response_hook(span, instance, response): typing.Callable[[Span, redis.connection.Connection, Any], None] ] +_REDIS_ASYNCIO_VERSION = (4, 2, 0) +if redis.VERSION >= _REDIS_ASYNCIO_VERSION: + import redis.asyncio + def _set_connection_attributes(span, conn): if not span.is_recording(): @@ -176,6 +180,22 @@ def _traced_execute_pipeline(func, instance, args, kwargs): f"{pipeline_class}.immediate_execute_command", _traced_execute_command, ) + if redis.VERSION >= _REDIS_ASYNCIO_VERSION: + wrap_function_wrapper( + "redis.asyncio", + f"{redis_class}.execute_command", + _traced_execute_command, + ) + wrap_function_wrapper( + "redis.asyncio.client", + f"{pipeline_class}.execute", + _traced_execute_pipeline, + ) + wrap_function_wrapper( + "redis.asyncio.client", + f"{pipeline_class}.immediate_execute_command", + _traced_execute_command, + ) class RedisInstrumentor(BaseInstrumentor): @@ -222,3 +242,8 @@ def _uninstrument(self, **kwargs): unwrap(redis.Redis, "pipeline") unwrap(redis.client.Pipeline, "execute") unwrap(redis.client.Pipeline, "immediate_execute_command") + if redis.VERSION >= _REDIS_ASYNCIO_VERSION: + unwrap(redis.asyncio.Redis, "execute_command") + unwrap(redis.asyncio.Redis, "pipeline") + unwrap(redis.asyncio.client.Pipeline, "execute") + unwrap(redis.asyncio.client.Pipeline, "immediate_execute_command") From 0765defe035157537c65801af4d1e162853ffc0a Mon Sep 17 00:00:00 2001 From: Ben Sully Date: Fri, 29 Apr 2022 11:20:19 +0100 Subject: [PATCH 3/4] Update CHANGELOG with async Redis support --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c892e6af8..33af250b7e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - `opentelemetry-instrument` and `opentelemetry-bootstrap` now include a `--version` flag ([#1065](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1065)) +- `opentelemetry-instrumentation-redis` now instruments asynchronous Redis clients, if the installed redis-py includes async support (>=4.2.0). + ([#1076](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1076)) ## [1.11.1-0.30b1](https://github.com/open-telemetry/opentelemetry-python/releases/tag/v1.11.1-0.30b1) - 2022-04-21 From 0498d3e3df0b330f01435dfe6ad9e055598f8f00 Mon Sep 17 00:00:00 2001 From: Ben Sully Date: Mon, 2 May 2022 18:44:36 +0100 Subject: [PATCH 4/4] Add message and example to Redis Usage section --- .../instrumentation/redis/__init__.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py index a67a255c8d..8aaec87952 100644 --- a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py @@ -38,6 +38,22 @@ client = redis.StrictRedis(host="localhost", port=6379) client.get("my-key") +Async Redis clients (i.e. redis.asyncio.Redis) are also instrumented in the same way: + +.. code:: python + + from opentelemetry.instrumentation.redis import RedisInstrumentor + import redis.asyncio + + + # Instrument redis + RedisInstrumentor().instrument() + + # This will report a span with the default settings + async def redis_get(): + client = redis.asyncio.Redis(host="localhost", port=6379) + await client.get("my-key") + The `instrument` method accepts the following keyword args: tracer_provider (TracerProvider) - an optional tracer provider