DEV Community

Cover image for OpenTelemetry Celery Instrumentation Guide
Uptrace profile image Alexandr Bandurchin
Alexandr Bandurchin for Uptrace

Posted on • Originally published at uptrace.dev

OpenTelemetry Celery Instrumentation Guide

#performance #tutorial #monitoring #python

OpenTelemetry enables comprehensive monitoring of Celery applications by automatically collecting telemetry data including task execution times, worker performance, queue depths, and error rates. By integrating OpenTelemetry, you can capture distributed traces across your task pipeline and export this data to observability backends for analysis and visualization.

What is Celery?

Celery is a distributed task queue for Python that allows you to run asynchronous and scheduled tasks. It's built on message passing and can operate with multiple brokers like Redis, RabbitMQ, and Amazon SQS. Celery is commonly used for background processing, periodic tasks, and distributed computing.

Celery consists of several components: producers (clients that send tasks), brokers (message transport), workers (processes that execute tasks), and result backends (stores for task results). This architecture makes it ideal for scaling applications horizontally and handling time-consuming operations without blocking user requests.

What is OpenTelemetry?

OpenTelemetry is an open-source observability framework that aims to standardize and simplify the collection, processing, and export of telemetry data from applications and systems.

OpenTelemetry supports multiple programming languages and platforms, making it suitable for a wide range of applications and environments. For detailed Python instrumentation, see the OpenTelemetry Python guide.

OpenTelemetry enables developers to instrument their code and collect telemetry data, which can then be exported to various OpenTelemetry backends or observability platforms for analysis and visualization. Using the OpenTelemetry Collector, you can centralize telemetry data collection, perform data transformations, and route data to multiple observability backends simultaneously.

Installation

To instrument a Celery application with OpenTelemetry, install the required packages:

pip install opentelemetry-api opentelemetry-sdk opentelemetry-instrumentation-celery
Enter fullscreen mode Exit fullscreen mode

Additional Instrumentation

Depending on your broker and result backend, you may also want to install additional instrumentation:

# For Redis broker/backend pip install opentelemetry-instrumentation-redis # For RabbitMQ (if using kombu) pip install opentelemetry-instrumentation-kombu # For database result backends pip install opentelemetry-instrumentation-psycopg2 # PostgreSQL pip install opentelemetry-instrumentation-sqlite3 # SQLite
Enter fullscreen mode Exit fullscreen mode

Exporter Installation

To export telemetry data to observability backends, install an appropriate exporter:

# For OTLP (recommended) pip install opentelemetry-exporter-otlp # For console output (development/testing) pip install opentelemetry-exporter-otlp-proto-http
Enter fullscreen mode Exit fullscreen mode

Basic Instrumentation

Automatic Instrumentation

The simplest way to instrument Celery is using the worker process initialization hook. This ensures proper initialization in Celery's prefork worker model:

from celery import Celery from celery.signals import worker_process_init from opentelemetry.instrumentation.celery import CeleryInstrumentor from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ( BatchSpanProcessor, ConsoleSpanExporter, ) from opentelemetry.sdk.resources import SERVICE_NAME, Resource # Create Celery app first app = Celery('tasks', broker='redis://localhost:6379/0') @worker_process_init.connect(weak=False) def init_celery_tracing(*args, **kwargs): """Initialize OpenTelemetry in each worker process""" # Configure OpenTelemetry  resource = Resource(attributes={ SERVICE_NAME: "celery-worker" }) provider = TracerProvider(resource=resource) processor = BatchSpanProcessor(ConsoleSpanExporter()) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # Instrument Celery  CeleryInstrumentor().instrument() @app.task def add(x, y): return x + y
Enter fullscreen mode Exit fullscreen mode

Manual Instrumentation

For more control over tracing, you can manually instrument specific tasks:

from celery import Celery from celery.signals import worker_process_init from opentelemetry.instrumentation.celery import CeleryInstrumentor from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from opentelemetry.sdk.resources import SERVICE_NAME, Resource # Create Celery app first app = Celery('tasks', broker='redis://localhost:6379/0') @worker_process_init.connect(weak=False) def init_celery_tracing(*args, **kwargs): """Initialize OpenTelemetry in each worker process""" resource = Resource(attributes={SERVICE_NAME: "celery-worker"}) provider = TracerProvider(resource=resource) processor = BatchSpanProcessor(ConsoleSpanExporter()) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # Instrument Celery  CeleryInstrumentor().instrument() # Get tracer after initialization tracer = trace.get_tracer(__name__) @app.task def process_data(data_id): with tracer.start_as_current_span("process_data") as span: # Add custom attributes  span.set_attribute("data.id", data_id) span.set_attribute("worker.name", "data_processor") # Your task logic here  result = expensive_computation(data_id) # Record result information  span.set_attribute("result.size", len(result)) span.set_attribute("task.status", "completed") return result
Enter fullscreen mode Exit fullscreen mode

Worker Configuration

Worker Startup with Instrumentation

Create a worker startup script that initializes OpenTelemetry properly:

# worker.py import os from celery import Celery from celery.signals import worker_process_init from opentelemetry.instrumentation.celery import CeleryInstrumentor from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import SERVICE_NAME, Resource # Create Celery app app = Celery('tasks', broker='redis://localhost:6379/0') @worker_process_init.connect(weak=False) def initialize_tracing(*args, **kwargs): """Initialize OpenTelemetry tracing for Celery worker""" # Create resource with worker information  resource = Resource(attributes={ SERVICE_NAME: "celery-worker", "service.version": "1.0.0", "deployment.environment": os.environ.get("ENVIRONMENT", "development"), "worker.hostname": os.environ.get("HOSTNAME", "unknown"), }) # Configure tracer provider  provider = TracerProvider(resource=resource) # Configure OTLP exporter  otlp_exporter = OTLPSpanExporter( endpoint="http://localhost:4317", insecure=True, ) processor = BatchSpanProcessor(otlp_exporter) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # Instrument Celery  CeleryInstrumentor().instrument() @app.task def example_task(data): # Your task logic here  return f"Processed: {data}"
Enter fullscreen mode Exit fullscreen mode

Worker Execution

Start the worker with the instrumented configuration:

# Start worker with instrumentation celery -A worker worker --loglevel=info # For production with multiple workers celery -A worker worker --loglevel=info --concurrency=4
Enter fullscreen mode Exit fullscreen mode

Producer (Client) Instrumentation

Instrument the client code that sends tasks to Celery:

# client.py from celery import Celery from opentelemetry.instrumentation.celery import CeleryInstrumentor from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import SERVICE_NAME, Resource # Initialize OpenTelemetry for producer resource = Resource(attributes={ SERVICE_NAME: "celery-producer" }) provider = TracerProvider(resource=resource) otlp_exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True) processor = BatchSpanProcessor(otlp_exporter) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # Instrument Celery CeleryInstrumentor().instrument() # Create Celery app app = Celery('tasks', broker='redis://localhost:6379/0') def send_task(): # This will be traced automatically  result = app.send_task('tasks.process_data', args=[123]) return result
Enter fullscreen mode Exit fullscreen mode

Advanced Configuration

Custom Span Attributes

Add custom attributes to Celery spans for better observability:

from celery import Celery from celery.signals import worker_process_init from opentelemetry.instrumentation.celery import CeleryInstrumentor from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from opentelemetry.sdk.resources import SERVICE_NAME, Resource # Create Celery app app = Celery('tasks', broker='redis://localhost:6379/0') @worker_process_init.connect(weak=False) def init_celery_tracing(*args, **kwargs): """Initialize OpenTelemetry in each worker process""" resource = Resource(attributes={SERVICE_NAME: "celery-worker"}) provider = TracerProvider(resource=resource) processor = BatchSpanProcessor(ConsoleSpanExporter()) provider.add_span_processor(processor) trace.set_tracer_provider(provider) def custom_span_processor(span, task): """Custom function to add attributes to Celery spans""" # Add task-specific attributes  span.set_attribute("celery.task.queue", task.request.delivery_info.get('routing_key', 'default')) span.set_attribute("celery.task.retries", task.request.retries) span.set_attribute("celery.task.eta", str(task.request.eta) if task.request.eta else "immediate") # Add worker information  span.set_attribute("celery.worker.hostname", task.request.hostname) # Add custom business logic attributes  if hasattr(task.request, 'correlation_id'): span.set_attribute("business.correlation_id", task.request.correlation_id) # Configure instrumentation with custom processor  CeleryInstrumentor().instrument( span_name_callback=lambda task: f"celery.task.{task.name}", span_processor_callback=custom_span_processor )
Enter fullscreen mode Exit fullscreen mode

Error Handling and Exception Tracking

Enhance error tracking in Celery tasks:

from celery import Celery from celery.signals import worker_process_init from celery.exceptions import Retry from opentelemetry.instrumentation.celery import CeleryInstrumentor from opentelemetry import trace from opentelemetry.trace import Status, StatusCode from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from opentelemetry.sdk.resources import SERVICE_NAME, Resource # Create Celery app app = Celery('tasks', broker='redis://localhost:6379/0') @worker_process_init.connect(weak=False) def init_celery_tracing(*args, **kwargs): """Initialize OpenTelemetry in each worker process""" resource = Resource(attributes={SERVICE_NAME: "celery-worker"}) provider = TracerProvider(resource=resource) processor = BatchSpanProcessor(ConsoleSpanExporter()) provider.add_span_processor(processor) trace.set_tracer_provider(provider) CeleryInstrumentor().instrument() @app.task(bind=True) def risky_task(self, data): span = trace.get_current_span() try: # Your task logic here  result = process_risky_operation(data) # Set success attributes  span.set_attribute("task.result.success", True) span.set_attribute("task.result.items_processed", len(result)) return result except RetryableError as e: # Handle retryable errors  span.record_exception(e) span.set_attribute("task.retry.attempt", self.request.retries + 1) span.set_attribute("task.retry.reason", str(e)) # Retry with exponential backoff  raise self.retry(exc=e, countdown=60, max_retries=3) except Exception as e: # Handle permanent failures  span.record_exception(e) span.set_status(Status(StatusCode.ERROR, str(e))) span.set_attribute("task.error.type", type(e).__name__) span.set_attribute("task.error.fatal", True) raise
Enter fullscreen mode Exit fullscreen mode

Task Result Tracking

Track task results and completion status:

from celery.signals import task_success, task_failure, task_retry from opentelemetry import trace @task_success.connect def task_success_handler(sender=None, result=None, **kwargs): """Handle successful task completion""" span = trace.get_current_span() span.set_attribute("celery.task.status", "success") span.set_attribute("celery.task.result.type", type(result).__name__) @task_failure.connect def task_failure_handler(sender=None, task_id=None, exception=None, traceback=None, einfo=None, **kwargs): """Handle task failures""" span = trace.get_current_span() span.set_attribute("celery.task.status", "failure") span.set_attribute("celery.task.exception", str(exception)) span.record_exception(exception) @task_retry.connect def task_retry_handler(sender=None, task_id=None, reason=None, einfo=None, **kwargs): """Handle task retries""" span = trace.get_current_span() span.set_attribute("celery.task.status", "retry") span.set_attribute("celery.task.retry.reason", str(reason))
Enter fullscreen mode Exit fullscreen mode

Broker-Specific Configuration

Redis Configuration

For Redis broker, add Redis instrumentation:

from celery import Celery from celery.signals import worker_process_init from opentelemetry.instrumentation.redis import RedisInstrumentor from opentelemetry.instrumentation.celery import CeleryInstrumentor from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from opentelemetry.sdk.resources import SERVICE_NAME, Resource # Create Celery app with Redis configuration app = Celery('tasks') app.conf.update( broker_url='redis://localhost:6379/0', result_backend='redis://localhost:6379/0', task_serializer='json', accept_content=['json'], result_serializer='json', timezone='UTC', enable_utc=True, ) @worker_process_init.connect(weak=False) def init_celery_tracing(*args, **kwargs): """Initialize OpenTelemetry in each worker process""" resource = Resource(attributes={SERVICE_NAME: "celery-worker"}) provider = TracerProvider(resource=resource) processor = BatchSpanProcessor(ConsoleSpanExporter()) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # Instrument Redis  RedisInstrumentor().instrument() # Instrument Celery  CeleryInstrumentor().instrument()
Enter fullscreen mode Exit fullscreen mode

RabbitMQ Configuration

For RabbitMQ broker, configure with appropriate instrumentation:

from celery import Celery from celery.signals import worker_process_init from opentelemetry.instrumentation.celery import CeleryInstrumentor from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from opentelemetry.sdk.resources import SERVICE_NAME, Resource # Create Celery app with RabbitMQ configuration app = Celery('tasks') app.conf.update( broker_url='pyamqp://guest@localhost//', result_backend='rpc://', task_serializer='json', accept_content=['json'], result_serializer='json', ) @worker_process_init.connect(weak=False) def init_celery_tracing(*args, **kwargs): """Initialize OpenTelemetry in each worker process""" resource = Resource(attributes={SERVICE_NAME: "celery-worker"}) provider = TracerProvider(resource=resource) processor = BatchSpanProcessor(ConsoleSpanExporter()) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # Instrument Celery (includes kombu instrumentation)  CeleryInstrumentor().instrument()
Enter fullscreen mode Exit fullscreen mode

Monitoring and Metrics

Custom Metrics Collection

Collect custom metrics alongside traces:

from opentelemetry import metrics from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter # Configure metrics metric_reader = PeriodicExportingMetricReader( OTLPMetricExporter(endpoint="http://localhost:4317", insecure=True), export_interval_millis=30000, ) metrics.set_meter_provider(MeterProvider(metric_readers=[metric_reader])) # Create meter and instruments meter = metrics.get_meter("celery_metrics") task_duration_histogram = meter.create_histogram( "celery.task.duration", description="Duration of Celery task execution", unit="ms" ) task_counter = meter.create_counter( "celery.tasks.total", description="Total number of Celery tasks" ) @app.task def monitored_task(data): start_time = time.time() try: # Your task logic  result = process_data(data) # Record metrics  duration = (time.time() - start_time) * 1000 task_duration_histogram.record(duration, {"task_name": "monitored_task", "status": "success"}) task_counter.add(1, {"task_name": "monitored_task", "status": "success"}) return result except Exception as e: # Record failure metrics  duration = (time.time() - start_time) * 1000 task_duration_histogram.record(duration, {"task_name": "monitored_task", "status": "error"}) task_counter.add(1, {"task_name": "monitored_task", "status": "error"}) raise
Enter fullscreen mode Exit fullscreen mode

Production Deployment

Environment Configuration

Use environment variables for production configuration:

import os from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter def configure_tracing(): """Configure OpenTelemetry based on environment variables""" # OTLP endpoint configuration  otlp_endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317") # Service configuration  service_name = os.environ.get("OTEL_SERVICE_NAME", "celery-worker") service_version = os.environ.get("OTEL_SERVICE_VERSION", "1.0.0") environment = os.environ.get("DEPLOYMENT_ENVIRONMENT", "production") # Resource attributes  resource = Resource(attributes={ SERVICE_NAME: service_name, "service.version": service_version, "deployment.environment": environment, }) # Configure exporter  exporter = OTLPSpanExporter(endpoint=otlp_endpoint) processor = BatchSpanProcessor(exporter) provider = TracerProvider(resource=resource) provider.add_span_processor(processor) trace.set_tracer_provider(provider)
Enter fullscreen mode Exit fullscreen mode

Docker Configuration

Example Dockerfile for containerized Celery workers:

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . # Set OpenTelemetry environment variables ENV OTEL_SERVICE_NAME=celery-worker ENV OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4317 # Start worker with instrumentation CMD ["celery", "-A", "worker", "worker", "--loglevel=info"]
Enter fullscreen mode Exit fullscreen mode

What is Uptrace?

Uptrace is a OpenTelemetry APM that supports distributed tracing, metrics, and logs. You can use it to monitor applications and troubleshoot issues. Compare with other top APM tools for task queue monitoring.

Uptrace Overview

Uptrace comes with an intuitive query builder, rich dashboards, alerting rules with notifications, and integrations for most languages and frameworks.

Uptrace can process billions of spans and metrics on a single server and allows you to monitor your applications at 10x lower cost.

In just a few minutes, you can try Uptrace by visiting the cloud demo (no login required) or running it locally with Docker. The source code is available on GitHub.

uptrace-python with Celery

For simplified configuration, you can use the uptrace-python wrapper:

Installation 

pip install uptrace
Enter fullscreen mode Exit fullscreen mode

Configuration 

import uptrace from opentelemetry.instrumentation.celery import CeleryInstrumentor # Configure OpenTelemetry with Uptrace uptrace.configure_opentelemetry( # Set DSN or use UPTRACE_DSN environment variable  dsn="<your-uptrace-dsn>", service_name="celery-worker", service_version="1.0.0", deployment_environment="production", ) # Instrument Celery CeleryInstrumentor().instrument() # Your Celery app from celery import Celery app = Celery('tasks', broker='redis://localhost:6379/0')
Enter fullscreen mode Exit fullscreen mode

Environment Variables 

# Uptrace configuration export UPTRACE_DSN="https://<token>@uptrace.dev/<project_id>" # OpenTelemetry configuration export OTEL_SERVICE_NAME="celery-worker" export OTEL_SERVICE_VERSION="1.0.0" # Start worker celery -A tasks worker --loglevel=info
Enter fullscreen mode Exit fullscreen mode

Troubleshooting

Common Issues

  1. Double instrumentation: Ensure you only call CeleryInstrumentor().instrument() once per worker process
  2. Missing broker traces: Install appropriate broker instrumentation (Redis/RabbitMQ)
  3. Worker startup issues: Initialize OpenTelemetry using worker_process_init hook, not at module level
  4. Span not appearing: Check that exporters are configured correctly and tracer provider is set
  5. High overhead: Adjust sampling rates and batch processor settings
  6. Fork-safety issues: Use worker process initialization hook to avoid sharing connections between processes

Debug Configuration

Enable debug logging to troubleshoot instrumentation:

import logging # Enable OpenTelemetry debug logging logging.getLogger("opentelemetry").setLevel(logging.DEBUG) # Enable Celery debug logging logging.getLogger("celery").setLevel(logging.DEBUG) # Configure root logger logging.basicConfig(level=logging.DEBUG)
Enter fullscreen mode Exit fullscreen mode

Performance Optimization

Configure appropriate settings for production:

from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.sdk.trace.sampling import TraceIdRatioBased # Configure sampling (sample 10% of traces) sampler = TraceIdRatioBased(0.1) provider = TracerProvider(sampler=sampler) # Optimize batch processor processor = BatchSpanProcessor( exporter, max_queue_size=2048, schedule_delay_millis=5000, max_export_batch_size=512, )
Enter fullscreen mode Exit fullscreen mode

What's next?

By integrating OpenTelemetry with Celery, you gain valuable insights into your distributed task processing pipeline. You can monitor task performance, track queue depths, identify bottlenecks, and troubleshoot issues across your asynchronous workflow.

The telemetry data collected helps you:

  • Monitor task execution times and success rates
  • Track worker performance and resource utilization
  • Identify bottlenecks in task processing pipelines
  • Debug failed tasks and retry patterns
  • Optimize queue management and worker scaling
  • Understand task dependencies and data flow
  • Improve overall system reliability and performance

Top comments (0)