From fa56c6c09f86b9f8c87c6ddee403cdda8221d9bc Mon Sep 17 00:00:00 2001 From: Ben Sully Date: Thu, 5 May 2022 02:09:29 +0100 Subject: [PATCH] Instrument redis.asyncio clients (#1076) --- CHANGELOG.md | 2 + .../instrumentation/redis/__init__.py | 41 ++++++ .../tests/redis/test_redis_functional.py | 117 ++++++++++++++++++ tox.ini | 2 +- 4 files changed, 161 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c892e6af..33af250b7 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 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 080b69009..8aaec8795 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 @@ Usage 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 @@ -102,6 +118,10 @@ _ResponseHookT = typing.Optional[ 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 +196,22 @@ def _instrument( 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 +258,8 @@ class RedisInstrumentor(BaseInstrumentor): 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") 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 b5a4b1702..0b5d49ca1 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 @@ class TestRedisInstrument(TestBase): 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 87c77b50c..ca1cea222 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