diff --git a/django_q/admin.py b/django_q/admin.py index ae1d97e..0ae5d01 100644 --- a/django_q/admin.py +++ b/django_q/admin.py @@ -13,7 +13,7 @@ from django_q.tasks import async_task class TaskAdmin(admin.ModelAdmin): """model admin for success tasks.""" - list_display = ("name", "group", "func", "started", "stopped", "time_taken") + list_display = ("name", "group", "func", "cluster", "started", "stopped", "time_taken") def has_add_permission(self, request): """Don't allow adds.""" @@ -26,7 +26,7 @@ class TaskAdmin(admin.ModelAdmin): search_fields = ("name", "func", "group") readonly_fields = [] - list_filter = ("group",) + list_filter = ("group", "cluster") def get_readonly_fields(self, request, obj=None): """Set all fields readonly.""" @@ -36,7 +36,8 @@ class TaskAdmin(admin.ModelAdmin): def retry_failed(FailAdmin, request, queryset): """Submit selected tasks back to the queue.""" for task in queryset: - async_task(task.func, *task.args or (), hook=task.hook, **task.kwargs or {}) + async_task(task.func, *task.args or (), hook=task.hook, + group=task.group, cluster=task.cluster, **task.kwargs or {}) task.delete() @@ -46,7 +47,7 @@ retry_failed.short_description = _("Resubmit selected tasks to queue") class FailAdmin(admin.ModelAdmin): """model admin for failed tasks.""" - list_display = ("name", "group", "func", "started", "stopped", "short_result") + list_display = ("name", "group", "func", "cluster", "started", "stopped", "short_result") def has_add_permission(self, request): """Don't allow adds.""" @@ -54,7 +55,7 @@ class FailAdmin(admin.ModelAdmin): actions = [retry_failed] search_fields = ("name", "func", "group") - list_filter = ("group",) + list_filter = ("group", "cluster") readonly_fields = [] def get_readonly_fields(self, request, obj=None): @@ -123,6 +124,8 @@ class QueueAdmin(admin.ModelAdmin): """queue admin for ORM broker""" list_display = ("id", "key", "name", "group", "func", "lock", "task_id") + fields = ("key", "lock", "task_id", "name", "group", "func", "args", "kwargs", "q_options") + readonly_fields = fields[2:] def save_model(self, request, obj, form, change): obj.save(using=Conf.ORM) diff --git a/django_q/brokers/__init__.py b/django_q/brokers/__init__.py index 7c132d2..fb091ba 100644 --- a/django_q/brokers/__init__.py +++ b/django_q/brokers/__init__.py @@ -7,7 +7,9 @@ from django_q.conf import Conf class Broker: - def __init__(self, list_key: str = Conf.PREFIX): + def __init__(self, list_key: str = None): + # With same BROKER_CLASS, `list_key` is just a synonym for `queue_name` except for RedisBroker + list_key = list_key or Conf.CLUSTER_NAME self.connection = self.get_connection(list_key) self.list_key = list_key self.cache = self.get_cache() @@ -151,7 +153,7 @@ class Broker: return None @staticmethod - def get_connection(list_key: str = Conf.PREFIX): + def get_connection(list_key: str = None): """ Gets a connection to the broker :param list_key: Optional queue name @@ -160,13 +162,14 @@ class Broker: return 0 -def get_broker(list_key: str = Conf.PREFIX) -> Broker: +def get_broker(list_key: str = None) -> Broker: """ Gets the configured broker type :param list_key: optional queue name :type list_key: str :return: a broker instance """ + list_key = list_key or Conf.CLUSTER_NAME # custom if Conf.BROKER_CLASS: module, func = Conf.BROKER_CLASS.rsplit(".", 1) diff --git a/django_q/brokers/aws_sqs.py b/django_q/brokers/aws_sqs.py index c6f69d9..6c0ba50 100644 --- a/django_q/brokers/aws_sqs.py +++ b/django_q/brokers/aws_sqs.py @@ -8,7 +8,7 @@ QUEUE_DOES_NOT_EXIST = "AWS.SimpleQueueService.NonExistentQueue" class Sqs(Broker): - def __init__(self, list_key: str = Conf.PREFIX): + def __init__(self, list_key: str = None): self.sqs = None super(Sqs, self).__init__(list_key) self.queue = self.get_queue() @@ -77,7 +77,7 @@ class Sqs(Broker): return "AWS SQS" @staticmethod - def get_connection(list_key: str = Conf.PREFIX) -> Session: + def get_connection(list_key: str = None) -> Session: config = Conf.SQS if "aws_region" in config: config["region_name"] = config["aws_region"] diff --git a/django_q/brokers/ironmq.py b/django_q/brokers/ironmq.py index fb148d6..0482107 100644 --- a/django_q/brokers/ironmq.py +++ b/django_q/brokers/ironmq.py @@ -46,6 +46,7 @@ class IronMQBroker(Broker): return self.delete(task_id) @staticmethod - def get_connection(list_key: str = Conf.PREFIX) -> Queue: + def get_connection(list_key: str = None) -> Queue: + list_key = list_key or Conf.CLUSTER_NAME ironmq = IronMQ(name=None, **Conf.IRON_MQ) return ironmq.queue(queue_name=list_key) diff --git a/django_q/brokers/mongo.py b/django_q/brokers/mongo.py index ef42ac2..70ea864 100644 --- a/django_q/brokers/mongo.py +++ b/django_q/brokers/mongo.py @@ -15,7 +15,7 @@ def _timeout(): class Mongo(Broker): - def __init__(self, list_key=Conf.PREFIX): + def __init__(self, list_key: str = None): super(Mongo, self).__init__(list_key) self.collection = self.get_collection() @@ -24,7 +24,7 @@ class Mongo(Broker): self.collection = self.get_collection() @staticmethod - def get_connection(list_key: str = Conf.PREFIX) -> MongoClient: + def get_connection(list_key: str = None) -> MongoClient: return MongoClient(**Conf.MONGO) def get_collection(self): diff --git a/django_q/brokers/orm.py b/django_q/brokers/orm.py index 3c11261..9fcaf02 100644 --- a/django_q/brokers/orm.py +++ b/django_q/brokers/orm.py @@ -16,7 +16,7 @@ def _timeout(): class ORM(Broker): @staticmethod - def get_connection(list_key: str = Conf.PREFIX): + def get_connection(list_key: str = None): if transaction.get_autocommit( using=Conf.ORM ): # Only True when not in an atomic block @@ -55,8 +55,9 @@ class ORM(Broker): self.delete(task_id) def enqueue(self, task): + # list_key might be null (e.g. in a test setup) but OrmQ.key has not-null constraint package = self.get_connection().create( - key=self.list_key, payload=task, lock=timezone.now() + key=self.list_key or Conf.CLUSTER_NAME, payload=task, lock=timezone.now() ) return package.pk diff --git a/django_q/brokers/redis_broker.py b/django_q/brokers/redis_broker.py index eafda41..02abd5b 100644 --- a/django_q/brokers/redis_broker.py +++ b/django_q/brokers/redis_broker.py @@ -11,7 +11,8 @@ except ImportError: class Redis(Broker): - def __init__(self, list_key: str = Conf.PREFIX): + def __init__(self, list_key: str = None): + list_key = list_key or Conf.CLUSTER_NAME super(Redis, self).__init__(list_key=f"django_q:{list_key}:q") def enqueue(self, task): @@ -57,7 +58,7 @@ class Redis(Broker): return self.connection.mget(keys) @staticmethod - def get_connection(list_key: str = Conf.PREFIX) -> Redis: + def get_connection(list_key: str = None) -> Redis: if django_redis and Conf.DJANGO_REDIS: return django_redis.get_redis_connection(Conf.DJANGO_REDIS) if isinstance(Conf.REDIS, str): diff --git a/django_q/cluster.py b/django_q/cluster.py index 13d7a2f..99243d0 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -48,14 +48,16 @@ from .utils import get_func_repr, localtime class Cluster: def __init__(self, broker: Broker = None): - self.broker = broker or get_broker() + # Cluster do not need an init or default broker except for testing, + # The sentinel will create a broker for cluster and utilize ALT_CLUSTERS config in Conf. + self.broker = broker # DON'T USE get_broker() to set a default broker here. self.sentinel = None self.stop_event = None self.start_event = None self.pid = current_process().pid self.cluster_id = uuid.uuid4() self.host = socket.gethostname() - self.timeout = Conf.TIMEOUT + self.timeout = None signal.signal(signal.SIGTERM, self.sig_handler) signal.signal(signal.SIGINT, self.sig_handler) @@ -141,7 +143,7 @@ class Sentinel: start_event, cluster_id, broker=None, - timeout=Conf.TIMEOUT, + timeout=None, start=True, ): # Make sure we catch signals for the pool @@ -158,7 +160,7 @@ class Sentinel: self.start_event = start_event self.pool_size = Conf.WORKERS self.pool = [] - self.timeout = timeout + self.timeout = timeout or Conf.TIMEOUT self.task_queue = ( Queue(maxsize=Conf.QUEUE_LIMIT) if Conf.QUEUE_LIMIT else Queue() ) @@ -169,6 +171,10 @@ class Sentinel: if start: self.start() + def queue_name(self): + # multi-queue: cluster name is (broker's) queue_name + return self.broker.list_key if self.broker else '--' + def start(self): self.broker.ping() self.spawn_cluster() @@ -287,14 +293,14 @@ class Sentinel: _("%(name)s guarding cluster %(cluster_name)s") % { "name": current_process().name, - "cluster_name": humanize(self.cluster_id.hex), + "cluster_name": humanize(self.cluster_id.hex) + f" [{self.queue_name()}]", } ) self.start_event.set() Stat(self).save() logger.info( _("Q Cluster %(cluster_name)s running.") - % {"cluster_name": humanize(self.cluster_id.hex)} + % {"cluster_name": humanize(self.cluster_id.hex) + f" [{self.queue_name()}]"} ) counter = 0 cycle = Conf.GUARD_CYCLE # guard loop sleep in seconds @@ -401,6 +407,7 @@ def pusher(task_queue: Queue, event: Event, broker: Broker = None): logger.exception("Failed to push task to queue") broker.fail(ack_id) continue + task["cluster"] = Conf.CLUSTER_NAME # save actual cluster name to orm task table task["ack_id"] = ack_id task_queue.put(task) logger.debug( @@ -518,6 +525,7 @@ def worker( pre_execute.send(sender="django_q", func=f, task=task) # execute the payload timer.value = timer_value # Busy + try: if f is None: # raise a meaningfull error if task["func"] is not a valid function @@ -614,6 +622,7 @@ def save_task(task, broker: Broker): hook=task.get("hook"), args=task["args"], kwargs=task["kwargs"], + cluster=task.get("cluster"), started=task["started"], stopped=task["stopped"], result=task["result"], @@ -685,13 +694,16 @@ def scheduler(broker: Broker = None): broker = get_broker() close_old_django_connections() try: + # Only default cluster will handler schedule with default(null) cluster + Q_default = db.models.Q(cluster__isnull=True) if Conf.CLUSTER_NAME == Conf.PREFIX else db.models.Q(pk__in=[]) + with db.transaction.atomic(using=db.router.db_for_write(Schedule)): for s in ( Schedule.objects.select_for_update() .exclude(repeats=0) .filter(next_run__lt=timezone.now()) .filter( - db.models.Q(cluster__isnull=True) | db.models.Q(cluster=Conf.PREFIX) + Q_default | db.models.Q(cluster=Conf.CLUSTER_NAME) ) ): args = () @@ -733,14 +745,11 @@ def scheduler(broker: Broker = None): s.next_run = next_run s.repeats += -1 - # send it to the cluster - scheduled_broker = broker - try: - scheduled_broker = get_broker(q_options["broker_name"]) - except: # noqa: E722 - # invalid broker_name or non existing broker with broker_name - pass - q_options["broker"] = scheduled_broker + # send it to the cluster; any cluster name is allowed in multi-queue scenarios + # because `broker_name` is confusing, using `cluster` name is recommended and takes precedence + q_options["cluster"] = s.cluster or q_options.get("cluster", q_options.pop("broker_name", None)) + if q_options['cluster'] is None or q_options['cluster'] == Conf.CLUSTER_NAME: + q_options["broker"] = broker q_options["group"] = q_options.get("group", s.name or s.id) kwargs["q_options"] = q_options s.task = django_q.tasks.async_task(s.func, *args, **kwargs) diff --git a/django_q/conf.py b/django_q/conf.py index fb556ba..a9877c7 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -39,10 +39,22 @@ class Conf: """ try: - conf = settings.Q_CLUSTER + conf = settings.Q_CLUSTER.copy() except AttributeError: conf = {} + _Q_CLUSTER_NAME = os.getenv("Q_CLUSTER_NAME") + if _Q_CLUSTER_NAME and _Q_CLUSTER_NAME != conf.get("name") and \ + _Q_CLUSTER_NAME != conf.get("cluster_name"): + conf["cluster_name"] = _Q_CLUSTER_NAME + alt_conf = conf.pop("ALT_CLUSTERS") + if isinstance(alt_conf, dict): + alt_conf = alt_conf.get(_Q_CLUSTER_NAME) + if isinstance(alt_conf, dict): + alt_conf.pop('name', None) + alt_conf.pop('cluster_name', None) + conf.update(alt_conf) + # Redis server configuration . Follows standard redis keywords REDIS = conf.get("redis", {}) @@ -70,8 +82,14 @@ class Conf: MONGO_DB = conf.get("mongo_db", None) # Name of the cluster or site. For when you run multiple sites on one redis server + # It's also the `salt` for signing OrmQ, and part of the Redis stats caching key + # For all clusters in one site, PREFIX should be the same value to be able to decrypt payloads PREFIX = conf.get("name", "default") + # Support alternative cluster name to use multiple queues in one site. + # cluster name and queue name are interchangeable, same thing. + CLUSTER_NAME = conf.get("cluster_name", PREFIX) + # Log output level LOG_LEVEL = conf.get("log_level", "INFO") diff --git a/django_q/management/commands/qcluster.py b/django_q/management/commands/qcluster.py index 1972bdc..2330c01 100644 --- a/django_q/management/commands/qcluster.py +++ b/django_q/management/commands/qcluster.py @@ -2,6 +2,7 @@ from django.core.management.base import BaseCommand from django.utils.translation import gettext as _ from django_q.cluster import Cluster +import os class Command(BaseCommand): @@ -16,8 +17,21 @@ class Command(BaseCommand): default=False, help="Run once and then stop.", ) + parser.add_argument( + "-n", + "--name", + dest="cluster_name", + default=None, + help="Set alternative cluster name instead of the name in Q_CLUSTER settings (for multi-queue setup). " + "On Linux you should set name through `Q_CLUSTER_NAME=cluster_name python manage.py qcluster` instead." + ) def handle(self, *args, **options): + # Set alternative cluster_name before creating the cluster (cluster_name is broker's queue_name, too) + cluster_name = options.get("cluster_name") + if cluster_name: + os.environ["Q_CLUSTER_NAME"] = cluster_name + q = Cluster() q.start() if options.get("run_once", False): diff --git a/django_q/migrations/0017_task_cluster_alter.py b/django_q/migrations/0017_task_cluster_alter.py new file mode 100644 index 0000000..a37219c --- /dev/null +++ b/django_q/migrations/0017_task_cluster_alter.py @@ -0,0 +1,43 @@ +# Generated by Django 4.1.5 on 2023-03-07 12:18 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("django_q", "0016_schedule_intended_date_kwarg"), + ] + + operations = [ + migrations.AddField( + model_name="task", + name="cluster", + field=models.CharField(blank=True, default=None, max_length=100, null=True), + ), + migrations.AlterField( + model_name="ormq", + name="key", + field=models.CharField( + help_text="Name of the target cluster", max_length=100 + ), + ), + migrations.AlterField( + model_name="ormq", + name="lock", + field=models.DateTimeField( + help_text="Prevent any cluster from pulling until", null=True + ), + ), + migrations.AlterField( + model_name="schedule", + name="cluster", + field=models.CharField( + blank=True, + default=None, + help_text="Name of the target cluster", + max_length=100, + null=True, + ), + ), + ] diff --git a/django_q/models.py b/django_q/models.py index 8bf7f5c..e754776 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -11,6 +11,7 @@ from django.utils import timezone from django.utils.timezone import is_aware from django.utils.html import format_html from django.utils.translation import gettext_lazy as _ +from django.utils.functional import cached_property # External from picklefield import PickledObjectField @@ -33,6 +34,7 @@ class Task(models.Model): kwargs = PickledObjectField(null=True, protocol=-1) result = PickledObjectField(null=True, protocol=-1) group = models.CharField(max_length=100, editable=False, null=True) + cluster = models.CharField(max_length=100, default=None, null=True, blank=True) started = models.DateTimeField(editable=False) stopped = models.DateTimeField(editable=False) success = models.BooleanField(default=True, editable=False) @@ -215,7 +217,10 @@ class Schedule(models.Model): help_text=_("Cron expression"), ) task = models.CharField(max_length=100, null=True, editable=False) - cluster = models.CharField(max_length=100, default=None, null=True, blank=True) + cluster = models.CharField( + max_length=100, default=None, null=True, blank=True, + help_text=_("Name of the target cluster") + ) intended_date_kwarg = models.CharField( max_length=100, null=True, @@ -302,24 +307,38 @@ class Schedule(models.Model): class OrmQ(models.Model): - key = models.CharField(max_length=100) + key = models.CharField(max_length=100, help_text=_("Name of the target cluster")) payload = models.TextField() - lock = models.DateTimeField(null=True) + lock = models.DateTimeField(null=True, help_text=_("Prevent any cluster from pulling until")) + @cached_property def task(self): - return SignedPackage.loads(self.payload) + try: + return SignedPackage.loads(self.payload) + except Exception as e: + return {"id": "*" + e.__class__.__name__} def func(self): - return get_func_repr(self.task()["func"]) + return get_func_repr(self.task.get("func")) def task_id(self): - return self.task()["id"] + return self.task.get("id") def name(self): - return self.task()["name"] + return self.task.get("name") def group(self): - return self.task().get("group") + return self.task.get("group") + + def args(self): + return self.task.get("args") + + def kwargs(self): + return self.task.get("kwargs") + + def q_options(self): + exclude = {"id", "name", "group", "func", "args", "kwargs"} + return {k: v for k, v in self.task.items() if k not in exclude} class Meta: app_label = "django_q" diff --git a/django_q/tasks.py b/django_q/tasks.py index 5f91ba5..5ffdd94 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -31,6 +31,7 @@ def async_task(func, *args, **kwargs): "iter_cached", "chain", "broker", + "cluster", "timeout", ) q_options = keywords.pop("q_options", {}) @@ -52,7 +53,7 @@ def async_task(func, *args, **kwargs): elif key in keywords: task[key] = keywords.pop(key) # don't serialize the broker - broker = task.pop("broker", get_broker()) + broker = task.pop("broker", None) or get_broker(task.get("cluster")) # overrides if "cached" not in task and Conf.CACHED: task["cached"] = Conf.CACHED @@ -71,7 +72,7 @@ def async_task(func, *args, **kwargs): return _sync(pack) # push it enqueue_id = broker.enqueue(pack) - logger.info(f"Enqueued {enqueue_id}") + logger.info(f"Enqueued [{broker.list_key}] {enqueue_id}") logger.debug(f"Pushed {tag}") return task["id"] @@ -273,6 +274,7 @@ def fetch_cached(task_id, wait=0, broker=None): hook=task.get("hook"), args=task["args"], kwargs=task["kwargs"], + cluster=task.get("cluster"), started=task["started"], stopped=task["stopped"], result=task["result"], @@ -343,6 +345,7 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None) hook=task.get("hook"), args=task["args"], kwargs=task["kwargs"], + cluster=task.get("cluster"), started=task["started"], stopped=task["stopped"], result=task["result"], diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index bf226bb..cb6546d 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -354,7 +354,7 @@ def test_scheduler(broker, monkeypatch): assert schedule.next_run.date() == (timezone.now() + timedelta(weeks=2)).date() broker.delete_queue() - monkeypatch.setattr(Conf, "PREFIX", "some_cluster_name") + monkeypatch.setattr(Conf, "CLUSTER_NAME", "some_cluster_name") # create a schedule on another cluster schedule = create_schedule( "math.copysign", @@ -378,7 +378,7 @@ def test_scheduler(broker, monkeypatch): # queue must be empty assert task_queue.qsize() == 0 - monkeypatch.setattr(Conf, "PREFIX", "default") + monkeypatch.setattr(Conf, "CLUSTER_NAME", "default") # create a schedule on the same cluster schedule = create_schedule( "math.copysign", diff --git a/django_q/utils.py b/django_q/utils.py index 936214d..aed3c29 100644 --- a/django_q/utils.py +++ b/django_q/utils.py @@ -54,7 +54,7 @@ def get_func_repr(func): f"{func.__self__.__module__}." f"{func.__self__.__name__}.{func.__name__}" ) else: - return str(func) + return str(func) if func else None def localtime(value=None) -> datetime: diff --git a/docs/cluster.rst b/docs/cluster.rst index dc73bb9..fb6aa8b 100644 --- a/docs/cluster.rst +++ b/docs/cluster.rst @@ -52,6 +52,30 @@ You can have multiple clusters on multiple machines, working on the same queue a - They use the same cluster name. See :doc:`configure` - They share the same ``SECRET_KEY`` for Django. +.. _multiple-queues + +Multiple Queues +----------------- +You can have multiple queues in one Django site, and use multiple cluster to work on each queue. +Different queues are identified by different queue names which are also cluster names. +To run an alternate cluster, e.g. to work on the 'long' queue, start your cluster with command:: + + # On Linux + $ Q_CLUSTER_NAME=long python manage.py qcluster + + # On Windows + $ python manage.py qcluster --name long + +You can set different Q_CLUSTER options for alternative clusters, such as 'timeout', 'queue_limit' +and any other options which are valid in :doc:`configure`. See :ref:`alt-clusters`. + +.. note:: + + To use multiple queue, use the keyword argument `cluster` in async_task() and schedule(): + + * if `cluster` is not set (the default), async_task() and schedule() will be handled by the default cluster; + * if `cluster` is set, only clusters with matching cluster name will run the task or do the schedule. + Using a Procfile ---------------- If you host on `Heroku `__ or you are using `Honcho `__ you can start the cluster from a :file:`Procfile` with an entry like this:: diff --git a/docs/configure.rst b/docs/configure.rst index 3b96df8..713c2ce 100644 --- a/docs/configure.rst +++ b/docs/configure.rst @@ -20,7 +20,18 @@ Configuration is handled via the ``Q_CLUSTER`` dictionary in your :file:`setting 'redis': { 'host': '127.0.0.1', 'port': 6379, - 'db': 0, } + 'db': 0, }, + 'ALT_CLUSTERS': { + 'long': { + 'timeout': 3000, + 'retry': 3600, + 'max_attempts': 2, + }, + 'short': { + 'timeout': 10, + 'max_attempts': 1, + }, + } } All configuration settings are optional: @@ -459,6 +470,24 @@ As a rule of thumb; cpu_affinity 1 favors repetitive short running tasks, while *Psutil does not support cpu affinity on OS X at this time.* +.. _alt-clusters: + +ALT_CLUSTERS +~~~~~~~~~~~~ + +For multiple clusters working on multiple queues to run in one Django site. +ALT_CLUSTERS should be a dict with cluster_name as its key, and the value is the configuration for the cluster +with the key as its name. The configuration items are consistent with Q_CLUSTER, +except for a few items such as name/cluster_name/ALT_CLUSTER, which are not available of course. + +See :ref:`multiple-queues`. + +.. note:: + + For a cluster, if its name is in ALT_CLUSTERS, the config item in ALT_CLUSTER will override + the same config item in the Q_CLUSTER root. Other config items in Q_CLUSTER root remain in effect for this cluster. + + .. py:module:: django_q .. rubric:: Footnotes