# Celery

> Instrument Celery task queue applications with OpenTelemetry for distributed tracing and task monitoring

Source: https://last9.io/docs/integrations/frameworks/python/celery/

Use OpenTelemetry to instrument your Celery task queue application and send telemetry data to Last9. This integration provides automatic instrumentation for task execution, message brokers, and distributed task monitoring across workers.

Celery is a distributed task queue system that allows you to run background tasks asynchronously. With OpenTelemetry instrumentation, you can track task execution, performance, and failures across your distributed worker infrastructure.

## Prerequisites

Before setting up Celery monitoring, ensure you have:

- Python 3.7 or higher installed
- Celery application with tasks defined
- Message broker (Redis, RabbitMQ, or others) configured
- Last9 account with integration credentials
- `pip` package manager available

1.  **Install OpenTelemetry Packages**

    Install the required OpenTelemetry packages for Celery instrumentation:

    ```bash
    pip install opentelemetry-instrumentation-celery opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp
    ```

    You can also add these packages to your `requirements.txt` file:

    ```txt
    opentelemetry-instrumentation-celery
    opentelemetry-api
    opentelemetry-sdk
    opentelemetry-exporter-otlp
    ```

2.  **Set Environment Variables**

    Configure OpenTelemetry environment variables for your Celery application:

    ```bash
    export OTEL_SERVICE_NAME=<service_name>
    export OTEL_EXPORTER_OTLP_ENDPOINT=$last9_otlp_endpoint
    export OTEL_EXPORTER_OTLP_HEADERS="Authorization=$last9_otlp_auth_header"
    export OTEL_TRACES_EXPORTER=otlp
    export OTEL_TRACES_SAMPLER="always_on"
    export OTEL_RESOURCE_ATTRIBUTES="deployment.environment=production"
    export OTEL_LOG_LEVEL=error
    ```

    Replace `<service_name>` with your Celery service name (e.g., `task-processor`, `email-worker`).

3.  **Instrument Your Celery Application**

    The key to Celery tracing is proper initialization timing. Tracing and instrumentation must be initialized **after** the Celery worker process is initialized for threading components like BatchSpanProcessor to work correctly.

**Worker Instrumentation**

    Here's how to properly instrument Celery workers using the `worker_process_init` signal:

    ```python
    from opentelemetry.sdk.resources import Resource
    from opentelemetry import trace
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.export import BatchSpanProcessor
    from opentelemetry.semconv.resource import ResourceAttributes
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    from opentelemetry.instrumentation.celery import CeleryInstrumentor

    from celery import Celery
    from celery.signals import worker_process_init
    import os

    @worker_process_init.connect(weak=False)
    def init_celery_tracing(*args, **kwargs):
        """Initialize OpenTelemetry tracing for Celery worker processes."""
        CeleryInstrumentor().instrument()

        resource = Resource(attributes={
            ResourceAttributes.SERVICE_NAME: os.getenv('OTEL_SERVICE_NAME', 'celery-worker'),
            ResourceAttributes.SERVICE_VERSION: "1.0.0",
            ResourceAttributes.DEPLOYMENT_ENVIRONMENT: os.getenv('DEPLOYMENT_ENV', 'production'),
        })

        provider = TracerProvider(resource=resource)
        processor = BatchSpanProcessor(OTLPSpanExporter())
        provider.add_span_processor(processor)
        trace.set_tracer_provider(provider)

    # Configure your Celery app
    app = Celery("tasks", broker="redis://localhost:6379/0")

    # Optional: Configure result backend
    app.conf.result_backend = "redis://localhost:6379/0"

    @app.task
    def process_data(data_id):
        """Example task that processes data."""
        # Your task logic here
        print(f"Processing data ID: {data_id}")
        # Simulate some work
        import time
        time.sleep(1)
        return {"status": "completed", "data_id": data_id}

    @app.task
    def send_email(recipient, subject, body):
        """Example task for sending emails."""
        # Your email sending logic here
        print(f"Sending email to {recipient}: {subject}")
        return {"status": "sent", "recipient": recipient}
    ```

**Producer Instrumentation**

    For applications that **produce** Celery tasks (not workers), you can use simpler instrumentation:

    ```python
    from opentelemetry import configure_once
    from opentelemetry.instrumentation.celery import CeleryInstrumentor
    from celery import Celery
    import os

    # Initialize OpenTelemetry
    configure_once()
    CeleryInstrumentor().instrument()

    # Configure Celery app
    app = Celery("tasks", broker="redis://localhost:6379/0")

    # Example: Enqueue tasks
    def enqueue_tasks():
        """Example function to enqueue tasks."""
        result1 = app.send_task('tasks.process_data', args=[123])
        result2 = app.send_task('tasks.send_email',
                               args=["user@example.com", "Welcome", "Hello!"])

        return [result1.id, result2.id]
    ```

**Combined Application**

    For applications that both produce and consume tasks:

    ```python
    from opentelemetry.sdk.resources import Resource
    from opentelemetry import trace, configure_once
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.export import BatchSpanProcessor
    from opentelemetry.semconv.resource import ResourceAttributes
    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
    from opentelemetry.instrumentation.celery import CeleryInstrumentor

    from celery import Celery
    from celery.signals import worker_process_init
    import os

    # Initialize tracing for producer/web app
    configure_once()
    CeleryInstrumentor().instrument()

    @worker_process_init.connect(weak=False)
    def init_worker_tracing(*args, **kwargs):
        """Initialize tracing specifically for worker processes."""
        # Reinitialize for worker context
        resource = Resource(attributes={
            ResourceAttributes.SERVICE_NAME: os.getenv('OTEL_SERVICE_NAME', 'celery-worker'),
            ResourceAttributes.SERVICE_VERSION: "1.0.0",
            ResourceAttributes.DEPLOYMENT_ENVIRONMENT: os.getenv('DEPLOYMENT_ENV', 'production'),
        })

        provider = TracerProvider(resource=resource)
        processor = BatchSpanProcessor(OTLPSpanExporter())
        provider.add_span_processor(processor)
        trace.set_tracer_provider(provider)

    app = Celery("tasks", broker="redis://localhost:6379/0")

    @app.task
    def complex_task(param1, param2):
        """Example of a complex task with custom tracing."""
        tracer = trace.get_tracer(__name__)

        with tracer.start_as_current_span("complex_task_processing") as span:
            span.set_attribute("task.param1", param1)
            span.set_attribute("task.param2", param2)

            # Your task logic here
            result = param1 + param2
            span.set_attribute("task.result", result)

            return result
    ```

4.  **Start Your Celery Workers**

    Start your Celery workers with the instrumentation enabled:

    ```bash
    # Start a single worker
    celery -A your_app_name worker --loglevel=info

    # Start multiple workers
    celery -A your_app_name worker --loglevel=info --concurrency=4

    # Start worker with specific queue
    celery -A your_app_name worker --loglevel=info --queue=high_priority
    ```

5.  **Test Task Execution**

    Test your setup by enqueuing some tasks:

    ```python
    # In a Python shell or separate script
    from your_app_name import app

    # Enqueue tasks
    result1 = app.send_task('tasks.process_data', args=[456])
    result2 = app.send_task('tasks.send_email',
                           args=["test@example.com", "Test", "Testing tracing"])

    print(f"Task IDs: {result1.id}, {result2.id}")
    ```

## Understanding Celery Tracing

### What Gets Traced

When you use OpenTelemetry with Celery, the following operations are automatically traced:

- **Task Execution**: Start, duration, and completion of tasks
- **Task Routing**: Queue selection and message routing
- **Message Broker Operations**: Publishing and consuming messages
- **Task Retries**: Retry attempts and failure handling
- **Result Backend**: Result storage and retrieval operations

### Trace Context Propagation

OpenTelemetry automatically propagates trace context between:

- **Producers and Workers**: Tasks maintain trace context across process boundaries
- **Nested Tasks**: Child tasks inherit parent trace context
- **Chain Operations**: Task chains maintain continuous tracing
- **Callback Tasks**: Success and failure callbacks preserve context

### Custom Attributes and Spans

Add custom attributes to enhance observability:

```python
from opentelemetry import trace

@app.task
def enhanced_task(user_id, action_type):
    """Task with custom tracing attributes."""
    tracer = trace.get_tracer(__name__)

    with tracer.start_as_current_span("enhanced_task_execution") as span:
        # Add custom attributes
        span.set_attribute("user.id", user_id)
        span.set_attribute("action.type", action_type)
        span.set_attribute("task.priority", "high")

        # Your task logic
        try:
            result = perform_action(user_id, action_type)
            span.set_attribute("task.status", "success")
            span.set_attribute("task.result_count", len(result))
            return result
        except Exception as e:
            span.set_attribute("task.status", "error")
            span.set_attribute("error.message", str(e))
            raise
```

## Advanced Configuration

### Sampling Configuration

Control trace sampling for high-volume task processing:

```bash
# Production: Sample 10% of traces
export OTEL_TRACES_SAMPLER="traceidratio"
export OTEL_TRACES_SAMPLER_ARG="0.1"

# Development: Sample all traces
export OTEL_TRACES_SAMPLER="always_on"
```

### Resource Attributes

Add comprehensive service metadata:

```bash
export OTEL_RESOURCE_ATTRIBUTES="service.name=task-processor,service.version=2.1.0,deployment.environment=production,team=backend,service.instance.id=worker-001"
```

### Broker-Specific Configuration

**Redis Configuration**

```python
app = Celery("tasks", broker="redis://localhost:6379/0")
app.conf.update(
    result_backend="redis://localhost:6379/0",
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
)
```

**RabbitMQ Configuration**

```python
app = Celery("tasks", broker="amqp://guest@localhost//")
app.conf.update(
    result_backend="rpc://",
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
)
```

## Verification

1. **Check Worker Startup**

   When starting Celery workers, you should see OpenTelemetry initialization messages in the console output.

2. **Generate Task Load**

   Create some test tasks to generate telemetry data:

   ```python
   # Enqueue various types of tasks
   for i in range(10):
       app.send_task('tasks.process_data', args=[i])
       app.send_task('tasks.send_email', args=[f"user{i}@test.com", f"Subject {i}", "Body"])
   ```

3. **Monitor Worker Logs**

   Check that workers are processing tasks successfully:

   ```bash
   # Monitor worker output
   tail -f celery_worker.log

   # Or check journald if using systemd
   journalctl -u celery-worker -f
   ```

4. **Verify Traces in Last9**

   Log into your Last9 account and check that Celery traces are being received in the [Traces dashboard](https://app.last9.io/traces).

   Look for:

   - Task execution spans with timing information
   - Message broker operations
   - Task retry attempts and error traces
   - Cross-service trace propagation

## Troubleshooting

### Common Issues

**Tasks Not Being Traced**

Ensure the worker process initialization is correct:

```python
# Make sure this is called in worker processes
@worker_process_init.connect(weak=False)
def init_celery_tracing(*args, **kwargs):
    CeleryInstrumentor().instrument()
    # ... rest of initialization
```

**Missing Trace Context**

Verify environment variables are set correctly:

```bash
env | grep OTEL_
```

**Worker Process Crashes**

Check for threading issues with BatchSpanProcessor:

```bash
# View detailed worker logs
celery -A your_app_name worker --loglevel=debug
```

### Performance Considerations

- **Sampling**: Use appropriate sampling rates for high-volume task processing
- **Batch Processing**: BatchSpanProcessor is recommended for performance
- **Resource Limits**: Monitor memory usage in workers with tracing enabled
- **Queue Management**: Consider separate queues for traced vs non-traced tasks

### Best Practices

- **Service Naming**: Use descriptive names that distinguish between producers and workers
- **Environment Segregation**: Use different service names per environment
- **Task Categorization**: Add meaningful attributes to distinguish task types
- **Error Handling**: Implement proper exception handling with span status updates
- **Monitoring**: Set up alerts for task failure rates and execution times

## Need Help?

If you encounter any issues or have questions:

- Join our [Discord community](https://discord.com/channels/652153247672729619/652153247672729621) for real-time support
- Contact our support team at [support@last9.io](mailto:support@last9.io)