Monitoring Celery | An Actionable Guide

Have you ever experienced a web/mobile application accepting your request to upload videos or images immediately but performing the actual video or image upload processing work behind the scenes? If so, you’ve seen an application that works based on the task queue concept.

Task queues aid in the registration of time-consuming tasks and execute them in another compute layer, thereby freeing up the user-facing HTTP request and response lifecycle. In addition to freeing up the application server to handle other incoming requests, this kind of approach also enhances the user experience. You don’t need to wait around for your long-running task to get completed. After the request registration in the queue, you will be immediately given an acknowledgment with a task ID for tracking so you can focus on other things.

Once the task processing is complete, you’ll be updated on the task status in one of the following ways:

• By making a call to an API exposed for such a tracking purpose
• By checking the report/dashboard page designed for that purpose
• Via a notification service (if you have an active notification subscription in the system)

Celery, “a task queue implementation for Python web applications used to asynchronously execute work outside the HTTP request-response cycle,” was designed and created to facilitate the above characteristics in an application. It is especially useful for serving use cases that deal with long-running processes.

In this article, you will learn more about Celery by exploring how it works, how and why it should be monitored, and how Cronitor can help you ensure that your scheduled and background tasks in Celery are properly executed.

The sample Flask app from this article is in this GitHub repository.

Components of a Celery-Based Tech Stack

Any tech stack involving Celery has the following key components:

• Task producer: your application that, on receiving a manual or system-generated event-based request or one based on a schedule, registers the task in the task queue
• Task queue: the message broker, which can also act as a storage layer
• Task scheduler: handles the scheduled jobs via celery beat, the recurring job default scheduler in the Celery ecosystem
• Task consumer: consumes the registered tasks and executes the actual processing logic, also known as celery workers
• Backend: stores the result of the task

For more information, refer to Celery’s documentation.

Why Do You Need to Monitor Celery?

The key components that make up a Celery-based tech stack must operate in harmony to meet the goal of processing an assigned task. However, things can go wrong in production for any of these components, leading to process failure. For instance, common issues might involve any of the following scenarios:

• The queue occupancy rate is faster than usual (anomaly situation).
• The same task gets processed in multiple workers against the expectation (duplicate task-processing scenario).
• Celery workers may cause latency in the processing front, leading to slow progress in the task queue.

Monitoring Celery helps you keep an eye out for issues like these and more. You’ll be able to easily identify things like the following:

• How many workers are online, and how many are offline?
• How many active tasks are handled by a specific worker?
• How many scheduled tasks are involved?
• How many tasks failed in processing?
• How many tasks got registered in the queue?
• How much is the latency or elapsed time for a task?
• What is the configured frequency of a scheduled job/task?
• What details are available about system performance?

How to Monitor Celery

Let’s get started and take a closer look at monitoring Celery. In this section, you will learn how to do so using the following methods:

• A command-line utility of Celery
• Flower, a GUI-based monitoring tool
• Cronitor

Monitoring Celery Tasks Using Command-Line Utilities

This section will provide some example source codes of the celery command-line utility.

Execute the following commands in the machine’s/container’s terminal where Celery is installed or where the Celery worker is running. All the given commands are applicable to monitor regular tasks and scheduled tasks.

To List All Active Tasks Processed by Workers

celery inspect active

To List All Registered Tasks Processed by Workers

celery inspect registered

To List All Active Queues in Celery

celery inspect active_queues

To Monitor System Performance

celery inspect stats

To Monitor the Scheduled Job Frequency

celery inspect conf

Output snippet:

->
    {
        "beat_schedule": {
            "scheduled_delayed_greetings": {
                "schedule": "<crontab: * * * * * (m/h/d/dM/MY)>",
                "task": "celery_beat_tasks.tasks.scheduled_delayed_greetings"
            }
        },

Here, the schedule variable holds the task frequency. In this example, the task is scheduled to execute for every minute using celery beat.

To List Workers and Their Availability Status

celery status

This command will list all workers and tell you how many are online.

Monitoring Celery Tasks Using Flower

Flower is a GUI-based tool, and it comes in quite handy to surf through web pages to monitor various aspects of Celery. Using the Flower Dashboard, you can monitor the following:

• The number of workers and their status
• The number of active tasks, failed tasks, and successful tasks
• The workload of the worker instance

With the Flower Tasks view, you can monitor the following:

• The tasks assigned to each worker
• The task status
• Latency in the queue for a particular task (the time difference between “Received” and “Started” is the latency for that task to be picked up from the queue)
• The elapsed time (“Runtime” field) of each task by the worker
• Arguments, if any are supplied as part of the task processing

Flower also provides relevant system usage statistics on the “System” tab:

If you have scheduled a celery beat task, Flower can help you view the job/task frequency on the “Config” tab:

Flower is a useful view of the workers. It does not send an alert when a beat task skips its schedule. That is the gap the next section fills with Cronitor.

Monitoring Celery with Cronitor

Cronitor watches scheduled jobs and HTTP endpoints, and it alerts when a run is missed, fails, or runs too long. For Celery, prefer the Python SDK. Its Celery integration creates job monitors and sends a ping when a task starts, finishes, or fails. The same steps are in the app at Get Started, then Jobs, then Celery.

Sign up, then copy an SDK Integration key from Settings, then API. A Telemetry key can send pings, and a ping for a new monitor key creates that monitor (Telemetry API), but it cannot call the Monitor API, so celery beat cannot register schedules. Do not commit the key.

Install

pip install cronitor

Install Celery the same way: pip install celery. The integration needs Celery 4.0 or newer and uses Celery message protocol version 2.

Register the Celery app

Set CRONITOR_API_KEY before the process starts. initialize also takes api_key= if the key is already in memory. You can still assign cronitor.api_key.

import os
import cronitor.celery
from celery import Celery

celery = Celery(
    __name__,
    backend=os.getenv("CELERY_RESULT_BACKEND"),
    broker=os.getenv("CELERY_BROKER_URL"),
)
celery.conf.imports = ("tasks",)
celery.conf.timezone = "UTC"
celery.conf.beat_schedule = {
    "scheduled_delayed_greetings": {
        "task": "tasks.scheduled_delayed_greetings",
        "schedule": 20,  # seconds
    }
}

cronitor.celery.initialize(celery)

On celery beat startup, the integration creates a job monitor for each periodic task and sends a schedule string taken from that beat entry. An interval such as 20 is sent as every 20 seconds. Open the job afterward and confirm the expression. Other tasks are monitored too: the monitor key is the task name, and Cronitor opens a monitor on the first run if one does not exist yet.

cronitor.celery.initialize(celery, celerybeat_only=True)

celerybeat_only=True ignores tasks that are not on the beat schedule.

A few limits, from the package readme and the integration itself:

• Solar schedules are skipped.
• django-celery-beat is not supported.
• With Celery's default PersistentScheduler, the integration copies celerybeat-schedule to a temp file and restarts beat on that copy so it can attach headers. A custom schedule path is no longer used.

There is a shorter version of this setup in Creating Cron Jobs in Python.

Example tasks

Add these to tasks.py, next to the Celery app above. celery.conf.imports = ("tasks",) loads that module, and the Flask route imports get_celery_stats from it. scheduled_delayed_greetings is the beat task. get_celery_stats is an ordinary task the route calls. Both use @celery.task. Only the beat entry makes a task periodic.

import time

@celery.task()
def scheduled_delayed_greetings():
    print("Inside scheduled delayed greetings")
    time.sleep(20)
    print("Completed scheduled delayed greetings")


@celery.task()
def get_celery_stats():
    print("Inside post_celery_stats")
    inspect_output = celery.control.inspect()
    print("Stats ", inspect_output.stats(), flush=True)
    return inspect_output.stats()

inspect().stats() is the same data celery inspect stats prints. One worker's payload looks like this:

{'celery@b1903b37c511': {'total': {'tasks.get_celery_stats': 13, 'tasks.scheduled_delayed_greetings': 21}, 'pid': 1, 'clock': '700', 'uptime': 479, 'pool': {'max-concurrency': 1, 'processes': [8], 'max-tasks-per-child': 'N/A', 'put-guarded-by-semaphore': False, 'timeouts': [0, 0], 'writes': {'total': 34, 'avg': '1.00', 'all': '1.00', 'raw': '34', 'strategy': 'fair', 'inqueues': {'total': 1, 'active': 0}}}, 'broker': {'hostname': 'redis', 'userid': None, 'virtual_host': '0', 'port': 6379, 'insist': False, 'ssl': False, 'transport': 'redis', 'connect_timeout': 4, 'transport_options': {}, 'login_method': None, 'uri_prefix': None, 'heartbeat': 120.0, 'failover_strategy': 'round-robin', 'alternates': []}, 'prefetch_count': 4, 'rusage': {'utime': 0.9243969999999999, 'stime': 0.16507, 'maxrss': 44788, 'ixrss': 0, 'idrss': 0, 'isrss': 0, 'minflt': 17220, 'majflt': 13, 'nswap': 0, 'inblock': 0, 'oublock': 688, 'msgsnd': 0, 'msgrcv': 0, 'nsignals': 0, 'nvcsw': 1877, 'nivcsw': 25}}}

The Flask route blocks on that task and returns the JSON:

from flask import Flask, jsonify
from tasks import get_celery_stats

app = Flask(__name__)

@app.route("/celery/stats")
def get_celery_stats_route():
    response = get_celery_stats.delay().get()
    return jsonify(response), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True, use_reloader=True)

Check the stats URL

If that route is on a public host, a Check monitor can request it. Monitor.put is the Python call. The same form is in the app: open Checks, then Create Check. Request Details is the URL, method, interval, timeout, and regions. Assertions are further down the page.

import cronitor

cronitor.Monitor.put([{
    "type": "check",
    "key": "celery-stats",
    "request": {
        "url": "https://example.com/celery/stats",
        "regions": ["us-east-1", "eu-central-1", "ap-south-1"],
    },
    "assertions": [
        "response.code = 200",
        "response.time < 1s",
        "response.body contains tasks.get_celery_stats",
    ],
}])

Those regions are US East (N. Virginia), EU Central (Frankfurt), and AP South (Mumbai). The list is in uptime monitoring. An assertion is the healthy condition. If the body does not contain tasks.get_celery_stats, or the status is not 200, the check fails.

Do not point a Check at a Celery broker, Flower, or a stats route that is only reachable on a private network. Exposing it just so Cronitor can poll it is the wrong trade. Monitor private endpoints keeps the request inside your network.

What you see after a few runs

Open Jobs. Each card shows Duration, Expected, Last Event, and Last Issue. Edit on a job is where you confirm the schedule beat copied over. The Basics is the name, an optional group, and tags. Job Details is the schedule, platform, server timezone, and metric assertions.

Open a job for Status, Success, Performance, Executions, Alerts, the execution chart, and Latest Activity.

Alert Settings, further down the same edit form, is Notify, a note, reminders, callbacks, grace period, and tolerance. If you do not pick a list, Cronitor uses the account's default notification list. The list in this shot is named Standard Alert.

A failed job can email you, and a later success can email you again. The messages below are the failure and recovery mails:

Latest Activity on a Check is the same idea as the job page: each request, the response, and whether the assertions passed. Open the check from Checks.

Conclusion

celery inspect and Flower tell you what the workers are doing right now. They do not page you when a beat task goes quiet. cronitor.celery.initialize is the Celery path, and a Check covers an HTTP stats URL you are willing to expose. Sign up if you do not have an account yet.