New method of checking timeouted jobs, add maxconcurrency status in dashboard + test fixes

Author: sylvinus

Makefile

@@ -67,7 +67,7 @@ pep8:
 autopep8:
 	autopep8 --max-line-length 99 -aaaaaaaa --in-place --recursive mrq
 
-pypi: linterrors
+pypi: linterrors linterrors3
 	python setup.py sdist upload
 
 build_docs:

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

@@ -296,6 +296,7 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
                   'timeout': "label-danger",
                   'failed': "label-danger",
                   'maxretries': "label-danger",
+                  'maxconcurrency': "label-danger",
                   'interrupt': "label-danger",
                   'cancel': "label-warning",
                   'abort': "label-warning",

mrq/dashboard/templates/index.html

@@ -397,13 +397,16 @@ <h4 class="modal-title"></h4>
                     "failed": "failed",
                     "retry": "retry",
                     "maxretries": "maxretries",
+                    "maxconcurrency": "maxconcurrency",
                     "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}"
+                    "failed-retry-maxretries-maxconcurrency-timeout-interrupt-cancel-abort": "{NOT OK}",
+                    "failed-maxretries-maxconcurrency-timeout": "{ERROR}",
+                    "failed-retry-maxretries-maxconcurrency-interrupt-timeout": "{ERROR+RETRY}",
+
                   }, function(k, i) { %>
                   <option <%= filters.status==i?"selected='selected'":"" %> value="<%= i %>"><%= k %></option>
                 <% }) %>

mrq/job.py

@@ -23,6 +23,10 @@
 from . import context
 
 
+FINAL_STATUSES = {"timeout", "abort", "failed", "success", "interrupt", "retry", "maxretries", "maxconcurrency"}
+TRANSIENT_STATUSES = {"cancel", "queued", "started"}
+
+
 class Job(object):
 
     timeout = None
@@ -438,6 +442,11 @@ def _save_status(self, status, updates=None, exception=False, w=None, j=None):
         if self.id is None:
             return
 
+        # Forbid some status transitions
+        if self.data and self.data.get("status") in FINAL_STATUSES and status not in TRANSIENT_STATUSES:
+            context.log.error("Can't go from status %s to %s" % (self.data["status"], status))
+            return
+
         context.metric("jobs.status.%s" % status)
 
         if self.stored is False and self.statuses_no_storage is not None and status in self.statuses_no_storage:
@@ -476,7 +485,8 @@ def _save_status(self, status, updates=None, exception=False, w=None, j=None):
             db_updates["traceback"] = trace
             db_updates["exceptiontype"] = exc.__name__
 
-            self._save_traceback_history(status, trace, exc)
+        if self.data:
+            self.data.update(db_updates)
 
         # In the most common case, we allow an optimization on Mongo writes
         if status == "success":
@@ -500,8 +510,9 @@ def _save_status(self, status, updates=None, exception=False, w=None, j=None):
                 "_id": self.id
             }, {"$set": db_updates}, w=w, j=j, manipulate=False)
 
-        if self.data:
-            self.data.update(db_updates)
+        if exception:
+            self._save_traceback_history(status, trace, exc)
+
 
     def set_current_io(self, io_data):
 

mrq/worker.py

@@ -356,6 +356,28 @@ def report_worker(self, w=0):
         except Exception as e:  # pylint: disable=broad-except
             self.log.debug("Worker report failed: %s" % e)
 
+    def greenlet_timeouts(self):
+        """ This greenlet kills jobs in other greenlets if they timeout.
+        """
+
+        while True:
+            now = datetime.datetime.utcnow()
+            for greenlet in list(self.gevent_pool):
+                job = get_current_job(id(greenlet))
+                if job and job.timeout and job.datestarted:
+                    expires = job.datestarted + datetime.timedelta(seconds=job.timeout)
+                    if now > expires:
+                        greenlet.kill(block=False)
+                        if job.data["status"] != "timeout":
+                            updates = {
+                                "exceptiontype": "TimeoutInterrupt",
+                                "traceback": "".join(traceback.format_stack(greenlet.gr_frame))
+                            }
+                            job._save_status("timeout", updates=updates, exception=False)
+
+            time.sleep(1)
+
+
     def greenlet_admin(self):
         """ This greenlet is used to get status information about the worker
             when --admin_port was given
@@ -456,6 +478,8 @@ def work_init(self):
         if self.config["admin_port"]:
             self.greenlets["admin"] = gevent.spawn(self.greenlet_admin)
 
+        self.greenlets["timeouts"] = gevent.spawn(self.greenlet_timeouts)
+
         if self.config["scheduler"] and self.config["scheduler_interval"] > 0:
 
             from .scheduler import Scheduler
@@ -645,19 +669,6 @@ def perform_job(self, job):
 
         set_current_job(job)
 
-        gevent_timeout = None
-        if job.timeout:
-
-            gevent_timeout = gevent.Timeout(
-                job.timeout,
-                TimeoutInterrupt(
-                    'Job exceeded maximum timeout value in greenlet (%d seconds).' %
-                    job.timeout
-                )
-            )
-
-            gevent_timeout.start()
-
         try:
             job.perform()
 
@@ -691,9 +702,6 @@ def perform_job(self, job):
 
         finally:
 
-            if gevent_timeout:
-                gevent_timeout.cancel()
-
             set_current_job(None)
 
             self.done_jobs += 1

tests/fixtures/config-scheduler6.py

@@ -5,9 +5,9 @@
     {
         "path": "tests.tasks.general.MongoInsert",
         "params": {
-            "monthday": i
+            "monthday": i + 1
         },
-        "monthday": i,
+        "monthday": i + 1,
         "dailytime": datetime.datetime.fromtimestamp(float(os.environ.get("MRQ_TEST_SCHEDULER_TIME"))).time()
     } for i in range(31)
 ]

tests/tasks/general.py

@@ -17,6 +17,16 @@
 class Add(Task):
 
     def run(self, params):
+        if params.get("broadexcept"):
+            try:
+                return self._add(params)
+            except BaseException, e:  # Will catch greenlet.GreenletExit & all others
+                print("Got base exception %s" % type(e))
+                return
+        else:
+            return self._add(params)
+
+    def _add(self, params):
         log.info("adding", params)
         res = params.get("a", 0) + params.get("b", 0)
 
@@ -296,4 +306,14 @@ def run(self, params):
 
 class QueueAllKnown(Task):
     def run(self, params):
-        return list(Queue.all_known())
+        return list(Queue.all_known())
+
+
+class Uninterruptable(Task):
+
+    def run(self, params):
+        while True:
+            try:
+                time.sleep(1)
+            except:
+                pass

tests/test_general.py

@@ -189,12 +189,14 @@ def test_known_queues_lifecycle(worker):
     time.sleep(1)
 
     all_known_plus_sub = worker.send_task("tests.tasks.general.QueueAllKnown", {}, queue="default")
-    assert set(all_known_plus_sub).difference(set(all_known)) == set(["test_raw/sub"])
+    assert set(all_known_plus_sub) == set(all_known).union(set(["test_raw/sub"]))
 
-    Queue("test_raw/sub").remove_raw_jobs(["a", "b", "c"])
+    # This behavious was removed in https://github.com/pricingassistant/mrq/commit/dcb7c954c998d0d8f32b799da0fb0aa11524e5b9
+    # We might restore it if we find a way to improve performance
+    # Queue("test_raw/sub").remove_raw_jobs(["a", "b", "c"])
 
-    all_known_plus_sub = worker.send_task("tests.tasks.general.QueueAllKnown", {}, queue="default")
-    assert set(all_known_plus_sub).difference(set(all_known)) == set()
+    # all_known_plus_sub = worker.send_task("tests.tasks.general.QueueAllKnown", {}, queue="default")
+    # assert set(all_known_plus_sub) == set(all_known)
 
 
 def test_general_exception_status(worker):

tests/test_scheduler.py

@@ -166,7 +166,7 @@ def test_scheduler_weekday_dailytime(worker):
 
 
 def test_scheduler_monthday(worker):
-    # Task is scheduled in 3 seconds
+    # Task is scheduled in 10 seconds
     worker.start(
         flags="--scheduler --config tests/fixtures/config-scheduler6.py",
         env={
@@ -183,7 +183,7 @@ def test_scheduler_monthday(worker):
     assert len(inserts) == 0
 
     # And then only once
-    time.sleep(6)
+    time.sleep(10)
     inserts = list(collection.find())
     assert len(inserts) == 1
     assert collection.find({"params.monthday": datetime.datetime.utcnow().day}).count() == 1

tests/test_timeout.py

@@ -12,6 +12,10 @@ def test_timeout_normal(worker):
                          "a": 1, "b": 2, "sleep": 1000}, block=True, accept_statuses=["timeout"])
     assert r != 3
 
+    r = worker.send_task("tests.tasks.general.TimeoutFromConfig", {
+                         "a": 1, "b": 2, "sleep": 1000, "broadexcept": True}, block=True, accept_statuses=["timeout"])
+    assert r != 3
+
 
 def test_timeout_global_config(worker):