worker.py

import json
import logging
import time

import pika
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.propagate import extract
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(__name__)

# OpenTelemetry
resource = Resource.create({"service.name": "worker"})
provider = TracerProvider(resource=resource)
otlp_exporter = OTLPSpanExporter()
processor = BatchSpanProcessor(otlp_exporter)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)

# RabbitMQ
RABBITMQ_HOST = "rabbitmq"
QUEUE_NAME = "my_queue"
connection = pika.BlockingConnection(
    pika.ConnectionParameters(RABBITMQ_HOST, heartbeat=3600)
)
channel = connection.channel()

def callback(ch, method, properties, body):
    headers = properties.headers or {}
    ctx = extract(headers)
    with tracer.start_span(
        "consume_message", context=ctx, kind=trace.SpanKind.CONSUMER
    ):
        sleep = json.loads(body)["sleep"]
        logger.warning("Sleeping for %ss", sleep)
        time.sleep(sleep)
        ch.basic_ack(delivery_tag=method.delivery_tag)

channel.queue_declare(queue=QUEUE_NAME, durable=True)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(
    queue=QUEUE_NAME,
    on_message_callback=callback
)
channel.start_consuming()