Traceback histories

mrq/config.py

@@ -419,10 +419,15 @@ def get_config(
                 from_args[k] = v
 
     # If we were given another config file, use it
+    print "hello"
+    print file_path
+    print from_args.get("config")
+
     if file_path is not None:
         config_file = file_path
     elif from_args.get("config"):
         config_file = from_args.get("config")
+        print config_file
     # If a mrq-config.py file is in the current directory, use it!
     elif os.path.isfile(os.path.join(os.getcwd(), "mrq-config.py")):
         config_file = os.path.join(os.getcwd(), "mrq-config.py")

mrq/job.py

@@ -17,6 +17,7 @@
 import encodings
 import copy_reg
 from . import context
+from mrq.context import get_current_config
 
 
 class Job(object):
@@ -203,7 +204,7 @@ def insert(cls, jobs_data, queue=None, statuses_no_storage=None, return_jobs=Tru
 
     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. """
+            This Keep track of the first exception for debugging purposes """
 
         original_exception = sys.exc_info()
         if original_exception[0] is not None:
@@ -350,6 +351,23 @@ def save_retry(self, retry_exc):
 
             self._save_status("retry", updates, exception=True)
 
+    def _save_traceback_history(self, status, trace, job_exc):
+        """ Create traceback history or add a new traceback to history. """
+        failure_date = datetime.datetime.utcnow()
+
+        new_history = {
+            "date": failure_date,
+            "status": status,
+            "exceptiontype": job_exc.__name__
+        }
+        traces = trace.split("---- Original exception: -----")
+        if len(traces) > 1:
+            new_history["original_traceback"] = traces[1]
+        new_history["traceback"] = traces[0]
+        self.collection.update({
+            "_id": self.id
+        }, {"$push": {"traceback_history": new_history}})
+
     def save_success(self, result=None):
 
         dateexpires = datetime.datetime.utcnow() + datetime.timedelta(seconds=self.result_ttl)
@@ -413,7 +431,12 @@ def _save_status(self, status, updates=None, exception=False, w=None, j=None):
             trace = traceback.format_exc()
             context.log.error(trace)
             db_updates["traceback"] = trace
-            db_updates["exceptiontype"] = sys.exc_info()[0].__name__
+            exc = sys.exc_info()[0]
+            db_updates["exceptiontype"] = exc.__name__
+
+            if get_current_config().get("save_traceback_history"):
+
+                self._save_traceback_history(status, trace, exc)
 
         # In the most common case, we allow an optimization on Mongo writes
         if status == "success":

tests/fixtures/config-tracebackhistory.py

@@ -0,0 +1 @@
+SAVE_TRACEBACK_HISTORY = True

tests/tasks/general.py

@@ -6,13 +6,14 @@
 import json
 import time
 import copy
+from mrq.config import get_config
 from mrq.utils import MongoJSONEncoder
 
 
 class Add(Task):
 
     def run(self, params):
-
+        print get_config()
         log.info("adding", params)
         res = params.get("a", 0) + params.get("b", 0)
 
@@ -86,6 +87,30 @@ def run(self, params):
         raise Exception("Should not be reached")
 
 
+class InRetryException(BaseException):
+    pass
+
+
+class RetryOnFailed(Task):
+
+    def run(self, params):
+
+        log.info("Retrying in %s on %s" %
+                 (params.get("delay"), params.get("queue")))
+
+        connections.mongodb_jobs.tests_inserts.insert(params)
+        try:
+            raise InRetryException
+        except InRetryException:
+            retry_current_job(
+                queue=params.get("queue"),
+                delay=params.get("delay"),
+                max_retries=params.get("max_retries")
+            )
+
+        raise Exception("Should not be reached")
+
+
 class WaitForFlag(Task):
 
     def run(self, params):

tests/test_retry.py

@@ -159,3 +159,37 @@ def test_retry_max_retries_zero(worker):
 
     job = Job(job_id).fetch()
     assert job.data["status"] == "maxretries"
+
+
+def test_retry_traceback_history(worker):
+
+    worker.start(flags="--config tests/fixtures/config-tracebackhistory.py")
+    # delay = 0 should requeue right away.
+
+    worker.send_task(
+        "tests.tasks.general.Retry", {"queue": "noexec", "delay": 60}, block=True, accept_statuses=["retry"]
+    )
+
+    job = worker.mongodb_jobs.mrq_jobs.find()[0]
+
+    assert len(job["traceback_history"]) == 1
+    assert not job["traceback_history"][0].get("original_traceback")
+
+    worker.send_task(
+        "tests.tasks.general.RetryOnFailed", {"queue": "default", "delay": 1}, block=True, accept_statuses=["retry"]
+    )
+
+    job = worker.mongodb_jobs.mrq_jobs.find({
+        "path": "tests.tasks.general.RetryOnFailed"})[0]
+
+    assert len(job["traceback_history"]) == 1
+    assert "InRetryException" in job["traceback_history"][0].get("original_traceback")
+    time.sleep(2)
+    worker.send_task("mrq.basetasks.cleaning.RequeueRetryJobs", {}, block=True)
+    time.sleep(2)
+    job = worker.mongodb_jobs.mrq_jobs.find({
+        "path": "tests.tasks.general.RetryOnFailed"})[0]
+
+    print job["traceback_history"]
+    assert len(job["traceback_history"]) == 2
+