Add usage example for Celery

I recently wanted to collect custom application metrics from a rather large Celery app. I tried several approaches with varying degrees of awkwardness, until finally settling on having a "sidecar" HTTP server process for all Celery nodes.
This works pretty well, but took me a while to get right. To spare others from wasting as much time as me, I would like to propose adding a section on getting Celery metrics set up with the Prometheus client. I wrote up the solution on StackOverflow: https://stackoverflow.com/a/75799358/2532203

If you're interested, I can properly summarise it and open a PR?

Also, if you look at the code, I needed to copy some swaths of the library because the built-in Prometheus WSGI server is hard-coded to demonise the thread; this causes trouble when running in a demonised process. Maybe this could be made optional with another keyword argument that defaults to True?

Thanks a lot for this as it is exactly what I was looking.

Should it work with Celery in Django as well?

@NicoCaldo

@NicoCaldo That's exactly my setup. It works fine because the metrics HTTP server doesn't rely on Django, so there's nothing additional to configure

I ran into this myself. While your solution looks correct, I found this to also work:

# Whatever hook you want, you can do this outside of a hook too.
@signals.celeryd_init.connect 
def name_this_what_you_want(sender=None, conf=None, **kwargs):
    registry = CollectorRegistry()
    multiprocess.MultiProcessCollector(registry)

    start_http_server(8000, registry=registry)

This way the processes in the celery pool can send back custom metrics (as in, non-host metrics) back to the one and only http_server. This avoids using a pushgateway.

This way the processes in the celery pool can send back custom metrics (as in, non-host metrics) back to the one and only http_server. This avoids using a pushgateway.

You wouldn't need a pushgateway with my solution either, but your approach uses way less code, so that definitely is an improvement!

Edit: looking back, I actually tried it this way and ran into multiprocessing issues, although I'm no longer quite sure which. Have you used this code in production, with multiple workers and multiple nodes?

This way the processes in the celery pool can send back custom metrics (as in, non-host metrics) back to the one and only http_server. This avoids using a pushgateway.

You wouldn't need a pushgateway with my solution either, but your approach uses way less code, so that definitely is an improvement!

Edit: looking back, I actually tried it this way and ran into multiprocessing issues, although I'm no longer quite sure which. Have you used this code in production, with multiple workers and multiple nodes?

Not quite extensively, though so far I haven't ran into issues. You could be right, I will write back if I ever run into any.

Reporting back for those who are interested, the code I provided has not caused any issues so far. Metric collection works fine. 32 concurrent cores writing to prometheus through my setup have proven to not crash. I should note that we do multithreading within a worker, it works.

@Radiergummi

@Radiergummi I highly recommend adding an example that shows your methodology and mine. It would have saved me a day of headaches. Feel free to @ me.

I ran into this myself. While your solution looks correct, I found this to also work:
# Whatever hook you want, you can do this outside of a hook too.
@signals.celeryd_init.connect 
def name_this_what_you_want(sender=None, conf=None, **kwargs):
    registry = CollectorRegistry()
    multiprocess.MultiProcessCollector(registry)

    start_http_server(8000, registry=registry)
This way the processes in the celery pool can send back custom metrics (as in, non-host metrics) back to the one and only http_server. This avoids using a pushgateway.

Error occurred for me when Celery received a new task and start to process, not sure why

objc[7678]: +[NSCharacterSet initialize] may have been in progress in another thread when fork() was called.
objc[7678]: +[NSCharacterSet initialize] may have been in progress in another thread when fork() was called. We cannot safely call it or ignore it in the fork() child process. Crashing instead. Set a breakpoint on objc_initializeAfterForkError to debug.
[2024-02-22 16:02:08,420: ERROR/MainProcess] Process 'ForkPoolWorker-8' pid:7678 exited with 'signal 6 (SIGABRT)'
[2024-02-22 16:02:08,434: ERROR/MainProcess] Task handler raised error: WorkerLostError('Worker exited prematurely: signal 6 (SIGABRT) Job: 0.')
Traceback (most recent call last):
  File "/Users/folyd/UTA/chatbot/chatbot/.venv/lib/python3.11/site-packages/billiard/pool.py", line 1264, in mark_as_worker_lost
    raise WorkerLostError(
billiard.einfo.ExceptionWithTraceback: 
"""
Traceback (most recent call last):
  File "/Users/folyd/UTA/chatbot/chatbot/.venv/lib/python3.11/site-packages/billiard/pool.py", line 1264, in mark_as_worker_lost
    raise WorkerLostError(
billiard.exceptions.WorkerLostError: Worker exited prematurely: signal 6 (SIGABRT) Job: 0.

@Folyd

@Folyd That looks unrelated to the metrics code, at least on first glance. Have you tried my original code from StackOverflow, too? The fork issue vaguely looks similar to my issue back then.

@Radiergummi

Sorry for reviving a very old thread but this thread and your stackoverflow answer are the only examples I could find covering this issue @Radiergummi I have one question, the documentation mentions some limitations when setting celery to multiprocess mode:

To handle this the client library can be put in multiprocess mode. This comes with a number of limitations:

Registries can not be used as normal, all instantiated metrics are exported
    Registering metrics to a registry later used by a MultiProcessCollector may cause duplicate metrics to be exported
Custom collectors do not work (e.g. cpu and memory metrics)
Info and Enum metrics do not work
The pushgateway cannot be used
Gauges cannot use the pid label
Exemplars are not supported
Remove and Clear of labels are currently not supported in multiprocess mode.

Have you run into this issues yourself with your code ? Thanks again for your code!

@deathwebo

Have you run into this issues yourself with your code ? Thanks again for your code!

@deathwebo it's running smoothly in production since then, and we didn't encounter any obvious issues (yet). The limitations stated in the documentation sound like a lot, but didn't actually impact our use case. You'll have to evaluate this for your application yourself. Assuming you register metrics to the default registry, don't try to measure system resource usage in multiple threads (just use node exporter for that), don't create fancy custom metrics, and use gauges instead of info metrics, you should be fine.

All those limitations really stem from the fact that you want multiple applications to contribute to a single counter. If you can picture that in your head and account for the consequences, you're golden.

Let me know if you hit any specific roadblocks!

@deathwebo

Have you run into this issues yourself with your code ? Thanks again for your code!

@deathwebo it's running smoothly in production since then, and we didn't encounter any obvious issues (yet). The limitations stated in the documentation sound like a lot, but didn't actually impact our use case. You'll have to evaluate this for your application yourself. Assuming you register metrics to the default registry, don't try to measure system resource usage in multiple threads (just use node exporter for that), don't create fancy custom metrics, and use gauges instead of info metrics, you should be fine.

All those limitations really stem from the fact that you want multiple applications to contribute to a single counter. If you can picture that in your head and account for the consequences, you're golden.

Let me know if you hit any specific roadblocks!

Nice! I just set it up for prod where we run multiple nodes and workers and it's all starting to flow into prometheus. Thanks again for this thread and information.

Hello,
I have noticed issues with this setup in celery, esp. when using in a threaded worker with multiple threads (not sure if the cause is due to multiple threads or the threading mechanism).

Notably, some metrics fail to show up, even though nothing wrong happens when measuring them. This seem to especially happen to Histogram metrics. Other strangeness include some metrics disappearing from a pod when

When debugging this with cursor, it suggested passing the registry to the metrics, i.e.

import os
from celery import signals
from prometheus_client import CollectorRegistry, multiprocess, start_http_server

REGISTRY = CollectorRegistry()

def start_celery_exporter(**kwargs):
    multiprocess.MultiProcessCollector(REGISTRY)
    port = int(os.getenv("CELERY_METRICS_EXPORTER_PORT", 4321))
    start_http_server(port=port, registry=REGISTRY)

test_histogram_registered = Histogram(
    "test_histogram_registered", "test Histogram",
    labelnames=["source",],
    buckets=(1, 5, 10, 20, INF),
   # ADD REGISTRY HERE
    registry=REGISTRY,
)

And , well, things seem to work 🎉 !!

Is this something you guys did as well ?

( it also came to me that in case of a threaded worker, one could just use a regular, non-multithread collector)

Edit: the doc in fact explicitly recommends to NOT do that:

The application must initialize a new CollectorRegistry, and store the multi-process collector inside. It is a best practice to create this registry inside the context of a request to avoid metrics registering themselves to a collector used by a MultiProcessCollector. If a registry with metrics registered is used by a MultiProcessCollector duplicate metrics may be exported, one for multiprocess, and one for the process serving the request.

Although this seems to apply mostly to "standard" metrics, that would be provided by a library, and not metrics set by the application, which is what I am interested in.

However, it seems that registry=None is used sometimes in the tests, is there a point to do that ?

We implemented the solution described in the stackoverflow for our celery workers. The metrics are registered only if there is not labels configured. I'm just curious if anyone has run into such problem?

This code works

from tasks.celery_app import app

debug_counter = Counter("debug_task_total", "Count of debug tasks run")

@app.task
def emit_debug_metric():
    debug_counter.inc()

This code does not work, the metrics do not show up. No *.db created in the PROMETHEUS_MULTIPROC_DIR . And no error either.

from tasks.celery_app import app

debug_counter = Counter("debug_task_total", "Count of debug tasks run", ["env"])

@app.task
def emit_debug_metric():
    debug_counter.labels(env='local').inc()

We implemented the solution described in the stackoverflow for our celery workers. The metrics are registered only if there is not labels configured. I'm just curious if anyone has run into such problem?

This code works
from tasks.celery_app import app

debug_counter = Counter("debug_task_total", "Count of debug tasks run")

@app.task
def emit_debug_metric():
    debug_counter.inc()
This code does not work, the metrics do not show up. No *.db created in the PROMETHEUS_MULTIPROC_DIR . And no error either.
from tasks.celery_app import app

debug_counter = Counter("debug_task_total", "Count of debug tasks run", ["env"])

@app.task
def emit_debug_metric():
    debug_counter.labels(env='local').inc()

I ended up implementing the metrics in our tasks this way

from prometheus_client import CollectorRegistry, multiprocess, start_http_server
from celery.signals import after_setup_logger, celeryd_init

@celeryd_init.connect
def start_metrics_server(_sender=None, _conf=None, **_kwargs):  # type: ignore
    registry = CollectorRegistry()
    multiprocess.MultiProcessCollector(registry)

    start_http_server(9090, registry=registry)

Hope it helps you! Metrics are looking great in our dashboards

@deathwebo

We implemented the solution described in the stackoverflow for our celery workers. The metrics are registered only if there is not labels configured. I'm just curious if anyone has run into such problem?
This code works
from tasks.celery_app import app

debug_counter = Counter("debug_task_total", "Count of debug tasks run")

@app.task
def emit_debug_metric():
    debug_counter.inc()
This code does not work, the metrics do not show up. No *.db created in the PROMETHEUS_MULTIPROC_DIR . And no error either.
from tasks.celery_app import app

debug_counter = Counter("debug_task_total", "Count of debug tasks run", ["env"])

@app.task
def emit_debug_metric():
    debug_counter.labels(env='local').inc()
I ended up implementing the metrics in our tasks this way

from prometheus_client import CollectorRegistry, multiprocess, start_http_server
from celery.signals import after_setup_logger, celeryd_init

@celeryd_init.connect
def start_metrics_server(_sender=None, _conf=None, **_kwargs): # type: ignore
registry = CollectorRegistry()
multiprocess.MultiProcessCollector(registry)
start_http_server(9090, registry=registry)
Hope it helps you! Metrics are looking great in our dashboards

Much appreciated! @deathwebo
Indeed, I could start registering the metrics with your example and my debug metric was also recorded successfully.
Though, it's still not showing up if I start adding the labels for some odd reasons.

If you don't mind, just wanted to confirm that you're able to add labels in your metrics ok?

@deathwebo

We implemented the solution described in the stackoverflow for our celery workers. The metrics are registered only if there is not labels configured. I'm just curious if anyone has run into such problem?
This code works
from tasks.celery_app import app

debug_counter = Counter("debug_task_total", "Count of debug tasks run")

@app.task
def emit_debug_metric():
    debug_counter.inc()
This code does not work, the metrics do not show up. No *.db created in the PROMETHEUS_MULTIPROC_DIR . And no error either.
from tasks.celery_app import app

debug_counter = Counter("debug_task_total", "Count of debug tasks run", ["env"])

@app.task
def emit_debug_metric():
    debug_counter.labels(env='local').inc()
I ended up implementing the metrics in our tasks this way

from prometheus_client import CollectorRegistry, multiprocess, start_http_server
from celery.signals import after_setup_logger, celeryd_init

@celeryd_init.connect
def start_metrics_server(_sender=None, _conf=None, **_kwargs): # type: ignore
registry = CollectorRegistry()
multiprocess.MultiProcessCollector(registry)
start_http_server(9090, registry=registry)
Hope it helps you! Metrics are looking great in our dashboards
Much appreciated! @deathwebo
Indeed, I could start registering the metrics with your example and my debug metric was also recorded successfully.
Though, it's still not showing up if I start adding the labels for some odd reasons.

If you don't mind, just wanted to confirm that you're able to add labels in your metrics ok?

Yeah we use labels in our metrics all the time.

Ok I found the reason for my problem (cf #902 (comment)).

See #1123 for more info

We were declaring a historgram without labelnames . This apparently raises an exception, but because of how we load the modules, for some reason the exception didn't show up.

Here's the stacktrace when the exception is not swallowed:

worker-threads-1  |     self._sum = values.ValueClass(self._type, self._name, self._name + '_sum', self._labelnames, self._labelvalues, self._documentation)
worker-prefork-1  |   File "/app/celery_experiments_app.py", line 62, in <module>
worker-threads-1  |                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
worker-prefork-1  |     test_histogram_no_label = Histogram(
worker-threads-1  |   File "/usr/local/lib/python3.11/site-packages/prometheus_client/values.py", line 68, in __init__
worker-prefork-1  |                               ^^^^^^^^^^
worker-threads-1  |     self.__reset()
worker-prefork-1  |   File "/usr/local/lib/python3.11/site-packages/prometheus_client/metrics.py", line 577, in __init__
worker-threads-1  |   File "/usr/local/lib/python3.11/site-packages/prometheus_client/values.py", line 82, in __reset
worker-prefork-1  |     super().__init__(
worker-threads-1  |     files[file_prefix] = MmapedDict(filename)
worker-prefork-1  |   File "/usr/local/lib/python3.11/site-packages/prometheus_client/metrics.py", line 127, in __init__
worker-threads-1  |                          ^^^^^^^^^^^^^^^^^^^^
worker-prefork-1  |     self._metric_init()
worker-threads-1  |   File "/usr/local/lib/python3.11/site-packages/prometheus_client/mmap_dict.py", line 64, in __init__
worker-prefork-1  |   File "/usr/local/lib/python3.11/site-packages/prometheus_client/metrics.py", line 605, in _metric_init
worker-threads-1  |     self._f = open(filename, 'rb' if read_mode else 'a+b')
worker-prefork-1  |     self._sum = values.ValueClass(self._type, self._name, self._name + '_sum', self._labelnames, self._labelvalues, self._documentation)
worker-threads-1  |               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
worker-prefork-1  |                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
worker-threads-1  | FileNotFoundError: [Errno 2] No such file or directory: '/tmp/prom/histogram_1.db'
worker-prefork-1  |   File "/usr/local/lib/python3.11/site-packages/prometheus_client/values.py", line 68, in __init__
worker-threads-1  | 
worker-prefork-1  |     self.__reset()
worker-prefork-1  |   File "/usr/local/lib/python3.11/site-packages/prometheus_client/values.py", line 82, in __reset
worker-prefork-1  |     files[file_prefix] = MmapedDict(filename)
worker-prefork-1  |                          ^^^^^^^^^^^^^^^^^^^^
worker-prefork-1  |   File "/usr/local/lib/python3.11/site-packages/prometheus_client/mmap_dict.py", line 64, in __init__
worker-prefork-1  |     self._f = open(filename, 'rb' if read_mode else 'a+b')
worker-prefork-1  |               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
worker-prefork-1  | FileNotFoundError: [Errno 2] No such file or directory: '/tmp/prom/histogram_1.db'

Edit: realized this stacktrace is because of my particular setup, where the multiproc dir is not present when the app starts. The problem here is that a metric is written at load time, and this somehow can cause issues.

I wrote an exporter specifically for Celery. It tracks task success, failure, duration and queue length.

https://github.com/danihodovic/celery-exporter

I wrote an exporter specifically for Celery. It tracks task success, failure, duration and queue length.

https://github.com/danihodovic/celery-exporter

That solves a bit of a different use case, though. I wanted to be able to expose custom application level metrics from my Python code, not just task queue related metrics.

That said, if you don't need anything other than that, your exporter looks like a fine solution to me.

Opened PR #1168 with a Celery sidecar exporter example + notes on multiprocess mode.

Navigation Menu

Repository navigation

Add usage example for Celery #902

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Add usage example for Celery #902

Description

Activity

NicoCaldo commented on May 16, 2023

Radiergummi commented on May 16, 2023

Andrew-Cha commented on Aug 16, 2023

Radiergummi commented on Aug 17, 2023

Andrew-Cha commented on Aug 21, 2023

Andrew-Cha commented on Oct 26, 2023

Andrew-Cha commented on Oct 26, 2023

Folyd commented on Feb 23, 2024

Radiergummi commented on Mar 10, 2024

deathwebo commented on Oct 4, 2024

Radiergummi commented on Oct 5, 2024

deathwebo commented on Oct 7, 2024

gg-mmill commented on Jun 3, 2025

kayleevo9x commented on Jun 23, 2025

deathwebo commented on Jun 23, 2025

kayleevo9x commented on Jun 23, 2025

deathwebo commented on Jun 23, 2025

gg-mmill commented on Jun 27, 2025

danihodovic commented on Sep 3, 2025

Radiergummi commented on Sep 3, 2025

ayushanand293 commented on Apr 23, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions