add support to delay a job and enqueue it N seconds later

Dockerfile

@@ -1,4 +1,4 @@
-FROM debian:jessie
+FROM debian:buster-slim
 
 #
 # httpredir.debian.org is often unreliable
@@ -11,32 +11,36 @@ FROM debian:jessie
 #     deb http://security.debian.org jessie/updates main\n' \
 #     > /etc/apt/sources.list
 
-RUN apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv 0C49F3730359A14518585931BC711F9BA15703C6
-RUN echo "deb http://repo.mongodb.org/apt/debian jessie/mongodb-org/3.4 main" > /etc/apt/sources.list.d/mongodb-org-3.4.list
 RUN apt-get update && \
 	apt-get install -y --no-install-recommends \
 				curl \
 				gcc \
 				python-dev \
 				python-pip \
+				python-setuptools \
 				python3-pip \
     			python3-dev \
+    			python3-setuptools \
     			make \
     			git \
     			vim \
     			bzip2 \
-				mongodb-org \
 				nginx redis-server \
 				g++ \
+				apt-utils \
 	&& \
 	apt-get clean -y && \
 	rm -rf /var/lib/apt/lists/*
 
-RUN curl -sL https://deb.nodesource.com/setup_7.x | bash -
+RUN curl -sL https://deb.nodesource.com/setup_12.x | bash -
 RUN apt-get install -y --no-install-recommends nodejs
 
+RUN curl -sL https://www.mongodb.org/static/pgp/server-4.4.asc | apt-key add -
+RUN echo "deb http://repo.mongodb.org/apt/debian buster/mongodb-org/4.4 main" > /etc/apt/sources.list.d/mongodb-org-4.4.list
+RUN apt-get update && apt-get install -y mongodb-org
+
 # Download pypy
-RUN curl -sL 'https://bitbucket.org/squeaky/portable-pypy/downloads/pypy-5.8-1-linux_x86_64-portable.tar.bz2' > /pypy.tar.bz2 && tar jxvf /pypy.tar.bz2 && rm -rf /pypy.tar.bz2 && mv /pypy-* /pypy
+RUN curl -sL 'https://github.com/squeaky-pl/portable-pypy/releases/download/pypy-7.2.0/pypy-7.2.0-linux_x86_64-portable.tar.bz2' > /pypy.tar.bz2 && tar jxvf /pypy.tar.bz2 && rm -rf /pypy.tar.bz2 && mv /pypy* /pypy
 
 # Upgrade pip
 RUN pip install --upgrade --ignore-installed pip
@@ -48,16 +52,16 @@ ADD requirements-base.txt /app/requirements-base.txt
 ADD requirements-dev.txt /app/requirements-dev.txt
 ADD requirements-dashboard.txt /app/requirements-dashboard.txt
 
-RUN pip3 install -r /app/requirements-heroku.txt && \
-	pip3 install -r /app/requirements-base.txt && \
-	pip3 install -r /app/requirements-dev.txt && \
-	pip3 install -r /app/requirements-dashboard.txt && \
+RUN python3 -m pip install -r /app/requirements-heroku.txt && \
+	python3 -m pip install -r /app/requirements-base.txt && \
+	python3 -m pip install -r /app/requirements-dev.txt && \
+	python3 -m pip install -r /app/requirements-dashboard.txt && \
 	rm -rf ~/.cache
 
-RUN pip install -r /app/requirements-heroku.txt && \
-	pip install -r /app/requirements-base.txt && \
-	pip install -r /app/requirements-dev.txt && \
-	pip install -r /app/requirements-dashboard.txt && \
+RUN python -m pip install -r /app/requirements-heroku.txt && \
+	python -m pip install -r /app/requirements-base.txt && \
+	python -m pip install -r /app/requirements-dev.txt && \
+	python -m pip install -r /app/requirements-dashboard.txt && \
 	rm -rf ~/.cache
 
 RUN /pypy/bin/pip install -r /app/requirements-heroku.txt && \

README.md

@@ -150,6 +150,17 @@ This was a preview on the very basic features of MRQ. What makes it actually use
 These features will be demonstrated in a future example of a simple web crawler.
 
 
+## How to compile a new version of MRQ
+- Go to mrq/version.py file and update from 0.9.28.1 to 0.9.29 (for example)
+- Commit and pushes the last changes
+- Go to in the master branch (after merge pull request)
+- Execute the command `python setup.py sdist`
+- Make sure you have installed the python twine package (`pip install twine`).
+- Execute the twine command to upload the new version of mrq to pypi.org (`twine upload dist/mrq-custom-0.9.29.tar.gz`)
+- In the previous step you will be asked for the pypi.org credentials are in 1password, you must request them in #password_requests
+- Update all applications that use MRQ with this version in their requirements.txt
+
+
 # More
 
 Full documentation is available on [readthedocs](http://mrq.readthedocs.org/en/latest/)

docs/jobs-maintenance.md

@@ -7,6 +7,13 @@ To do that, you should add these recurring scheduled jobs to your mrq-config.py:
 ```
 SCHEDULER_TASKS = [
 
+  # This will queue jobs in the 'delayed' status.
+  {
+    "path": "mrq.basetasks.cleaning.QueueDelayedJobs",
+    "params": {},
+    "interval": 60
+  },
+
   # This will requeue jobs in the 'retry' status, until they reach their max_retries.
   {
     "path": "mrq.basetasks.cleaning.RequeueRetryJobs",

docs/jobs.md

@@ -8,8 +8,9 @@ A **Job** is an instance of the execution of a Task. It must link to a specific
 
 MRQ defines a list of statuses for jobs. A job can only be in one of them at a time.
 
-When everything goes fine, a job will go through 3 statuses:
+When everything goes fine, a job will go through 4 statuses:
 
+* ```delayed```: The Job has been created and it is delayed to be queued later.
 * ```queued```: The Job has been created and it is waiting to be dequeued by a Worker.
 * ```started```: A Worker has dequeued the job and started executing it.
 * ```success```: The job was successfully ran.
@@ -51,6 +52,11 @@ Queues a job. If `queue` is not provided, the default queue for that Task as def
 
 Queues multiple jobs at once. Returns a list of IDs of the jobs.
 
+* `queue_job(main_task_path, params, delay=120, queue=None)`
+
+Create a job with `delayed` status, the job will be queued after of at least `delay` seconds. 
+Remember to add the base delayed job as explained in [Jobs maintenance](jobs-maintenance.md) to have `delayed` jobs actually queued.
+
 * `queue_raw_jobs(queue, params_list, batch_size=1000)`
 
 Queues multiple jobs at once on a [raw queue](queues.md#raw-queues). The queued jobs have no IDs on a raw queue so this function has no return.

mrq/agent.py

@@ -8,7 +8,11 @@
 import traceback
 from collections import defaultdict
 from bson import ObjectId
-from redis.lock import LuaLock
+try:
+    from redis.lock import LuaLock
+except ImportError:
+    # Change name to avoid NameError raised when use of LuaLock at line 147
+    from redis.lock import Lock as LuaLock
 from .processes import Process, ProcessPool
 from .utils import MovingETA, normalize_command
 from .queue import Queue

mrq/basetasks/cleaning.py

@@ -27,14 +27,41 @@ class RequeueRetryJobs(Task):
     max_concurrency = 1
 
     def run(self, params):
-        print("IN")
         return run_task("mrq.basetasks.utils.JobAction", {
             "status": "retry",
             "dateretry": {"$lte": datetime.datetime.utcnow()},
             "action": "requeue_retry"
         })
 
 
+class QueueDelayedJobs(Task):
+
+    """ Requeue jobs that were marked as delayed. """
+
+    max_concurrency = 1
+
+    def run(self, params):
+        return run_task("mrq.basetasks.utils.JobAction", {
+            "status": "delayed",
+            "dateretry": {"$lte": datetime.datetime.utcnow()},
+            "action": "requeue"
+        })
+
+
+class DeleteExpiresJobs(Task):
+
+    """ Delete jobs that were dateexpires is less than the current date time. """
+
+    max_concurrency = 1
+
+    def run(self, params):
+        return run_task("mrq.basetasks.utils.JobAction", {
+            "status": ["success","cancel"],
+            "dateexpires": {"$lte": datetime.datetime.utcnow()},
+            "action": "delete"
+        })
+
+
 class RequeueStartedJobs(Task):
 
     """ Requeue jobs that were marked as status=started and never finished.

mrq/basetasks/utils.py

@@ -10,6 +10,7 @@
 from mrq.utils import group_iter
 import datetime
 import ujson as json
+import sys
 
 
 def get_task_cfg(taskpath):
@@ -45,6 +46,7 @@ def build_query(self):
                 "status",
                 "worker",
                 "path",
+                "dateexpires",
                 "dateretry",
                 "exceptiontype"]:
             if self.params.get(k):
@@ -70,7 +72,8 @@ def perform_action(self, action, query, destination_queue):
 
         stats = {
             "requeued": 0,
-            "cancelled": 0
+            "cancelled": 0,
+            "deleted": 0
         }
 
         if action == "cancel":
@@ -151,7 +154,12 @@ def perform_action(self, action, query, destination_queue):
                     self.collection.update({
                         "_id": {"$in": jobs_by_queue[queue]}
                     }, {"$set": updates}, multi=True)
-
-                set_queues_size({queue: len(jobs) for queue, jobs in jobs_by_queue.iteritems()})
-
+                if sys.version_info.major > 2:
+                    set_queues_size({queue: len(jobs) for queue, jobs in jobs_by_queue.items()})
+                else:
+                    set_queues_size({queue: len(jobs) for queue, jobs in jobs_by_queue.iteritems()})
+        elif action == 'delete':
+            amount_delete = self.collection.delete_many(query)
+            stats["deleted"] = amount_delete.deleted_count
         return stats
+

mrq/context.py

@@ -13,6 +13,8 @@
 from .utils import LazyObject, load_class_by_path
 from .config import get_config
 from .subpool import subpool_map, subpool_imap
+from .exceptions import JobRuntimeError
+
 
 # This should be MRQ's only Python object shared by all the jobs in the same process
 _GLOBAL_CONTEXT = {
@@ -113,10 +115,14 @@ def get_current_config():
     return _GLOBAL_CONTEXT["config"]
 
 
-def retry_current_job(delay=None, max_retries=None, queue=None):
+def retry_current_job(delay=None, max_retries=None, queue=None, exception=None):
     current_job = get_current_job()
     if current_job:
         current_job.retry(delay=delay, max_retries=max_retries, queue=queue)
+    elif exception:
+        raise JobRuntimeError("Error during MRQ task execution") from exception
+    else:
+        raise JobRuntimeError("Error during MRQ task execution")
 
 
 def abort_current_job():
@@ -139,22 +145,32 @@ def versiontuple(v):
         if isinstance(config_obj, basestring):
 
             import redis as pyredis
+            import redis.connection as pyredisconnection
 
+            connection_class = pyredisconnection.Connection
             urllib.parse.uses_netloc.append('redis')
             redis_url = urllib.parse.urlparse(config_obj)
-
+            
             log.info("%s: Connecting to Redis at %s..." %
                      (attr, redis_url.hostname))
 
-            redis_pool = pyredis.BlockingConnectionPool(
+            redis_params = dict(
                 host=redis_url.hostname,
                 port=redis_url.port,
                 db=int((redis_url.path or "").replace("/", "") or "0"),
-                password=redis_url.password,
+                password=redis_url.password if redis_url.password is not None else redis_url.username,
                 max_connections=int(config.get("redis_max_connections")),
                 timeout=int(config.get("redis_timeout")),
-                decode_responses=False
+                decode_responses=False,
+                connection_class=connection_class
             )
+
+            if redis_url.scheme == "rediss":
+                redis_params["connection_class"] = pyredisconnection.SSLConnection
+                redis_params["ssl_cert_reqs"] = "none"
+
+            redis_pool = pyredis.BlockingConnectionPool(**redis_params)
+
             return pyredis.StrictRedis(connection_pool=redis_pool)
 
         # Let's just assume we got a StrictRedis-like object!

mrq/dashboard/app.py

@@ -54,8 +54,8 @@ def root():
 @requires_auth
 def api_task_exceptions():
     stats = list(connections.mongodb_jobs.mrq_jobs.aggregate([
-        {"$match": {"status": "failed"}},
-        {"$group": {"_id": {"path": "$path", "exceptiontype": "$exceptiontype"},
+        {"$match": {"status": {"$in": ["failed","maxretries"]}}},
+        {"$group": {"_id": {"path": "$path", "exceptiontype": "$exceptiontype","status":"$status"},
                     "jobs": {"$sum": 1}}},
     ]))
 

mrq/dashboard/static/js/views/jobs.js

@@ -347,6 +347,9 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
                 if (source.time) {
                   display.push("cputime "+String(source.time).substring(0,6)+"s ("+source.switches+" switches)");
                 }
+                if (source.status === 'retry' || source.status === 'delayed') {
+                  display.push("requeue "+moment.utc(source.dateretry).fromNow());
+                }
 
                 return "<small>" + display.join("<br/>") + "</small>";
 

mrq/dashboard/static/js/views/taskexceptions.js

@@ -32,7 +32,7 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
             "sClass": "col-exception",
             "sType":"numeric",
             "mData":function(source, type, val) {
-              return "<a href='/#jobs?path="+source._id.path+"&status=failed&exceptiontype="+source._id.exceptiontype+ "'>"+source._id.exceptiontype+"</a>"
+              return "<a href='/#jobs?path=" + source._id.path + "&status=" + source._id.status +"&exceptiontype="+source._id.exceptiontype+ "'>"+source._id.exceptiontype+"</a>"
             }
           },
           {
@@ -43,7 +43,7 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
               var cnt = source.jobs || 0;
 
               if (type == "display") {
-                return "<a href='/#jobs?path="+source._id.path+"&status=failed&exceptiontype="+source._id.exceptiontype+"'>"+cnt+"</a>"
+                return "<a href='/#jobs?path=" + source._id.path + "&status=" + source._id.status +"&exceptiontype="+source._id.exceptiontype+"'>"+cnt+"</a>"
                  + "<br/>"
                  + '<span class="inlinesparkline" values="'+self.addToCounter("taskexceptions."+source._id.path+" "+source._id.exceptiontype, cnt, 50).join(",")+'"></span>';
               } else {

mrq/dashboard/templates/index.html

@@ -391,6 +391,7 @@ <h4 class="modal-title"></h4>
               <select class="form-control input-sm js-datatable-filters-status" id="jobs-form-status">
                 <option <%= filters.status==""?"selected='selected'":"" %> value="">-statuses-</option>
                 <% _.each({
+                    "delayed": "delayed",
                     "queued": "queued",
                     "started": "started",
                     "success": "success",

mrq/exceptions.py

@@ -24,22 +24,33 @@ class TimeoutInterrupt(_MrqInterrupt):
 
 
 class AbortInterrupt(_MrqInterrupt):
-    pass
+    path = None
+
+    def _get_exception_name(self):
+        return "%s" % (
+            self.path
+        )
 
 
 class RetryInterrupt(_MrqInterrupt):
     delay = None
     queue = None
     retry_count = 0
+    path = None
 
     def _get_exception_name(self):
         return "%s #%s: %s seconds, %s queue" % (
-            self.__class__.__name__, self.retry_count, self.delay, self.queue
+            self.path, self.retry_count, self.delay, self.queue
         )
 
 
 class MaxRetriesInterrupt(_MrqInterrupt):
-    pass
+    path = None
+
+    def _get_exception_name(self):
+        return "%s" % (
+            self.path
+        )
 
 
 class StopRequested(GreenletExit):
@@ -53,4 +64,13 @@ class JobInterrupt(GreenletExit):
 
 
 class MaxConcurrencyInterrupt(_MrqInterrupt):
+    path = None
+
+    def _get_exception_name(self):
+        return "%s" % (
+            self.path
+        )
+
+
+class JobRuntimeError(_MrqInterrupt):
     pass