Flask-SocketIO + Celery (RMQ) + Gunicorn + gevent

[codeffe]

The issue is described here - https://stackoverflow.com/questions/79250986/failed-to-deliver-acknowledgement-to-rabbit-mq-flask-socket-io-gevent-celery

celery somehow fails to send any acknowledgement to RMQ after 1-2 successful attempts. This further blocks the queue and worker does not pick any new tasks.

Should I skip monkey_patching imports in tasks.py?

[miguelgrinberg]

I explained in the comments that if your intention is to use gevent only in the Flask app, but not in the Celery workers, then you should not call the gevent monkey patching function in the Celery worker.

[codeffe]

How do we avoid monkey patching celery? The idea was to be able to create a socket connection with the client and backend service and emit to the same from a background process. In order to emit/broadcast to the socket, we have used rabbitmq as a message queue while creating a socket io instance:

socketio = SocketIO(app, cors_allowed_origins="*", async_mode='gevent', transports=['websocket'], message_queue=socket_io_rmq_url)
We are using gunicorn in front of flask application using gevent as worker class and monkey patching the flask code:

gunicorn --preload --workers 10 --worker-class gevent --timeout 600 --log-level debug --access-logfile logs/gunicorn_access.log --error-logfile logs/gunicorn_error.log main:app --bind 0.0.0.0:5000 &
Monkey patching the flask app_config.py file where we are initiating Socket IO instance, Flask App & Celery instance as well:

Celery command used to run the worker via docker:

celery -A main.celery worker --without-heartbeat --without-gossip --without-mingle --loglevel=debug --logfile=logs/celery_worker.log --concurrency=1 -n "${HOSTNAME}_METRIC_REFRESH_NEXTGEN" --max-tasks-per-child=1
Code Structure:

│ Dockerfile
│ docker_entrypoint.sh
│ main.py
│ requirements.txt
│ init.py
│
├───api
│ │ auth.py
│ │ connectors.py
│ │ constants.py
│ │ tasks.py
│ │ views.py
│ │ init.py
│
│
├───config
│ │ app_config.py
│ │ exceptions.py
│ │ utils.py
│ │ init.py
│
│
├───entities
│ metric_refresh.py
│ metric_validations.py
│ sample_template.py
│ init.py
│
About the code structure:

Socketio, Flask app and Celery instances are getting created in app_config.py which is where we have added gevent monkey patching code.
These instances are further imported in views.py(flask routes and socket connect/disconnect) and tasks.py(to emit messages from background task on queue)
Flask main.py that is executed by gunicorn command:

from api.views import app, socketio
from config.app_config import celery

if name == 'main':
socketio.run(app, host="0.0.0.0", debug=False, threaded=False)

We have other services running perfectly fine without this socket io implementation where celery executes tasks and sends acknowledgement to RMQ at the start since acks_late flag is set to False. But in the case where all this configuration is used to implement socket and emitting from background task using queue, celery somehow fails to send any acknowledgement to RMQ after 1-2 successful attempts and worker get lost. This further blocks the queue and worker does not pick any new tasks.

[miguelgrinberg]

The SocketIO instance has different arguments when used under Flask or under Celery, so it is easy to provide an alternative configuration. Here is one of my examples that does this: https://github.com/miguelgrinberg/flack/blob/master/flack/__init__.py#L39-L52.

The monkey patching you don't have to do explicitly when using Gunicorn, since Gunicorn does it for you. So you can remove it from the code, so that it does not happen when running under Celery.