Support older pika versions

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - `opentelemetry-instrumentation-aws-lambda` Adds support for configurable flush timeout  via `OTEL_INSTRUMENTATION_AWS_LAMBDA_FLUSH_TIMEOUT` property. ([#825](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/825))
+- `opentelemetry-instrumentation-pika` Adds support for versions between `0.12.0` to `1.0.0`. ([#837](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/837))
 
 ### Fixed
 

instrumentation/README.md

@@ -20,7 +20,7 @@
 | [opentelemetry-instrumentation-jinja2](./opentelemetry-instrumentation-jinja2) | jinja2 >= 2.7, < 4.0 |
 | [opentelemetry-instrumentation-logging](./opentelemetry-instrumentation-logging) | logging |
 | [opentelemetry-instrumentation-mysql](./opentelemetry-instrumentation-mysql) | mysql-connector-python ~= 8.0 |
-| [opentelemetry-instrumentation-pika](./opentelemetry-instrumentation-pika) | pika >= 1.1.0 |
+| [opentelemetry-instrumentation-pika](./opentelemetry-instrumentation-pika) | pika >= 0.12.0 |
 | [opentelemetry-instrumentation-psycopg2](./opentelemetry-instrumentation-psycopg2) | psycopg2 >= 2.7.3.1 |
 | [opentelemetry-instrumentation-pymemcache](./opentelemetry-instrumentation-pymemcache) | pymemcache ~= 1.3 |
 | [opentelemetry-instrumentation-pymongo](./opentelemetry-instrumentation-pymongo) | pymongo ~= 3.1 |

instrumentation/opentelemetry-instrumentation-pika/src/opentelemetry/instrumentation/pika/package.py

@@ -13,4 +13,4 @@
 # limitations under the License.
 from typing import Collection
 
-_instruments: Collection[str] = ("pika >= 1.1.0",)
+_instruments: Collection[str] = ("pika >= 0.12.0",)

instrumentation/opentelemetry-instrumentation-pika/src/opentelemetry/instrumentation/pika/pika_instrumentor.py

@@ -14,7 +14,9 @@
 from logging import getLogger
 from typing import Any, Collection, Dict, Optional
 
+import pkg_resources
 import wrapt
+from packaging import version
 from pika.adapters import BlockingConnection
 from pika.adapters.blocking_connection import BlockingChannel
 
@@ -41,8 +43,12 @@ def _instrument_blocking_channel_consumers(
         consume_hook: utils.HookT = utils.dummy_callback,
     ) -> Any:
         for consumer_tag, consumer_info in channel._consumer_infos.items():
+            callback_attr = (
+                PikaInstrumentor._consumer_callback_attribute_name()
+            )
+            consumer_callback = getattr(consumer_info, callback_attr)
             decorated_callback = utils._decorate_callback(
-                consumer_info.on_message_callback,
+                consumer_callback,
                 tracer,
                 consumer_tag,
                 consume_hook,
@@ -51,9 +57,9 @@ def _instrument_blocking_channel_consumers(
             setattr(
                 decorated_callback,
                 "_original_callback",
-                consumer_info.on_message_callback,
+                consumer_callback,
             )
-            consumer_info.on_message_callback = decorated_callback
+            setattr(consumer_info, callback_attr, decorated_callback)
 
     @staticmethod
     def _instrument_basic_publish(
@@ -126,12 +132,27 @@ def uninstrument_channel(channel: BlockingChannel) -> None:
             return
 
         for consumers_tag, client_info in channel._consumer_infos.items():
-            if hasattr(client_info.on_message_callback, "_original_callback"):
+            callback_attr = (
+                PikaInstrumentor._consumer_callback_attribute_name()
+            )
+            consumer_callback = getattr(client_info, callback_attr)
+            if hasattr(consumer_callback, "_original_callback"):
                 channel._consumer_infos[
                     consumers_tag
-                ] = client_info.on_message_callback._original_callback
+                ] = consumer_callback._original_callback
         PikaInstrumentor._uninstrument_channel_functions(channel)
 
+    @staticmethod
+    def _consumer_callback_attribute_name() -> str:
+        pika_version = version.parse(
+            pkg_resources.get_distribution("pika").version
+        )
+        return (
+            "on_message_callback"
+            if pika_version >= version.parse("1.0.0")
+            else "consumer_cb"
+        )
+
     def _decorate_channel_function(
         self,
         tracer_provider: Optional[TracerProvider],

instrumentation/opentelemetry-instrumentation-pika/tests/test_pika_instrumentation.py

@@ -26,7 +26,8 @@ class TestPika(TestCase):
     def setUp(self) -> None:
         self.channel = mock.MagicMock(spec=Channel)
         consumer_info = mock.MagicMock()
-        consumer_info.on_message_callback = mock.MagicMock()
+        callback_attr = PikaInstrumentor._consumer_callback_attribute_name()
+        setattr(consumer_info, callback_attr, mock.MagicMock())
         self.channel._consumer_infos = {"consumer-tag": consumer_info}
         self.mock_callback = mock.MagicMock()
 
@@ -72,8 +73,11 @@ def test_instrument_consumers(
         self, decorate_callback: mock.MagicMock
     ) -> None:
         tracer = mock.MagicMock(spec=Tracer)
+        callback_attr = PikaInstrumentor._consumer_callback_attribute_name()
         expected_decoration_calls = [
-            mock.call(value.on_message_callback, tracer, key, dummy_callback)
+            mock.call(
+                getattr(value, callback_attr), tracer, key, dummy_callback
+            )
             for key, value in self.channel._consumer_infos.items()
         ]
         PikaInstrumentor._instrument_blocking_channel_consumers(

opentelemetry-instrumentation/src/opentelemetry/instrumentation/bootstrap_gen.py

@@ -81,7 +81,7 @@
         "instrumentation": "opentelemetry-instrumentation-mysql==0.27b0",
     },
     "pika": {
-        "library": "pika >= 1.1.0",
+        "library": "pika >= 0.12.0",
         "instrumentation": "opentelemetry-instrumentation-pika==0.27b0",
     },
     "psycopg2": {

tox.ini

@@ -182,8 +182,8 @@ envlist =
     pypy3-test-propagator-ot-trace
 
     ; opentelemetry-instrumentation-pika
-    py3{6,7,8,9,10}-test-instrumentation-pika
-    pypy3-test-instrumentation-pika
+    py3{6,7,8,9,10}-test-instrumentation-pika{0,1}
+    pypy3-test-instrumentation-pika{0,1}
 
     lint
     docker-tests
@@ -216,6 +216,8 @@ deps =
   sqlalchemy11: sqlalchemy>=1.1,<1.2
   sqlalchemy14: aiosqlite
   sqlalchemy14: sqlalchemy~=1.4
+  pika0: pika>=0.12.0,<1.0.0
+  pika1: pika>=1.0.0
 
   ; FIXME: add coverage testing
   ; FIXME: add mypy testing
@@ -249,7 +251,7 @@ changedir =
   test-instrumentation-jinja2: instrumentation/opentelemetry-instrumentation-jinja2/tests
   test-instrumentation-logging: instrumentation/opentelemetry-instrumentation-logging/tests
   test-instrumentation-mysql: instrumentation/opentelemetry-instrumentation-mysql/tests
-  test-instrumentation-pika: instrumentation/opentelemetry-instrumentation-pika/tests
+  test-instrumentation-pika{0,1}: instrumentation/opentelemetry-instrumentation-pika/tests
   test-instrumentation-psycopg2: instrumentation/opentelemetry-instrumentation-psycopg2/tests
   test-instrumentation-pymemcache: instrumentation/opentelemetry-instrumentation-pymemcache/tests
   test-instrumentation-pymongo: instrumentation/opentelemetry-instrumentation-pymongo/tests
@@ -286,7 +288,7 @@ commands_pre =
 
   celery: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-celery[test]
 
-  pika: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-pika[test]
+  pika{0,1}: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-pika[test]
 
   grpc: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-grpc[test]