$ python --version
Python 3.10.2 

[[package]]
name = "opentelemetry-instrumentation-pika"
version = "0.29b0"
description = "OpenTelemetry pika instrumentation"
category = "main"
optional = false
python-versions = ">=3.6"

[package.dependencies]
opentelemetry-api = ">=1.5,<2.0"
packaging = ">=20.0"
wrapt = ">=1.0.0,<2.0.0"

[package.extras]
instruments = ["pika (>=0.12.0)"]
test = ["pytest", "wrapt (>=1.0.0,<2.0.0)", "opentelemetry-test-utils (==0.29b0)", "pika (>=0.12.0)"]

pip3 install pytest opentelemetry-autoinstrumentation-pika 'testcontainers[rabbitmq]'

import logging
import uuid
from typing import Any, Callable

import pika
import pytest
from opentelemetry import context, propagate
from opentelemetry import trace
from opentelemetry import trace as trace_api
from opentelemetry.instrumentation.pika import PikaInstrumentor
from opentelemetry.instrumentation.pika.utils import _get_span, _PikaGetter
from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
from opentelemetry.sdk.trace.export import (

    SimpleSpanProcessor,
)
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.semconv.trace import MessagingOperationValues
from opentelemetry.trace import  SpanKind, Tracer
from pika.channel import Channel
from pika.spec import Basic, BasicProperties
from testcontainers.rabbitmq import RabbitMqContainer

MESSAGING_SYSTEM="rabbitmq"

logger = logging.getLogger(__name__)

def unwrap_rabbimq_message(  # pylint: disable=too-many-arguments
    channel: Channel,
    method: Basic.Deliver,
    body: bytes,
    task_name: str,
    tracer: Tracer,
    callback: Callable[[Channel, Basic.Deliver, BasicProperties, bytes], Any],
    properties: BasicProperties = BasicProperties(headers={}),
) -> Any:
    if properties.headers is None:
        properties.headers = {}

    ctx = propagate.extract(properties.headers, getter=_PikaGetter())
    if not ctx:
        ctx = context.get_current()

    token: object = context.attach(ctx)
    span = _get_span(
        tracer,
        channel,
        destination=method.exchange if method.exchange else method.routing_key,
        span_kind=SpanKind.CONSUMER,
        task_name=task_name,
        properties=properties,
        operation=MessagingOperationValues.RECEIVE,
    )
    try:
        if span is not None:
            with trace.use_span(span, end_on_exit=True):
                retval = callback(channel, method, properties, body)
        else:
            retval = callback(channel, method, properties, body)
    finally:
        context.detach(token)
    return retval



def _assert_span_attributes(
    span: ReadableSpan,
    operation: str,
    queue_name: str,
    message_id: str,
    host: str,
    port: str,
) -> None:
    assert span.attributes["messaging.system"] == MESSAGING_SYSTEM
    assert span.attributes["messaging.operation"] == operation
    assert span.attributes["messaging.destination"] == queue_name
    assert span.attributes["messaging.message_id"] == message_id
    assert span.attributes["net.peer.name"] == host
    assert span.attributes["net.peer.port"] == port

def _instrument_python_testing():
    # If it hasn't been intrumented already
    if trace.get_tracer_provider() is None:
        trace_api.set_tracer_provider(TracerProvider())


@pytest.fixture
def inmemory_tracer():
    _instrument_python_testing()

    exporter = InMemorySpanExporter()
    trace.get_tracer_provider().add_span_processor(SimpleSpanProcessor(exporter))
    yield exporter


@pytest.fixture(scope="session")
def rabbitmq_container():
    """
    Make a 'rabbitmq_container' fixture available to test cases.
    """
    with RabbitMqContainer("rabbitmq:3.9.13-management") as container:
        yield container


@pytest.fixture
def pika_connection(rabbitmq_container):  # pylint: disable=redefined-outer-name
    """
    Make a 'pika_connection' fixture available to test cases.
    """
    PikaInstrumentor().instrument()

    parameters = rabbitmq_container.get_connection_params()
    connection = pika.BlockingConnection(parameters=parameters)
    yield connection
    connection.close()



def test_pika_instrumentation(
    pika_connection: pika.BlockingConnection, inmemory_tracer: InMemorySpanExporter
):
    queue = f"test_queue-{uuid.uuid4()}"

    channel = pika_connection.channel()
    channel.queue_declare(queue=queue, durable=False, auto_delete=True)
    channel.basic_publish(exchange="", routing_key=queue, body=b"Hello World!")
    method, _header_frame, _body = channel.basic_get(queue)
    if method:
        channel.basic_ack(method.delivery_tag)

    spans = inmemory_tracer.get_finished_spans()

    # Only basic_produce and basic_consume are instrumented
    assert len(spans) == 1
    producer_span = spans[0]
    assert queue in producer_span.name
    assert producer_span.kind == SpanKind.PRODUCER


def test_pika_manually_extracted_span(
    pika_connection: pika.BlockingConnection, inmemory_tracer: InMemorySpanExporter
):
    queue = f"test_queue-{uuid.uuid4()}"
    tracer = trace.get_tracer(__name__)

    channel = pika_connection.channel()
    channel.queue_declare(queue=queue, durable=False, auto_delete=True)
    channel.basic_publish(exchange="", routing_key=queue, body=b"Hello World!")
    method, properties, body = channel.basic_get(queue)
    value = unwrap_rabbimq_message(
        channel,
        method,
        body,
        f"{__name__}.test_pika_manually_extracted_span",
        tracer,
        lambda channel, method, properties, body: str(body),
        properties=properties,
    )

    if method:
        channel.basic_ack(method.delivery_tag)

    assert value != body

    spans = inmemory_tracer.get_finished_spans()

    # Only basic_produce and basic_consume are instrumented
    assert len(spans) == 2
    producer_span = spans[0]
    assert queue in producer_span.name
    assert producer_span.kind == SpanKind.PRODUCER

    # Manually extracted span
    consumer_span = spans[1]
    assert queue in consumer_span.name
    assert consumer_span.kind == SpanKind.CONSUMER


def test_pika_attributes(
    pika_connection: pika.BlockingConnection, inmemory_tracer: InMemorySpanExporter
):
    queue = f"test_queue-{uuid.uuid4()}"
    tracer = trace.get_tracer(__name__)

    channel = pika_connection.channel()
    channel.queue_declare(queue=queue, durable=False, auto_delete=True)
    channel.basic_publish(exchange="", routing_key=queue, body=b"Hello World!")
    method, properties, body = channel.basic_get(queue)
    value = unwrap_rabbimq_message(
        channel,
        method,
        body,
        f"{__name__}.test_pika_manually_extracted_span",
        tracer,
        lambda channel, method, properties, body: str(body),
        properties=properties,
    )

    if method:
        channel.basic_ack(method.delivery_tag)

    assert value != body

    spans = inmemory_tracer.get_finished_spans()

    # Only basic_produce and basic_consume are instrumented
    assert len(spans) == 2
    producer_span = spans[0]
    assert queue in producer_span.name
    assert producer_span.kind == SpanKind.PRODUCER

    assert producer_span is not None
    # Too maybe errros here :/
    # _assert_span_attributes(
    #     producer_span,
    #     MessagingOperationValues.PROCESS.value,
    #     queue,
    #     properties.message_id,
    #     channel.connection._impl.params.host,
    #     channel.connection._impl.params.port,
    # )

    # Manually extracted span
    consumer_span = spans[1]
    assert queue in consumer_span.name
    assert consumer_span.kind == SpanKind.CONSUMER

    assert consumer_span is not None
    _assert_span_attributes(
        consumer_span,
        MessagingOperationValues.RECEIVE.value,
        queue,
        properties.message_id,
        channel.connection._impl.params.host,
        channel.connection._impl.params.port,
    )

{
    "name": "test_queue-0197462d-bcbb-4742-813b-872fefde4fbf send",
    "context": {
        "trace_id": "0xe46daed06d380a9a1aa650539c00f9b3",
        "span_id": "0xed7efc3b0e7bcafe",
        "trace_state": "[]"
    },
    "kind": "SpanKind.PRODUCER",
    "parent_id": null,
    "start_time": "2022-04-04T13:20:18.665765Z",
    "end_time": "2022-04-04T13:20:18.666079Z",
    "status": {
        "status_code": "UNSET"
    },
    "attributes": {
        "messaging.system": "rabbitmq",
        "messaging.temp_destination": true,
        "messaging.destination": "(temporary)",
        "net.peer.name": "localhost",
        "net.peer.port": 49185
    },
    "events": [],
    "links": [],
    "resource": {
        "telemetry.sdk.language": "python",
        "telemetry.sdk.name": "opentelemetry",
        "telemetry.sdk.version": "1.10.0",
        "service.name": "example-service"
    }
}
{
    "name": "test_queue-0197462d-bcbb-4742-813b-872fefde4fbf receive",
    "context": {
        "trace_id": "0xe46daed06d380a9a1aa650539c00f9b3",
        "span_id": "0x81d0c34964d3e4c4",
        "trace_state": "[]"
    },
    "kind": "SpanKind.CONSUMER",
    "parent_id": "0xed7efc3b0e7bcafe",
    "start_time": "2022-04-04T13:20:18.666806Z",
    "end_time": "2022-04-04T13:20:18.666852Z",
    "status": {
        "status_code": "UNSET"
    },
    "attributes": {
        "messaging.system": "rabbitmq",
        "messaging.operation": "receive",
        "messaging.destination": "tests.test_pika.test_pika_manually_extracted_span",
        "net.peer.name": "localhost",
        "net.peer.port": 49185
    },
    "events": [],
    "links": [],
    "resource": {
        "telemetry.sdk.language": "python",
        "telemetry.sdk.name": "opentelemetry",
        "telemetry.sdk.version": "1.10.0",
        "service.name": "example-service"
    }
}