Introducing ttl for aborted jobs (default is 24h), configurable through mrq-config

Author: antoine-collet

mrq/config.py

@@ -156,7 +156,14 @@ 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, abort)')
+        help='Seconds the results are kept in MongoDB when status in (success)')
+
+    parser.add_argument(
+        '--default_job_aborted_or_canceled_ttl',
+        default=24 * 3600,
+        action='store',
+        type=int,
+        help='Seconds the tasks are kept in MongoDB when status in (cancel, abort)')
 
     parser.add_argument(
         '--default_job_timeout',

mrq/job.py

@@ -23,6 +23,7 @@ class Job(object):
 
     timeout = None
     result_ttl = None
+    aborted_or_canceled_ttl = None
     max_retries = None
     retry_delay = None
 
@@ -127,6 +128,7 @@ def set_data(self, data):
 
             self.timeout = task_def.get("timeout", cfg["default_job_timeout"])
             self.result_ttl = task_def.get("result_ttl", cfg["default_job_result_ttl"])
+            self.aborted_or_canceled_ttl = task_def.get("aborted_or_canceled_ttl", cfg["default_job_aborted_or_canceled_ttl"])
             self.max_retries = task_def.get("max_retries", cfg["default_job_max_retries"])
             self.retry_delay = task_def.get("retry_delay", cfg["default_job_retry_delay"])
 
@@ -340,13 +342,21 @@ def save_success(self, result=None):
 
     def save_cancel(self):
 
-        dateexpires = datetime.datetime.utcnow() + datetime.timedelta(seconds=self.result_ttl)
+        dateexpires = datetime.datetime.utcnow() + datetime.timedelta(seconds=self.aborted_or_canceled_ttl)
         updates = {
             "dateexpires": dateexpires
         }
 
         self._save_status("cancel", updates)
 
+    def save_abort(self):
+        dateexpires = datetime.datetime.utcnow() + datetime.timedelta(seconds=self.aborted_or_canceled_ttl)
+        updates = {
+            "dateexpires": dateexpires
+        }
+
+        self._save_status("abort", updates)
+
     def _save_status(self, status, updates=None, exception=False, w=1):
 
         if self.id is None:

mrq/worker.py

@@ -538,7 +538,7 @@ def perform_job(self, job):
 
         except AbortInterrupt:
             self.log.error("Caught abort")
-            job._save_status("abort", exception=True)
+            job.save_abort()
 
         except TimeoutInterrupt:
             self.log.error("Job timeouted after %s seconds" % job.timeout)

tests/test_abort.py

@@ -1,6 +1,7 @@
 from mrq.job import Job
 from mrq.queue import Queue
-import time
+from datetime import datetime
+from datetime import timedelta
 
 
 def test_abort(worker):
@@ -14,4 +15,7 @@ def test_abort(worker):
     db_jobs = list(worker.mongodb_jobs.mrq_jobs.find())
     assert len(db_jobs) == 1
 
-    assert db_jobs[0]["status"] == "abort"
+    job = db_jobs[0]
+    assert job["status"] == "abort"
+    assert job.get("dateexpires") is not None
+    assert job["dateexpires"] < datetime.utcnow() + timedelta(hours=24)