Monitoring Celery in Production
Tags: python, django, celery
Published: 2024-01-22 13:19:21

How can you determine if your Celery queues are healthy or overwhelmed with tasks?  

Anything from sudden surges of user traffic to abuse of your app can drag down your Celery cluster.  Knowing when this is happening is half the battle.

Our goal in this series is to build automated monitoring tools for Celery so that you don't have to find out about problems from your users.

## Monitoring Queue Wait Times

There are many valuable metrics to monitor such as queue length, processing time, failed task count, etc.

But we're going to start with what I think is the <b>most effective metric to monitor: <i>queue wait times.</i></b>

Queue wait time is simply how long a task spends waiting in the queue before a worker accepts it and begins processing.

Queue length is important, but harder to judge.  What constitutes too many tasks in the queue?  For one service, having 10 items in the queue might be too many, while 100,000 is perfectly normal for another service.

It's also easy to have short bursts where the queue grows, but as long as the workers process them quickly enough, is it really an issue?  This can lead to false alarms.

Queue wait times, on the other hand, are a true reflection of your user's experience.  

Let's say you have a Django app that auto-tunes videos of people singing.  You also implemented a helpful progress bar.  A user uploads a video of themselves singing a Beyonce song, but the progress bar doesn't budge for several minutes.  That user gives up, and tells all their friends that the app doesn't work.

What happened?  Tasks may have still been processing, but the <b>queue wait time</b> was too high.

### Setting Up the Example Django App

Let's build a very simple Django app to demonstrate how to monitor Celery queue wait times.  

We'll use Docker to make it simpler to set up Celery and Redis.

Create the Django project.

```bash
django-admin startproject djautotune
cd djautotune
python3 manage.py startapp core
```

Update the `djautotune/settings.py` file to add the `core` application.  We'll also add the Celery related settings while we're editing the file.

```python
INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "core",
]

REDIS_HOST = os.environ["REDIS_HOST"]
REDIS_PORT = os.environ["REDIS_PORT"]

CELERY_METRICS_TTL = int(os.environ.get("CELERY_METRICS_TTL", 300))

CELERY_IMPORTS = ("core.tasks",)
CELERY_BROKER_URL = os.environ["CELERY_BROKER_URL"]
CELERY_RESULT_BACKEND = os.environ["CELERY_RESULT_BACKEND"]
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
```

### Adding the Celery Worker Entrypoint
Create a new file named `celery.py` in the same directory as the `settings.py` file.  This file acts as the entrypoint for the Celery worker.

```python
import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "djautotune.settings")

app = Celery("djautotune")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()
```

### Implementing the Background Task Logic
Now create the `core/tasks.py` file to contain our task logic.

Since actually auto-tuning videos is beyond the scope of this tutorial, we'll add some dummy code that simulates the workload.

```python
import random
import time

from celery import shared_task

@shared_task
def autotune():
    short_duration = random.uniform(0.5, 2)  # Short duration for most tasks
    medium_duration = random.uniform(2, 5)  # Medium duration for 10% of tasks
    long_duration = random.uniform(5, 10)  # Long duration for 1% of tasks (P99)

    probability = random.random()

    if probability < 0.90:
        time_to_sleep = short_duration
    elif probability < 0.99:
        time_to_sleep = medium_duration
    else:
        time_to_sleep = long_duration

    time.sleep(time_to_sleep)
```

### Creating a View to Invoke the Background Task
We'll need a way to invoke this new `autotune` Celery task.  Open `core/views.py` to add the following HTTP endpoint.

```python
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from .tasks import autotune

@csrf_exempt
def enqueue_autotune_task(request):
    if request.method == "POST":
        result = autotune.delay()
        return JsonResponse(
            {"task_id": result.task_id, "status": "Task submitted successfully!"},
            status=200,
        )
    else:
        return JsonResponse({"error": "Only POST method is allowed."}, status=405)
```

Now, as usual, you'll need to add the new route to your Django router config.

To start, add the following in the `core/urls.py` file.

```python
from django.urls import path
from . import views

urlpatterns = [
    path("autotune/", views.enqueue_autotune_task, name="enqueue_autotune_task"),
]
```

Finally, update the main `urls.py` file to include the routes from the core app.

```python
from django.contrib import admin
from django.urls import include, path

urlpatterns = [
    path("admin/", admin.site.urls),
    path("", include("core.urls")),
]
```

### Setting Up the Docker Development Environment
Create the `requirements.txt` file.

```bash
Django==4.2.4
redis==4.5.5
celery==5.2.7
```

Create the `Dockerfile` for the Django app in the root project directory.

```bash
FROM python:3.11

ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

WORKDIR /usr/src/app

COPY requirements.txt /usr/src/app/

RUN pip install --no-cache-dir -r requirements.txt

COPY . /usr/src/app/

RUN mkdir -p /usr/src/app/
```

Now create the `docker-compose.yml` to manage the containers.

```bash
version: '3.8'

services:
  web:
    build:
      context: .
      dockerfile: Dockerfile
    command: python manage.py runserver 0.0.0.0:8000
    volumes:
      - ${PWD}:/usr/src/app
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
      - CELERY_RESULT_BACKEND=redis://redis:6379/0
      - CELERY_METRICS_TTL=300
      - REDIS_HOST=redis
      - REDIS_PORT=6379
    ports:
      - "8000:8000"
    depends_on:
      - redis

  celery:
    build:
      context: .
      dockerfile: Dockerfile
    command: celery -A djautotune worker --loglevel=info
    volumes:
      - ${PWD}:/usr/src/app
    environment:
      - CELERY_BROKER_URL=redis://redis:6379/0
      - CELERY_RESULT_BACKEND=redis://redis:6379/0
      - CELERY_METRICS_TTL=300
      - REDIS_HOST=redis
      - REDIS_PORT=6379
    depends_on:
      - redis

  redis:
    image: redis:latest
    ports:
      - "6379:6379"
```

### Starting Up and Testing the App
You should be ready to start the app now with the `docker-compose up` command.

If all goes well, you can enqueue a task with the following curl command.

```bash
curl -X POST http://localhost:8000/autotune/
```

Using the `docker logs` command on the Celery container, you should be able to see a success message like this for the task.

```bash
Task core.tasks.autotune[9cdfcb7d-1f65-49cc-aa33-20c694157e61] succeeded in 1.9952244879677892s: None
```

## Adding the Celery Monitoring Logic

We have a functioning background task, but no monitoring to speak of.

To add monitoring, we'll create a Django view that returns the longest queue wait time in the past 30 seconds.  The window of time can be easily adjusted, but it's important that we use a sliding window of some length.

We'll use a <a href="https://redis.io/docs/data-types/sorted-sets/" target="_blank">Redis sorted set</a> with an expiration to create the sliding window of queue wait times.

### Marking New Tasks with a Start Timestamp

The first step is to mark each new task with a timestamp.  

This timestamp represents the moment the task was created and published, but before any worker has picked up the task.

Open `core/apps.py` and add the following.

```python
import time
from django.apps import AppConfig
from celery.signals import before_task_publish

def before_task_publish_handler(sender=None, headers=None, body=None, **kwargs):
    headers["__metadata__"] = {"publish_timestamp": time.time()}

class CoreConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "core"

    def ready(self):
        before_task_publish.connect(before_task_publish_handler)
```

### Recording Queue Wait Time in Redis

Once a worker has picked up the task, we need to calculate the elapsed time and store it in Redis.

Since we're using the `task_prerun` signal, this represents the moment the task was picked up, and does not include actual processing time.

Open `djautotune/celery.py` and add the following.

```python
import time
import os
import redis

from celery import Celery
from celery.signals import task_prerun
from django.conf import settings

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "djautotune.settings")

app = Celery("djautotune")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

redis_client = redis.Redis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=0)

@task_prerun.connect
def task_prerun_handler(task_id, task, *args, **kwargs):
    metadata = getattr(task.request, "__metadata__", {})
    publish_timestamp = metadata.get("publish_timestamp")

    if publish_timestamp:
        elapsed_time = time.time() - publish_timestamp

        redis_client.zadd("task_elapsed_times", {task_id: elapsed_time})
        redis_client.expire("task_elapsed_times", settings.CELERY_METRICS_TTL, nx=True)
```

The `zadd` call adds the queue wait time to the sorted set.  The set is sorted by the queue wait time, so it's efficient to request the greatest value, which will be important when we add a view to ask for this value.

Because the `expire` call is passed `nx=True`, only the initial call sets the expiration.  In other words, even though `expire` is called for each task, the expiration is not reset.  Eventually, the key `task_elapsed_times` will expire, and an entirely new sorted set is created with a new expiration.

### Requesting the Longest Queue Wait Time

For the final piece of the puzzle, let's create a view to return the longest queue wait in the current time window (as defined by the `CELERY_METRICS_TTL` setting).

Open `core/views.py` and add the following view code.

```python
import redis

from django.conf import settings
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from .tasks import autotune

def max_queue_wait_time(request):
    redis_client = redis.Redis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=0)

    max_time_data = redis_client.zrevrange("task_elapsed_times", 0, 0, withscores=True)
    max_elapsed_time = max_time_data[0][1] if max_time_data else 0

    return JsonResponse({"max_elapsed_time": max_elapsed_time})
```

## Setting Up Alerting for Celery

We'll use <a href="https://cronitor.io" target="_blank">Cronitor</a> to set up alerting so that we receive a notification when queue wait times become too high.

To do this, Cronitor will make requests to our `max_queue_wait_time` endpoint and parse the JSON response.  

This means that Cronitor must have an endpoint that it can reach.  Normally, we can't do that when developing on a personal machine.  For this tutorial, however, we can use <a href="https://ngrok.com/" target="_blank">ngrok</a> to establish a tunnel to our local Django application for testing purposes.

### Using Ngrok to Establish a Tunnel

Head over to <a href="https://ngrok.com/" target="_blank">ngrok</a> and sign up for a free account.

Download the CLI tool, enter your token, and open a tunnel with the following command.

```python
ngrok http 8000
```

Copy the domain that is listed next to the `Forwarding` column.

### Adding the Celery Check in Cronitor

Sign up for a free account with the <a href="https://cronitor.io" target="_blank">Cronitor</a> service.

Add a new Check with details similar to the following.  Remember to add your own Ngrok domain.

Take note of the `ngrok-skip-browser-warning` header and be sure to add that.

<img src="https://circumeo.io/static/images/cronitor.png" style="width: 100%"/>

Under the assertions section, set up the following so that Cronitor understands when to trigger an alert.

<img src="https://circumeo.io/static/images/assertion.png" style="width: 100%"/>

This means that Cronitor will fire an alert when `max_elapsed_time` exceeds 10 seconds.  You can of course adjust this time to whatever metric suits your application.

Here's an example of what happened when I added a huge number of tasks so that my Celery queue became overwhelmed.

<img src="https://circumeo.io/static/images/assertion_failed.png" style="width: 100%"/>

## Further Celery Monitoring Enhancements

You've learned how to monitor <b>queue wait time</b>, a metric that directly reflects what your users are experiencing.

This is a great start, but there's a lot more to monitoring your service comprehensively.

As a sneak peek, in the next article we'll learn how to monitor more Celery stats, such as failed tasks and retries.  We'll also dive into building a custom Celery monitoring dashboard using Bokeh.

You can see the <a href="https://github.com/zchtodd/djautotune">entire djautotune sample project on GitHub</a> if you'd like to copy some of the code as a template.