I’m trying a FastAPI based API with celery, redis, and rabitMQ as the background tasks.
when doing docker-compose up
, the redis, rabbit, and flower parts work, I’m able to access the flower dashboard.
but it then gets stuck in the celery part.
the error:
rabbitmq_1 | 2020-09-08 06:32:38.552 [info] <0.716.0> connection <0.716.0> (172.22.0.6:49290 -> 172.22.0.2:5672): user 'user' authenticated and granted access to vhost '/'
celery-flower_1 | [W 200908 06:32:41 control:44] 'stats' inspect method failed
celery-flower_1 | [W 200908 06:32:41 control:44] 'active_queues' inspect method failed
celery-flower_1 | [W 200908 06:32:41 control:44] 'registered' inspect method failed
celery-flower_1 | [W 200908 06:32:41 control:44] 'scheduled' inspect method failed
celery-flower_1 | [W 200908 06:32:41 control:44] 'active' inspect method failed
celery-flower_1 | [W 200908 06:32:41 control:44] 'reserved' inspect method failed
celery-flower_1 | [W 200908 06:32:41 control:44] 'revoked' inspect method failed
celery-flower_1 | [W 200908 06:32:41 control:44] 'conf' inspect method failed
My docker-compose file:
version: "3.7"
services:
rabbitmq:
image: "bitnami/rabbitmq:3.7"
ports:
- "4000:4000"
- "5672:5672"
volumes:
- "rabbitmq_data:/bitnami"
redis:
image: "bitnami/redis:5.0.4"
environment:
- REDIS_PASSWORD=password123
ports:
- "5000:5000"
volumes:
- "redis_data:/bitnami/redis/data"
celery-flower:
image: gregsi/latest-celery-flower-docker:latest
environment:
- AMQP_USERNAME=user
- AMQP_PASSWORD=bitnami
- AMQP_ADMIN_USERNAME=user
- AMQP_ADMIN_PASSWORD=bitnami
- AMQP_HOST=rabbitmq
- AMQP_PORT=5672
- AMQP_ADMIN_HOST=rabbitmq
- AMQP_ADMIN_PORT=15672
- FLOWER_BASIC_AUTH=user:test
ports:
- "5555:5555"
depends_on:
- rabbitmq
- redis
fastapi:
build: .
ports:
- "8000:8000"
depends_on:
- rabbitmq
- redis
volumes:
- "./:/app"
command: "poetry run uvicorn app/app/main:app --bind 0.0.0.0:8000"
worker:
build: .
depends_on:
- rabbitmq
- redis
volumes:
- "./:/app"
command: "poetry run celery worker -A app.app.worker.celery_worker -l info -Q test-queue -c 1"
volumes:
rabbitmq_data:
driver: local
redis_data:
driver: local
My celery app:
celery_app = Celery(
"worker",
backend="redis://:password123@redis:6379/0",
broker="amqp://user:bitnami@rabbitmq:5672//"
)
celery_app.conf.task_routes = {
"app.app.worker.celery_worker.compute_stock_indicators": "stocks-queue"
}
celery_app.conf.update(task_track_started=True)
celery worker:
@celery_app.task(acks_late=True)
def compute_stock_indicators(stocks: list, background_task):
stocks_with_indicators = {}
for stock in stocks:
current_task.update_state(state=Actions.STARTED,
meta={f"starting to fetch {stock}'s indicators"})
stock_indicators = fetch_stock_indicators(stock) # Fetch the stock most recent indicators
current_task.update_state(state=Actions.FINISHED,
meta={f"{stock}'s indicators fetched"})
stocks_with_indicators.update({stock: stock_indicators})
current_task.update_state(state=Actions.PROGRESS,
meta={f"predicting {stocks}s..."})
The Fast API function:
log = logging.getLogger(__name__)
rabbit = RabbitMQHandler(host='localhost', port=5672, level="DEBUG")
log.addHandler(rabbit)
def celery_on_message(body):
"""
Logs the initiation of the function
"""
log.warning(body)
def background_on_message(task):
"""
logs the function when it is added to queue
"""
log.warning(task.get(on_message=celery_on_message, propagate=False))
app = FastAPI(debug=True)
@app.post("/")
async def initiator(stocks: FrozenSet, background_task: BackgroundTasks, ):
"""
:param stocks: stocks to be analyzed
:type stocks: set
:param background_task: initiate the tasks queue
:type background_task: starlette.background.BackgroundTasks
"""
log.warning(msg=f'beginning analysis on: {stocks}')
task_name = "app.app.worker.celery_worker.compute_stock_indicators"
task = celery_app.send_task(task_name, args=[stocks, background_task])
background_task.add_task(background_on_message, task)
return {"message": "Stocks indicators successfully calculated,stocks sent to prediction"}
2
Answers
On the
docker-compose
, on theworker
section, thecommand
reads:So essentially you are asking the worker to "watch" a queue named
test-queue
.But on the
celery_app
, on the following section:you are defining a queue named
stocks-queue
.Either change the
docker-compose
‘s or thecelery_app
‘s queue name to match the other.if you use Docker Toolbox on windows , so you should add port 5555 to VM virtualBOX network:
frist run following command on cmd:
then open VM virtualBOX , go to Settings >Networks > advanced>port forwarding >add a row with port 5555 and leave name field
click OK and on cmd, run following command: