Up and down Celery workers on demand.
Check the example using a Docker container to up and down Celery workers.
- Python 3.6
- Celery 4.3.0 or higher
- Your Celery Application
- celery_worker_on_demand Python package
pip install celery-worker-on-demand
Create a custom Class extended from CeleryWorkerOnDemand to customize Agent, UpWorker and DownWorker classes.
from celery_worker_on_demand import CeleryWorkerOnDemand
from celery_worker_on_demand import Agent
from celery_worker_on_demand import UpWorker
from celery_worker_on_demand import DownWorker
class MyDemand(CeleryWorkerOnDemand):
# Customize Agent methods to apply new rules to up (flag_up) and down (flag_down) Celery workers
Agent = Agent
# Overide UpWorker run method with your script to up new Celery worker
UpWorker = UpWorker
# Overide DownWorker run method with your script to up new Celery worker
DownWorker = DownWorker
You need overide UpWorker and DownWorker run method with your custom script to up and down Celery worker.
from celery_worker_on_demand import CeleryWorkerOnDemand
from celery_worker_on_demand import UpWorker
from celery_worker_on_demand import DownWorker
class MyUpWorker(UpWorkerU):
def run(self):
# Use self.cwod to retrieve CeleryWorkerOnDemand instance;
# Use self.queue to retrieve QueueStatus instance with all queue information;
... Your script here
class MyDownWorker(DownWorker):
def run(self):
# Use self.cwod to retrieve CeleryWorkerOnDemand instance;
# Use self.queue to retrieve QueueStatus instance with all queue information;
... Your script here
class MyDemand(CeleryWorkerOnDemand):
UpWorker = MyUpWorker
DownWorker = MyDownWorker
Create a runnable Python file like below, to run like a service.
#!/bin/env/python
from .cwod import MyDemand
from .celery_app import celery_app
# Run Celery Worker On Demand service
MyDemand(celery_app).run()
To customize up and down flag you just need overide Agent methods flag_up
and flag_down
.
from celery_worker_on_demand import CeleryWorkerOnDemand
from celery_worker_on_demand import Agent
class MyAgent(Agent):
# When returns True, the UpWorker.run() method is executed
def flag_up(self, queue):
# queue is a QueueStatus instance
return super().flag_up(queue)
# When returns True, the DownWorker.run() method is executed
def flag_down(self, queue):
# queue is a QueueStatus instance
return super().flag_down(queue)
class MyDemand(CeleryWorkerOnDemand):
Agent = MyAgent
- celery_app (arg): Your Celery Application instance.
- queue_updater_fill_rate (kwarg): Times to update per second queue status. Default is
2
. - api_server_address (kwarg): Tuple with address and port to serve API. Default is
('', 8000)
. - api_basic_authorization (kwarg): Add basic authorization in API. Set authorization header value to enable. Use this generator.
- celery_app: Your Celery Application instance
- queue_updater_fill_rate: Times to update per second queue status.
- api_server_address: Tuple with address and port to serve API.
- api_basic_authorization: Authorization header value.
- queues: Queue dictionary - key is the queue name and value is a QueueStatus instance.
- queue_updater: Thread instance of QueueUpdater
- worker_monitor: Thread instance of WorkerMonitor
- api_server: Thread instance of APIServer
- agent: Thread instance of Agent
- connection: Cached property, returns main broker connection without heartbeat.
- channel: Cached property, returns default channel.
- queue_name (arg): Queue name.
Add new queue to observer.
Start the Celery Worker On Demand service.
Returns a JSON serializable dictionary that represents the current state of instance.
- cwod (arg): CeleryWorkerOnDemand instance.
Watch current state of application to up and down workers.
- queue (arg): QueueStatus instance.
When returns True, the UpWorker.run() method is executed.
- queue (arg): QueueStatus instance.
When returns True, the DownWorker.run() method is executed
- agent (arg): Agent instance.
- queue (arg): QueueStatus instance.
Python script to up new worker to queue.
- agent (arg): Agent instance.
- queue (arg): QueueStatus instance.
Python script to down queue's worker(s).
- name (arg): Queue name.
- size (kwarg): Current queue size, default is
0
. - workers (kwarg): List of WorkerStatus instance, default is a empty list.
- name: Queue name.
- size: Current queue size.
- workers: List of WorkerStatus instance.
- has_worker: Returns True when queue has worker.
- many_workers: Returns how many workers the queue have.
Returns a JSON serializable dictionary that represents the current state of instance.
- hostname (arg): Worker hostname.
- *: args passed to WorkerStatus constructor.
- **: kwargs passed to WorkerStatus constructor.
Returns a cached WorkerStatus instance relative to hostname.
- hostname (arg): Worker hostname.
- last_heartbeat_at (kwarg): Timestamp of last hearbeat. Default is
None
. - last_task_received_at (kwarg): Timestamp of last task received. Default is
None
. - last_task_started_at (kwarg): Timestamp of last task started. Default is
None
. - last_task_succeeded_at (kwarg): Timestamp of last task succeeded. Default is
None
.
- hostname (arg): Worker hostname.
- last_heartbeat_at (kwarg): Timestamp of last hearbeat.
- last_task_received_at (kwarg): Timestamp of last task received.
- last_task_started_at (kwarg): Timestamp of last task started.
- last_task_succeeded_at (kwarg): Timestamp of last task succeeded.
Returns a JSON serializable dictionary that represents the current state of instance.