---
title: "Instrument Queues"
description: "Learn how to manually instrument your code to use Sentry's Queues module. "
url: https://docs.sentry.io/platforms/python/tracing/instrumentation/custom-instrumentation/queues-module/
---

# Instrument Queues | Sentry for Python

Sentry comes with a [queue-monitoring dashboard](https://sentry.io/orgredirect/organizations/:orgslug/dashboards/) that can be auto-instrumented for popular Python queue setups (like [Celery](https://docs.sentry.io/platforms/python/integrations/celery.md)).

In case yours isn't supported, you can still instrument custom spans and transactions around your queue producers and consumers to ensure that you have performance data about your messaging queues.

This page covers both transaction mode (default) and stream mode. See [Streamed Spans](https://docs.sentry.io/platforms/python/tracing/streamed-spans.md) to learn more.

## [Producer Instrumentation](https://docs.sentry.io/platforms/python/tracing/instrumentation/custom-instrumentation/queues-module.md#producer-instrumentation)

To start capturing performance metrics, use the `start_span()` function to wrap your queue producer events. Your span `op` must be set to `queue.publish`. Include the following span data to enrich your producer spans with queue metrics:

| Data Attribute                | Type   | Description                       |
| ----------------------------- | ------ | --------------------------------- |
| `messaging.message.id`        | string | The message identifier            |
| `messaging.destination.name`  | string | The queue or topic name           |
| `messaging.message.body.size` | int    | Size of the message body in bytes |

In transaction mode, your `queue.publish` span must exist inside a transaction in order to be recognized as a producer span. If you are using a [supported web framework](https://docs.sentry.io/platforms/python/integrations.md#web-frameworks), the transaction is created by the integration. If you use plain Python, you can start a new one using `sentry_sdk.start_transaction()`.

In stream mode, there's no separate transaction to depend on. If your `queue.process` span has no parent, it's automatically promoted to a service span. If you'd rather group it under an explicit service span, start one with `sentry_sdk.traces.start_span(parent_span=None)` first.

You must also include trace headers (`sentry-trace` and `baggage`) in your message so that your consumers can continue your trace once your message is picked up.

```python
from datetime import datetime, timezone

import sentry_sdk
import my_custom_queue

# Initialize Sentry
sentry_sdk.init(...)

connection = my_custom_queue.connect()

# The message you want to send to the queue
queue = "messages"
message = "Hello World!"
message_id = "abc123"

# Create transaction
# If you are using a web framework, the framework integration
# will create this for you and you can omit this.
with sentry_sdk.start_transaction(
    op="function",
    name="queue_producer_transaction",
):
    # Create the span
    with sentry_sdk.start_span(
        op="queue.publish",
        name="queue_producer",
    ) as span:
        # Set span data
        span.set_data("messaging.message.id", message_id)
        span.set_data("messaging.destination.name", queue)
        span.set_data("messaging.message.body.size", len(message.encode("utf-8")))

        # Publish the message to the queue (including trace information and current time stamp)
        now = int(datetime.now(timezone.utc).timestamp())
        connection.publish(
            queue=queue,
            body=message,
            timestamp=now,
            headers={
                "sentry-trace": sentry_sdk.get_traceparent(),
                "baggage": sentry_sdk.get_baggage(),
            },
        )
```

*Other available variations of the above snippet: Stream Mode*

## [Consumer Instrumentation](https://docs.sentry.io/platforms/python/tracing/instrumentation/custom-instrumentation/queues-module.md#consumer-instrumentation)

To start capturing performance metrics, use the `start_span()` function to wrap your queue consumers. Your span `op` must be set to `queue.process`. Include the following span data to enrich your consumer spans with queue metrics:

| Data Attribute                      | Type   | Description                                                         |
| ----------------------------------- | ------ | ------------------------------------------------------------------- |
| `messaging.message.id`              | string | The message identifier                                              |
| `messaging.destination.name`        | string | The queue or topic name                                             |
| `messaging.message.body.size`       | number | Size of the message body in bytes                                   |
| `messaging.message.retry.count`     | number | The number of times a message was attempted to be processed         |
| `messaging.message.receive.latency` | number | The time in milliseconds that a message awaited processing in queue |

In transaction mode, your `queue.process` span must exist inside a transaction in order to be recognized as a consumer span. If you are using a [supported web framework](https://docs.sentry.io/platforms/python/integrations.md#web-frameworks), the transaction is created by the integration. If you use plain Python, you can start a new one using `sentry_sdk.start_transaction()`.

In stream mode, there's no separate transaction to depend on. If your `queue.process` span has no parent, it's automatically promoted to a service span. If you'd rather group it under an explicit service span, start one with `sentry_sdk.traces.start_span(parent_span=None)` first.

Use `continue_trace()` to connect your consumer spans to their associated producer spans. To mark the trace of your message as success or failed, use `transaction.set_status()` in transaction mode, or set `span.status = "ok"/"error"` directly on the service span in stream mode.

```python
from datetime import datetime, timezone

import sentry_sdk
import my_custom_queue

# Initialize Sentry
sentry_sdk.init(...)

connection = my_custom_queue.connect()

# Pick up message from queues
queue = "messages"
message = connection.consume(queue=queue)

# Calculate latency (optional, but valuable)
now = datetime.now(timezone.utc)
message_time = datetime.fromtimestamp(message["timestamp"], timezone.utc)
latency = now - message_time

# Continue the trace started in the producer
# If you are using a web framework, the framework integration
# will create this for you and you can omit this.
transaction = sentry_sdk.continue_trace(
    message["headers"],
    op="function",
    name="queue_consumer_transaction",
)
with sentry_sdk.start_transaction(transaction):
    # Create the span
    with sentry_sdk.start_span(
        op="queue.process",
        name="queue_consumer",
    ) as span:
        # Set span data
        span.set_data("messaging.message.id", message["message_id"])
        span.set_data("messaging.destination.name", queue)
        span.set_data("messaging.message.body.size", len(message["body"].encode("utf-8")))
        span.set_data("messaging.message.receive.latency", latency.total_seconds() * 1000)
        span.set_data("messaging.message.retry.count", 0)

        try:
            # Process the message
            process_message(message)
        except Exception:
            # In case of an error set the status to "internal_error"
            transaction.set_status("internal_error")
```

*Other available variations of the above snippet: Stream Mode*