Porting redis instrumentation from contrib repo

.isort.cfg

@@ -15,4 +15,4 @@ multi_line_output=3
 skip=target
 skip_glob=**/gen/*,.venv*/*,venv*/*
 known_first_party=opentelemetry,opentelemetry_example_app
-known_third_party=psutil,pytest
+known_third_party=psutil,pytest,redis,redis_opentracing

docs/ext/redis/redis.rst

@@ -0,0 +1,7 @@
+OpenTelemetry Redis Instrumentation
+===================================
+
+.. automodule:: opentelemetry.instrumentation.redis
+    :members:
+    :undoc-members:
+    :show-inheritance:

ext/opentelemetry-ext-docker-tests/tests/check_availability.py

@@ -18,6 +18,7 @@
 import mysql.connector
 import psycopg2
 import pymongo
+import redis
 
 MONGODB_COLLECTION_NAME = "test"
 MONGODB_DB_NAME = os.getenv("MONGODB_DB_NAME", "opentelemetry-tests")
@@ -33,76 +34,82 @@
 POSTGRES_PASSWORD = os.getenv("POSTGRESQL_HOST", "testpassword")
 POSTGRES_PORT = int(os.getenv("POSTGRESQL_PORT", "5432"))
 POSTGRES_USER = os.getenv("POSTGRESQL_HOST", "testuser")
+REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
+REDIS_PORT = int(os.getenv("REDIS_PORT ", "6379"))
 RETRY_COUNT = 5
 RETRY_INTERVAL = 5  # Seconds
 
 logger = logging.getLogger(__name__)
 
 
+def retryable(func):
+    def wrapper():
+        # Try to connect to DB
+        for i in range(RETRY_COUNT):
+            try:
+                func()
+                break
+            except Exception as ex:  # pylint: disable=broad-except
+                if i == RETRY_COUNT - 1:
+                    raise ex
+                logger.error(
+                    "waiting for %s, retry %d/%d [%s]",
+                    func.__name__,
+                    i + 1,
+                    RETRY_COUNT,
+                    ex,
+                )
+            time.sleep(RETRY_INTERVAL)
+
+    return wrapper
+
+
 def check_pymongo_connection():
-    # Try to connect to DB
-    for i in range(RETRY_COUNT):
-        try:
-            client = pymongo.MongoClient(
-                MONGODB_HOST, MONGODB_PORT, serverSelectionTimeoutMS=2000
-            )
-            db = client[MONGODB_DB_NAME]
-            collection = db[MONGODB_COLLECTION_NAME]
-            collection.find_one()
-            client.close()
-            break
-        except Exception as ex:
-            if i == RETRY_COUNT - 1:
-                raise (ex)
-            logger.exception(ex)
-        time.sleep(RETRY_INTERVAL)
+    client = pymongo.MongoClient(
+        MONGODB_HOST, MONGODB_PORT, serverSelectionTimeoutMS=2000
+    )
+    db = client[MONGODB_DB_NAME]
+    collection = db[MONGODB_COLLECTION_NAME]
+    collection.find_one()
+    client.close()
 
 
+@retryable
 def check_mysql_connection():
-    # Try to connect to DB
-    for i in range(RETRY_COUNT):
-        try:
-            connection = mysql.connector.connect(
-                user=MYSQL_USER,
-                password=MYSQL_PASSWORD,
-                host=MYSQL_HOST,
-                port=MYSQL_PORT,
-                database=MYSQL_DB_NAME,
-            )
-            connection.close()
-            break
-        except Exception as ex:
-            if i == RETRY_COUNT - 1:
-                raise (ex)
-            logger.exception(ex)
-        time.sleep(RETRY_INTERVAL)
+    connection = mysql.connector.connect(
+        user=MYSQL_USER,
+        password=MYSQL_PASSWORD,
+        host=MYSQL_HOST,
+        port=MYSQL_PORT,
+        database=MYSQL_DB_NAME,
+    )
+    connection.close()
 
 
+@retryable
 def check_postgres_connection():
-    # Try to connect to DB
-    for i in range(RETRY_COUNT):
-        try:
-            connection = psycopg2.connect(
-                dbname=POSTGRES_DB_NAME,
-                user=POSTGRES_USER,
-                password=POSTGRES_PASSWORD,
-                host=POSTGRES_HOST,
-                port=POSTGRES_PORT,
-            )
-            connection.close()
-            break
-        except Exception as ex:
-            if i == RETRY_COUNT - 1:
-                raise (ex)
-            logger.exception(ex)
-        time.sleep(RETRY_INTERVAL)
+    connection = psycopg2.connect(
+        dbname=POSTGRES_DB_NAME,
+        user=POSTGRES_USER,
+        password=POSTGRES_PASSWORD,
+        host=POSTGRES_HOST,
+        port=POSTGRES_PORT,
+    )
+    connection.close()
+
+
+@retryable
+def check_redis_connection():
+    connection = redis.Redis(host=REDIS_HOST, port=REDIS_PORT)
+    connection.hgetall("*")
 
 
 def check_docker_services_availability():
     # Check if Docker services accept connections
     check_pymongo_connection()
     check_mysql_connection()
     check_postgres_connection()
+    check_redis_connection()
 
 
 check_docker_services_availability()

ext/opentelemetry-ext-docker-tests/tests/docker-compose.yml

@@ -23,4 +23,8 @@ services:
       POSTGRES_USER: testuser
       POSTGRES_PASSWORD: testpassword
       POSTGRES_DB: opentelemetry-tests
+  otredis:
+      image: redis:4.0-alpine
+      ports:
+          - "127.0.0.1:6379:6379"
 

ext/opentelemetry-ext-docker-tests/tests/redis/test_redis_functional.py

@@ -0,0 +1,144 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import redis
+
+from opentelemetry import trace
+from opentelemetry.instrumentation.redis import RedisInstrumentor
+from opentelemetry.test.test_base import TestBase
+
+
+class TestRedisPatch(TestBase):
+
+    TEST_SERVICE = "redis"
+    TEST_PORT = 6379
+
+    def setUp(self):
+        super().setUp()
+        self.redis_client = redis.Redis(port=self.TEST_PORT)
+        self.redis_client.flushall()
+        RedisInstrumentor().instrument(tracer_provider=self.tracer_provider)
+
+    def tearDown(self):
+        super().tearDown()
+        RedisInstrumentor().uninstrument()
+
+    def test_long_command(self):
+        self.redis_client.mget(*range(1000))
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        span = spans[0]
+        self.assertEqual(span.attributes["service"], self.TEST_SERVICE)
+        self.assertEqual(span.name, "redis.command")
+        self.assertIs(
+            span.status.canonical_code, trace.status.StatusCanonicalCode.OK
+        )
+
+        self.assertEqual(span.attributes.get("db.instance"), 0)
+        self.assertEqual(
+            span.attributes.get("db.url"), "redis://localhost:6379"
+        )
+
+        self.assertTrue(
+            span.attributes.get("db.statement").startswith("MGET 0 1 2 3")
+        )
+        self.assertTrue(span.attributes.get("db.statement").endswith("..."))
+
+    def test_basics(self):
+        self.assertIsNone(self.redis_client.get("cheese"))
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        span = spans[0]
+        self.assertEqual(span.attributes["service"], self.TEST_SERVICE)
+        self.assertEqual(span.name, "redis.command")
+        self.assertIs(
+            span.status.canonical_code, trace.status.StatusCanonicalCode.OK
+        )
+        self.assertEqual(span.attributes.get("db.instance"), 0)
+        self.assertEqual(
+            span.attributes.get("db.url"), "redis://localhost:6379"
+        )
+        self.assertEqual(span.attributes.get("db.statement"), "GET cheese")
+        self.assertEqual(span.attributes.get("redis.args_length"), 2)
+
+    def test_pipeline_traced(self):
+        with self.redis_client.pipeline(transaction=False) as pipeline:
+            pipeline.set("blah", 32)
+            pipeline.rpush("foo", "éé")
+            pipeline.hgetall("xxx")
+            pipeline.execute()
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        span = spans[0]
+        self.assertEqual(span.attributes["service"], self.TEST_SERVICE)
+        self.assertEqual(span.name, "redis.command")
+        self.assertIs(
+            span.status.canonical_code, trace.status.StatusCanonicalCode.OK
+        )
+        self.assertEqual(span.attributes.get("db.instance"), 0)
+        self.assertEqual(
+            span.attributes.get("db.url"), "redis://localhost:6379"
+        )
+        self.assertEqual(
+            span.attributes.get("db.statement"),
+            "SET blah 32\nRPUSH foo éé\nHGETALL xxx",
+        )
+        self.assertEqual(span.attributes.get("redis.pipeline_length"), 3)
+
+    def test_pipeline_immediate(self):
+        with self.redis_client.pipeline() as pipeline:
+            pipeline.set("a", 1)
+            pipeline.immediate_execute_command("SET", "b", 2)
+            pipeline.execute()
+
+        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.assertEqual(span.attributes["service"], self.TEST_SERVICE)
+        self.assertEqual(span.name, "redis.command")
+        self.assertEqual(span.attributes.get("db.statement"), "SET b 2")
+        self.assertIs(
+            span.status.canonical_code, trace.status.StatusCanonicalCode.OK
+        )
+        self.assertEqual(span.attributes.get("db.instance"), 0)
+        self.assertEqual(
+            span.attributes.get("db.url"), "redis://localhost:6379"
+        )
+
+    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(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)
+
+        self.assertEqual(parent_span.name, "redis_get")
+        self.assertEqual(parent_span.instrumentation_info.name, "redis_svc")
+
+        self.assertEqual(
+            child_span.attributes.get("service"), self.TEST_SERVICE
+        )
+        self.assertEqual(child_span.name, "redis.command")

ext/opentelemetry-instrumentation-redis/CHANGELOG.md

@@ -0,0 +1,5 @@
+# Changelog
+
+## Unreleased
+
+- Initial release

ext/opentelemetry-instrumentation-redis/MANIFEST.in

@@ -0,0 +1,9 @@
+graft src
+graft tests
+global-exclude *.pyc
+global-exclude *.pyo
+global-exclude __pycache__/*
+include CHANGELOG.md
+include MANIFEST.in
+include README.rst
+include LICENSE

ext/opentelemetry-instrumentation-redis/README.rst

@@ -0,0 +1,23 @@
+OpenTelemetry Redis Instrumentation
+===================================
+
+|pypi|
+
+.. |pypi| image:: https://badge.fury.io/py/opentelemetry-instrumentation-redis.svg
+   :target: https://pypi.org/project/opentelemetry-instrumentation-redis/
+
+This library allows tracing requests made by the Redis library.
+
+Installation
+------------
+
+::
+
+    pip install opentelemetry-instrumentation-redis
+
+
+References
+----------
+
+* `OpenTelemetry Redis Instrumentation <https://opentelemetry-python.readthedocs.io/en/latest/instrumentation/opentelemetry-instrumentation-redis/opentelemetry-instrumentation-redis.html>`_
+* `OpenTelemetry Project <https://opentelemetry.io/>`_