Add an "abort" status. Fixes #66

Author: sylvinus

docs/configuration.md

@@ -31,7 +31,7 @@ TASKS = {
         # Seconds before a job in retry status is requeued again
         "retry_delay": 600,
 
-        # Keep jobs with status in ("success", "cancel") that many seconds
+        # Keep jobs with status in ("success", "cancel", "abort") that many seconds
         # in MongoDB
         "result_ttl": 7 * 24 * 3600,
 

docs/jobs.md

@@ -19,7 +19,8 @@ When everything goes fine, a job will go through 3 statuses:
 However, to be reliable a task queue needs to prepare for everything that can go wrong. These statuses will help you manage those cases:
 
 * ```failed```: A Python Exception was raised during the execution of the job. It can be an Exception you raised yourself or an error in a module you are using. MRQ's Dashboard features a view where you can have a look at the tracebacks to debug these exceptions.
-* ```cancel```: The job was cancelled. This will happen mainly when you cancel jobs from the Dashboard before they run. Be careful, cancelling jobs when they are `started` won't interrupt the job.
+* ```cancel```: The job was cancelled. This will happen mainly when you cancel jobs from the Dashboard before they run. Be careful, cancelling jobs when they are `started` won't interrupt the currently running job.
+* ```abort```: The job was aborted. This happens while the job is running and `abort_current_job()` is called. Often this will be the result of an unrecoverable error that you don't want to retry but still want logged in the Dashboard for some time (see `result_ttl`)
 * ```interrupt```: While running this job, the worker was interrupted and had the time to save this status. This happens when the worker process receives the UNIX signal SIGTERM or two SIGINTs (which can happen by sending Ctrl-C two times). This status won't be set if the process is interrupted with a SIGKILL or any other abrupt means like a power off, and the task will stay in `started` state until requeued or cancelled by a maintenance job.
 * ```timeout```: The job took too long to finish and was interrupted by the worker. Timeouts can be set globally or for each task.
 * ```retry```: The method `task.retry()` was called to interrupt the job but mark it for being retried later. This may be useful when calling unreliable 3rd-party services.
@@ -70,6 +71,10 @@ The `max_retries` parameter defaults to the value of `max_retries` in the task c
 
 If the `queue` parameter is supplied, the job will be enventually requeued on that queue. If not, it will stay on its original queue.
 
+* `abort_current_job()`
+
+Stops the execution of the current job (by raising an `AbortInterrupt`) and mark it with the status `abort`. It will stay visible in the Dashboard for `result_ttl` seconds.
+
 * `get_current_job()`
 
 Returns the Job instance currently being executed. If None, you are outside of a Job context. This can only happen when calling code from your tasks outside of `mrq-run` or `mrq-worker`.

mrq/basetasks/utils.py

@@ -40,7 +40,10 @@ def build_query(self):
                 "dateretry",
                 "exceptiontype"]:
             if self.params.get(k):
-                query[k] = self.params.get(k)
+                if isinstance(self.params[k], (list, tuple)):
+                    query[k] = {"$in": list(self.params[k])}
+                else:
+                    query[k] = self.params[k]
 
         if self.params.get("params"):
             params_dict = json.loads(self.params.get("params"))  # pylint: disable=no-member

mrq/config.py

@@ -142,7 +142,7 @@ def add_parser_args(parser, config_type):
         default=7 * 24 * 3600,
         action='store',
         type=int,
-        help='Seconds the results are kept in MongoDB when status in ("success", "cancel")')
+        help='Seconds the results are kept in MongoDB when status in ("success", "cancel", "abort")')
 
     parser.add_argument(
         '--default_job_timeout',

mrq/context.py

@@ -82,6 +82,12 @@ def retry_current_job(delay=None, max_retries=None, queue=None):
         current_job.retry(delay=delay, max_retries=max_retries, queue=queue)
 
 
+def abort_current_job():
+    current_job = get_current_job()
+    if current_job:
+        current_job.abort()
+
+
 def _connections_factory(attr):
 
     config = get_current_config()

mrq/dashboard/app.py

@@ -125,9 +125,15 @@ def build_api_datatables_query(req):
             req.args.get("redisqueue")).list_job_ids(limit=1000)]}
     else:
 
-        for param in ["queue", "path", "status", "exceptiontype"]:
+        for param in ["queue", "path", "exceptiontype"]:
             if req.args.get(param):
                 query[param] = req.args.get(param)
+        if req.args.get("status"):
+            statuses = req.args.get(param).split("-")
+            if len(statuses) == 1:
+                query[param] = statuses[0]
+            else:
+                query[param] = {"$in": statuses}
         if req.args.get("id"):
             query["_id"] = ObjectId(req.args.get("id"))
         if req.args.get("worker"):
@@ -280,9 +286,11 @@ def api_job_traceback(job_id):
 @app.route('/api/jobaction', methods=["POST"])
 @requires_auth
 def api_job_action():
+    params = {k: v for k, v in request.form.iteritems()}
+    if params.get("status") and "-" in params.get("status"):
+        params["status"] = params.get("status").split("-")
     return jsonify({"job_id": queue_job("mrq.basetasks.utils.JobAction",
-                                        {k: v for k,
-                                         v in request.form.iteritems()},
+                                        params,
                                         queue=get_current_config()["dashboard_queue"])})
 
 

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

@@ -278,6 +278,7 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
                   'maxretries': "label-danger",
                   'interrupt': "label-danger",
                   'cancel': "label-warning",
+                  'abort': "label-warning",
                   'retry': "label-warning"
                 };
                 var css_class = status_classes[source.status] || "label-info";

mrq/dashboard/templates/index.html

@@ -318,8 +318,22 @@ <h4 class="modal-title"></h4>
             <div class="form-group">
               <select class="form-control input-sm js-datatable-filters-status" id="jobs-form-status">
                 <option <%= filters.status==""?"selected='selected'":"" %> value="">-statuses-</option>
-                <% _.each(["queued", "started", "success", "failed", "retry", "maxretries", "timeout", "interrupt", "cancel"], function(k) { %>
-                  <option <%= filters.status==k?"selected='selected'":"" %> value="<%= k %>"><%= k %></option>
+                <% _.each({
+                    "queued": "queued",
+                    "started": "started",
+                    "success": "success",
+                    "failed": "failed",
+                    "retry": "retry",
+                    "maxretries": "maxretries",
+                    "timeout": "timeout",
+                    "interrupt": "interrupt",
+                    "cancel": "cancel",
+                    "abort": "abort",
+                    "queued-started-success": "{OK}",
+                    "failed-retry-maxretries-timeout-interrupt-cancel-abort": "{NOT OK}",
+                    "failed-maxretries-timeout-abort": "{ERROR}"
+                  }, function(k, i) { %>
+                  <option <%= filters.status==i?"selected='selected'":"" %> value="<%= i %>"><%= k %></option>
                 <% }) %>
               </select>
             </div>

mrq/exceptions.py

@@ -1,28 +1,41 @@
 from gevent import GreenletExit
 import traceback
-import sys
 
 
 # Inherits from BaseException to avoid being caught when not intended.
-class TimeoutInterrupt(BaseException):
-    pass
-
+class _MrqInterrupt(BaseException):
 
-class RetryInterrupt(BaseException):
-    delay = None
-    queue = None
-    retry_count = 0
     original_exception = None
 
+    def _get_exception_name(self):
+        return self.__class__.__name__
+
     def __str__(self):
-        s = "<RetryInterrupt #%s: %s seconds, %s queue>" % (self.retry_count, self.delay, self.queue)
+        s = self._get_exception_name()
         if self.original_exception is not None:
             s += "\n---- Original exception: -----\n%s" % ("".join(traceback.format_exception(*self.original_exception)))
 
         return s
 
 
-class MaxRetriesInterrupt(BaseException):
+class TimeoutInterrupt(_MrqInterrupt):
+    pass
+
+
+class AbortInterrupt(_MrqInterrupt):
+    pass
+
+
+class RetryInterrupt(_MrqInterrupt):
+    delay = None
+    queue = None
+    retry_count = 0
+
+    def _get_exception_name(self):
+        return "%s #%s: %s seconds, %s queue" % (self.__class__.__name__, self.retry_count, self.delay, self.queue)
+
+
+class MaxRetriesInterrupt(_MrqInterrupt):
     pass
 
 

mrq/job.py

@@ -1,7 +1,7 @@
 import datetime
 from bson import ObjectId
 import time
-from .exceptions import RetryInterrupt, MaxRetriesInterrupt
+from .exceptions import RetryInterrupt, MaxRetriesInterrupt, AbortInterrupt
 from .utils import load_class_by_path
 from .queue import Queue
 from .context import get_current_worker, log, connections, get_current_config, metric
@@ -85,7 +85,7 @@ def fetch(self, start=False, full_data=True):
             self.set_data(self.collection.find_and_modify(
                 {
                     "_id": self.id,
-                    "status": {"$nin": ["cancel"]}
+                    "status": {"$nin": ["cancel", "abort", "maxretries"]}
                 },
                 {"$set": {
                     "status": "started",
@@ -171,6 +171,14 @@ def insert(cls, jobs_data, queue=None, return_jobs=True, w=1):
         else:
             return inserted
 
+    def _attach_original_exception(self, exc):
+        """ Often, a retry will be raised inside an "except" block.
+            This Keep track of the first exception for debugging purposes. """
+
+        original_exception = sys.exc_info()
+        if original_exception[0] is not None:
+            exc.original_exception = original_exception
+
     def retry(self, queue=None, delay=None, max_retries=None):
         """ Marks the current job as needing to be retried. Interrupts it. """
 
@@ -189,12 +197,14 @@ def retry(self, queue=None, delay=None, max_retries=None):
         if exc.delay is None:
             exc.delay = self.retry_delay
 
-        # Often, a retry will be raised inside an "except" block.
-        # Keep track of the first exception for debugging purposes.
-        original_exception = sys.exc_info()
-        if original_exception[0] is not None:
-            exc.original_exception = original_exception
+        self._attach_original_exception(exc)
+
+        raise exc
 
+    def abort(self):
+        """ Aborts the current task mid-excution. """
+        exc = AbortInterrupt()
+        self._attach_original_exception(exc)
         raise exc
 
     def cancel(self):