Added aio-pika instrumentation (#1095)

.github/component_owners.yml

@@ -1,5 +1,8 @@
 components:
 
+  instrumentation/opentelemetry-instrumentation-aio-pika:
+    - ofek1weiss
+
   instrumentation/opentelemetry-instrumentation-boto3sqs:
     - oxeye-nikolay
     - nikosokolik

.github/workflows/test.yml

@@ -6,7 +6,7 @@ on:
     - 'release/*'
   pull_request:
 env:
-  CORE_REPO_SHA: d4d7c67663cc22615748d632e1c8c5799e8eacae
+  CORE_REPO_SHA: 25771ecdac685a5bf7ada1da21092d2061dbfc02
 
 jobs:
   build:

CHANGELOG.md

@@ -38,6 +38,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#1127](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1127))
 - Add metric instrumentation for WSGI
   ([#1128](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1128))
+- `opentelemetry-instrumentation-aio-pika` added RabbitMQ aio-pika module instrumentation.
+  ([#1095](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1095))
 - `opentelemetry-instrumentation-requests` Restoring metrics in requests
   ([#1110](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1110))
 - Integrated sqlcommenter plugin into opentelemetry-instrumentation-django

instrumentation/README.md

@@ -1,6 +1,7 @@
 
 | Instrumentation | Supported Packages | Metrics support |
 | --------------- | ------------------ | --------------- |
+| [opentelemetry-instrumentation-aio-pika](./opentelemetry-instrumentation-aio-pika) | aio_pika ~= 7.2.0 | No
 | [opentelemetry-instrumentation-aiohttp-client](./opentelemetry-instrumentation-aiohttp-client) | aiohttp ~= 3.0 | No
 | [opentelemetry-instrumentation-aiopg](./opentelemetry-instrumentation-aiopg) | aiopg >= 0.13.0, < 1.3.0 | No
 | [opentelemetry-instrumentation-asgi](./opentelemetry-instrumentation-asgi) | asgiref ~= 3.0 | No

instrumentation/opentelemetry-instrumentation-aio-pika/README.rst

@@ -0,0 +1,23 @@
+OpenTelemetry Aio-pika Instrumentation
+======================================
+
+|pypi|
+
+.. |pypi| image:: https://badge.fury.io/py/opentelemetry-instrumentation-aio-pika.svg
+   :target: https://pypi.org/project/opentelemetry-instrumentation-aio-pika/
+
+This library allows tracing requests made by the Aio-pika library.
+
+Installation
+------------
+
+::
+
+    pip install opentelemetry-instrumentation-aio-pika
+
+References
+----------
+
+* `OpenTelemetry Aio-pika instrumentation <https://opentelemetry-python-contrib.readthedocs.io/en/latest/instrumentation/aio-pika/aio-pika.html>`_
+* `OpenTelemetry Project <https://opentelemetry.io/>`_
+* `OpenTelemetry Python Examples <https://github.com/open-telemetry/opentelemetry-python/tree/main/docs/examples>`_

instrumentation/opentelemetry-instrumentation-aio-pika/setup.cfg

@@ -0,0 +1,58 @@
+# 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.
+#
+[metadata]
+name = opentelemetry-instrumentation-aio-pika
+description = OpenTelemetry Aio-pika instrumentation
+long_description = file: README.rst
+long_description_content_type = text/x-rst
+author = OpenTelemetry Authors
+author_email = cncf-opentelemetry-contributors@lists.cncf.io
+url = https://github.com/open-telemetry/opentelemetry-python-contrib/instrumentation/opentelemetry-instrumentation-aio-pika
+platforms = any
+license = Apache-2.0
+classifiers =
+    Development Status :: 4 - Beta
+    Intended Audience :: Developers
+    License :: OSI Approved :: Apache Software License
+    Programming Language :: Python
+    Programming Language :: Python :: 3
+    Programming Language :: Python :: 3.6
+    Programming Language :: Python :: 3.7
+    Programming Language :: Python :: 3.8
+    Programming Language :: Python :: 3.9
+    Programming Language :: Python :: 3.10
+
+[options]
+python_requires = >=3.6
+package_dir=
+    =src
+packages=find_namespace:
+
+install_requires =
+    opentelemetry-api ~= 1.5
+    wrapt >= 1.0.0, < 2.0.0
+
+[options.extras_require]
+test =
+    pytest
+    wrapt >= 1.0.0, < 2.0.0
+    opentelemetry-test-utils == 0.31b0
+
+[options.packages.find]
+where = src
+
+[options.entry_points]
+opentelemetry_instrumentor =
+    aio-pika = opentelemetry.instrumentation.aio_pika:AioPikaInstrumentor

instrumentation/opentelemetry-instrumentation-aio-pika/setup.py

@@ -0,0 +1,99 @@
+# 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.
+
+
+# DO NOT EDIT. THIS FILE WAS AUTOGENERATED FROM templates/instrumentation_setup.py.txt.
+# RUN `python scripts/generate_setup.py` TO REGENERATE.
+
+
+import distutils.cmd
+import json
+import os
+from configparser import ConfigParser
+
+import setuptools
+
+config = ConfigParser()
+config.read("setup.cfg")
+
+# We provide extras_require parameter to setuptools.setup later which
+# overwrites the extras_require section from setup.cfg. To support extras_require
+# section in setup.cfg, we load it here and merge it with the extras_require param.
+extras_require = {}
+if "options.extras_require" in config:
+    for key, value in config["options.extras_require"].items():
+        extras_require[key] = [v for v in value.split("\n") if v.strip()]
+
+BASE_DIR = os.path.dirname(__file__)
+PACKAGE_INFO = {}
+
+VERSION_FILENAME = os.path.join(
+    BASE_DIR,
+    "src",
+    "opentelemetry",
+    "instrumentation",
+    "aio_pika",
+    "version.py",
+)
+with open(VERSION_FILENAME, encoding="utf-8") as f:
+    exec(f.read(), PACKAGE_INFO)
+
+PACKAGE_FILENAME = os.path.join(
+    BASE_DIR,
+    "src",
+    "opentelemetry",
+    "instrumentation",
+    "aio_pika",
+    "package.py",
+)
+with open(PACKAGE_FILENAME, encoding="utf-8") as f:
+    exec(f.read(), PACKAGE_INFO)
+
+# Mark any instruments/runtime dependencies as test dependencies as well.
+extras_require["instruments"] = PACKAGE_INFO["_instruments"]
+test_deps = extras_require.get("test", [])
+for dep in extras_require["instruments"]:
+    test_deps.append(dep)
+
+extras_require["test"] = test_deps
+
+
+class JSONMetadataCommand(distutils.cmd.Command):
+
+    description = (
+        "print out package metadata as JSON. This is used by OpenTelemetry dev scripts to ",
+        "auto-generate code in other places",
+    )
+    user_options = []
+
+    def initialize_options(self):
+        pass
+
+    def finalize_options(self):
+        pass
+
+    def run(self):
+        metadata = {
+            "name": config["metadata"]["name"],
+            "version": PACKAGE_INFO["__version__"],
+            "instruments": PACKAGE_INFO["_instruments"],
+        }
+        print(json.dumps(metadata))
+
+
+setuptools.setup(
+    cmdclass={"meta": JSONMetadataCommand},
+    version=PACKAGE_INFO["__version__"],
+    extras_require=extras_require,
+)

instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/__init__.py

@@ -0,0 +1,60 @@
+# 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.
+"""
+Instrument `aio_pika` to trace RabbitMQ applications.
+
+Usage
+-----
+
+* Start broker backend
+
+.. code-block:: python
+
+    docker run -p 5672:5672 rabbitmq
+
+* Run instrumented task
+
+.. code-block:: python
+
+    import asyncio
+
+    from aio_pika import Message, connect
+    from opentelemetry.instrumentation.aio_pika import AioPikaInstrumentor
+
+    AioPikaInstrumentor().instrument()
+
+
+    async def main() -> None:
+        connection = await connect("amqp://guest:guest@localhost/")
+        async with connection:
+            channel = await connection.channel()
+            queue = await channel.declare_queue("hello")
+            await channel.default_exchange.publish(
+                Message(b"Hello World!"),
+                routing_key=queue.name,
+            )
+
+
+    if __name__ == "__main__":
+        asyncio.run(main())
+
+API
+---
+"""
+# pylint: disable=import-error
+
+from .aio_pika_instrumentor import AioPikaInstrumentor
+from .version import __version__
+
+__all__ = ["AioPikaInstrumentor", "__version__"]

instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/aio_pika_instrumentor.py

@@ -0,0 +1,85 @@
+# 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.
+from typing import Any, Callable, Collection
+
+import wrapt
+from aio_pika import Exchange, Queue
+from aio_pika.abc import AbstractIncomingMessage
+
+from opentelemetry import trace
+from opentelemetry.instrumentation.aio_pika.callback_decorator import (
+    CallbackDecorator,
+)
+from opentelemetry.instrumentation.aio_pika.package import _instruments
+from opentelemetry.instrumentation.aio_pika.publish_decorator import (
+    PublishDecorator,
+)
+from opentelemetry.instrumentation.aio_pika.version import __version__
+from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
+from opentelemetry.instrumentation.utils import unwrap
+from opentelemetry.trace import Tracer
+
+_INSTRUMENTATION_MODULE_NAME = "opentelemetry.instrumentation.aio_pika"
+
+
+class AioPikaInstrumentor(BaseInstrumentor):
+    @staticmethod
+    def _instrument_queue(tracer: Tracer):
+        async def wrapper(wrapped, instance, args, kwargs):
+            async def consume(
+                callback: Callable[[AbstractIncomingMessage], Any],
+                *fargs,
+                **fkwargs
+            ):
+                decorated_callback = CallbackDecorator(
+                    tracer, instance
+                ).decorate(callback)
+                return await wrapped(decorated_callback, *fargs, **fkwargs)
+
+            return await consume(*args, **kwargs)
+
+        wrapt.wrap_function_wrapper(Queue, "consume", wrapper)
+
+    @staticmethod
+    def _instrument_exchange(tracer: Tracer):
+        async def wrapper(wrapped, instance, args, kwargs):
+            decorated_publish = PublishDecorator(tracer, instance).decorate(
+                wrapped
+            )
+            return await decorated_publish(*args, **kwargs)
+
+        wrapt.wrap_function_wrapper(Exchange, "publish", wrapper)
+
+    def _instrument(self, **kwargs):
+        tracer_provider = kwargs.get("tracer_provider", None)
+        tracer = trace.get_tracer(
+            _INSTRUMENTATION_MODULE_NAME, __version__, tracer_provider
+        )
+        self._instrument_queue(tracer)
+        self._instrument_exchange(tracer)
+
+    @staticmethod
+    def _uninstrument_queue():
+        unwrap(Queue, "consume")
+
+    @staticmethod
+    def _uninstrument_exchange():
+        unwrap(Exchange, "publish")
+
+    def _uninstrument(self, **kwargs):
+        self._uninstrument_queue()
+        self._uninstrument_exchange()
+
+    def instrumentation_dependencies(self) -> Collection[str]:
+        return _instruments