Issue starting worker with PrometheusMiddleware #173
Description
Hi, thanks for the awesome library. I am trying to add the PrometheusMiddleware to an application, however, I am seeing some strange behavior when starting the worker(s).
taskiq version: 0.8.2 (although I observed the same behavior on 0.8.1)
prometheus-client version: 0.17.1 (although I observed the same behavior on 0.16.0)
I have defined my (redis) broker as follows:
broker = (
ListQueueBroker(url=REDIS_URL, queue_name="nebula_tasks")
.with_result_backend(redis_async_result)
.with_id_generator(task_id_generator)
.with_middlewares(
PrometheusMiddleware(server_addr="0.0.0.0", server_port=9000),
)
)When I attempt to start my worker with default worker count I see the following trace (debug log enabled):
pipenv run taskiq worker --fs-discover --workers 2 --log-level DEBUG src.nebula.broker:broker
https://gist.github.com/nickdichev-firework/5b420be1d014f8fe925dcb63b771f5bd
When I attempt to start the worker with only one worker I see the following trace (debug log enabled):
pipenv run taskiq worker --fs-discover --workers 1 --log-level DEBUG src.nebula.broker:broker
https://gist.github.com/nickdichev-firework/36adf01a3ba11e22bf2c17766f61282b
When I remove the .with_middlewares call with the broker I am able to start the worker (and process tasks) with two workers. However, what is interesting is that when I reduce it to one worker I see the same output as one worker with prometheus middleware enabled: FileNotFoundError
Any ideas? I am relatively new to python so kind of scratching my head at what else could be useful to debug, please let me know if you need any additional info! Thanks in advance!
Activity
nickdichev-firework commentedon Jul 14, 2023
@s3rius thanks for the fix, I was debugging the middleware a bit today and I noticed that the
startupcallback wasn't being executed.I read the prometheus client documentation's section on multiprocessing and I noticed a couple of things. I implemented some changes which made the metrics start working with two workers. I still see the issue with one worker, not sure what that's about 🤷♂️
I opened #176 with my changes. Please take a look and let me know what you think! Again, I am not much of a python dev so lmk if anything looks off. Thanks again for the great work here, I learned a lot reading this codebase.
s3rius commentedon Jul 14, 2023
@nickdichev-firework Did you update you taskiq?
s3rius commentedon Jul 14, 2023
Yesterday I made changes. Today I published a new release. Please try version https://github.com/taskiq-python/taskiq/releases/tag/0.8.3.
The problem with registries is that it shows only taskiq's metrics. If user wants to add new metrics, they won't be shown by taskiq.
nickdichev-firework commentedon Jul 14, 2023
@s3rius yes I still get the error on 0.8.3 -- the only way I've been able to resolve the
Duplicated timeseries in CollectorRegistry: {'found_errors_total', 'found_errors', 'found_errors_created'}error is with my forkdanielphan2003 commentedon May 7, 2026
I'm still experiencing this issue on 0.12.3 with 4 workers. Here is my logs:
s3rius commentedon May 8, 2026
Huh. That's interesting. Seems like this is a regression.
@danielphan2003, can you please share you setup so I can reproduce it? Could it be that you manually create some metrics that are used by taskiq?
waza-ari commentedon Oct 13, 2026
We hit
Duplicated timeseriesagain on 0.13.0 and tracked down the cause in our setup. It is also the root cause of #397, so sharing it here in case others run into it.Cause:
--fs-discoverimports the broker module a second time under a different module name.import_tasksderives module names from the file paths relative to the current working directory (str(path.with_name(prefix)).replace(os.path.sep, ".")). With asrc/layout andPYTHONPATH=src, runningfrom the project root imports
src/worker/briefings/task_x.pyassrc.worker.briefings.task_x. Itsfrom ..broker import brokerthen resolves tosrc.worker.broker, which is a second copy ofworker.broker. Executing that module again constructs a secondPrometheusMiddleware, and its__init__registersfound_errors,received_tasks, ... on the globalREGISTRYa second time:Without the middleware this goes unnoticed: you silently end up with two broker instances, each with all tasks registered (
sys.modulescontains bothworker.brokerandsrc.worker.broker), and it keeps working because tasks are also looked up in the global registry.Fix on our side: drop
--fs-discover(ourworker/__init__.pyalready imports every task module), or run the worker from a directory where the discovered module names match the real import path. #682 (making the middleware import-safe) would avoid the crash as well.Note that the traceback in the previous comment (raised from
startup()→MultiProcessCollector(REGISTRY)) is a different problem that #620 already fixed: since thenstartup()uses a dedicatedCollectorRegistry.One more thing we noticed with
--workers > 1:__init__doesif not metrics_path.exists(): metrics_path.mkdir(parents=True)withoutexist_ok=True. Worker processes starting at the same time race on it, and the loser dies withFileExistsErroruntil the process manager restarts it. Pre-creating the directory works around it;mkdir(parents=True, exist_ok=True)would fix it in the middleware.