Merge branch '0.7.0-fixes'

Author: Matt George

pyres/__init__.py

@@ -170,7 +170,7 @@ def info(self):
         }
 
     def keys(self):
-        return [key.replace('resque:','') for key in self.redis.keys('*')]
+        return [key.replace('resque:','') for key in self.redis.keys('resque:*')]
 
     def reserve(self, queue):
         from pyres.job import Job

pyres/failure/base.py

@@ -20,26 +20,15 @@ def __init__(self, exp, queue, payload, worker=None):
         excc, _, tb = sys.exc_info()
 
         self._exception = excc
-        self._traceback = tb
+        self._traceback = traceback.format_exc()
         self._worker = worker
         self._queue = queue
         self._payload = payload
 
 
     def _parse_traceback(self, trace):
         """Return the given traceback string formatted for a notification."""
-        reversed_backtrace = list(
-                reversed(traceback.extract_tb(trace))
-        )
-        p_traceback = []
-        for filename, lineno, funcname, text in reversed_backtrace:
-            p_traceback.append("%s:%s:%d:in `%s`" % (text, filename, lineno, funcname))
-        #p_traceback = [ "%s:%d:in `%s'" % (filename, lineno, funcname) 
-        #               for filename, lineno, funcname, _
-        #                in traceback.extract_tb(trace) ]
-        #p_traceback.reverse()
-
-        return p_traceback
+        return trace
 
     def _parse_message(self, exc):
         """Return a message for a notification from the given exception."""

pyres/worker.py

@@ -7,6 +7,7 @@
 import os, sys
 import time
 import json_parser as json
+import commands
 
 class Worker(object):
     """Defines a worker. The ``pyres_worker`` script instantiates this Worker class and
@@ -22,6 +23,7 @@ def __init__(self, queues=[], server="localhost:6379", password=None):
         self._shutdown = False
         self.child = None
         self.pid = os.getpid()
+        self.hostname = os.uname()[1]
         if isinstance(server,basestring):
             self.resq = ResQ(server=server, password=password)
         elif isinstance(server, ResQ):
@@ -62,8 +64,21 @@ def unregister_worker(self):
         Stat("processed:%s" % self, self.resq).clear()
         Stat("failed:%s" % self, self.resq).clear()
 
+    def prune_dead_workers(self):
+        all_workers = Worker.all(self.resq)
+        known_workers = self.worker_pids()
+        for worker in all_workers:
+            host, pid, queues = worker.id.split(':')
+            if host != self.hostname:
+                continue
+            if pid in known_workers:
+                continue
+            logging.warning("pruning dead worker: %s" % worker)
+            worker.unregister_worker()
+
     def startup(self):
         self.register_signal_handlers()
+        self.prune_dead_workers()
         self.register_worker()
 
     def register_signal_handlers(self):
@@ -83,12 +98,11 @@ def kill_child(self, signum, frame):
         if self.child:
             logging.info("Killing child at %s" % self.child)
             os.kill(self.child, signal.SIGKILL)
-            
+
     def __str__(self):
         if getattr(self,'id', None):
             return self.id
-        hostname = os.uname()[1]
-        return '%s:%s:%s' % (hostname, self.pid, ','.join(self.queues))
+        return '%s:%s:%s' % (self.hostname, self.pid, ','.join(self.queues))
 
     def work(self, interval=5):
         """Invoked by ``run`` method. ``work`` listens on a list of queues and sleeps
@@ -204,6 +218,13 @@ def processing(self):
 
     def state(self):
         return 'working' if self.resq.redis.exists('resque:worker:%s' % self) else 'idle'
+
+    def worker_pids(self):
+        """Returns an array of all pids (as strings) of the workers on
+        this machine.  Used when pruning dead workers."""
+        return map(lambda l: l.split(' ')[0],
+                   commands.getoutput("ps -A -o pid,command | \
+                                       grep pyres_worker").split("\n"))
 
     @classmethod
     def run(cls, queues, server, interval):

resweb/server.py

@@ -30,7 +30,7 @@ def working(request):
 def queues(request):
     return str(Queues(HOST).render())
 
-@get('/queues/(?P<queue_id>\w+)/')
+@get('/queues/(?P<queue_id>\w.+)/')
 def queue(request, queue_id):
     start = int(request.GET.get('start',0))
     return str(Queue(HOST, queue_id, start).render())

resweb/views.py

@@ -212,9 +212,9 @@ def failed_jobs(self):
         for job in failure.all(self.resq, self._start, self._start + 20):
             item = job
             item['worker_url'] = '/workers/%s/' % job['worker']
-            item['payload_args'] = ','.join(job['payload']['args'])
+            item['payload_args'] = str(job['payload']['args'])
             item['payload_class'] = job['payload']['class']
-            item['traceback'] = '\n'.join(job['backtrace'])
+            item['traceback'] = job['backtrace']
             jobs.append(item)
         return jobs
 
@@ -287,7 +287,7 @@ def key_info(self):
 
             stats.append({
                 'key': str(key),
-                'type': str(self.resq.redis.get_type('resque:'+key)),
+                'type': str(self.resq.redis.type('resque:'+key)),
                 'size': str(redis_size(key, self.resq)) 
             })
         return stats
@@ -308,7 +308,7 @@ def key(self):
         return str(self.stat_id)
 
     def key_type(self):
-        return str(self.resq.redis.get_type(self.stat_id))
+        return str(self.resq.redis.type('resque:'+ str(self.stat_id)))
 
     def items(self):
         items = []
@@ -407,7 +407,7 @@ def runat(self):
         """
         pass
 def redis_size(key, resq):
-    key_type = resq.redis.get_type('resque:'+key)
+    key_type = resq.redis.type('resque:'+key)
     item = 0
     if key_type == 'list':
         item = resq.redis.llen('resque:'+key)

tests/test_worker.py

@@ -156,3 +156,19 @@ def test_state(self):
         worker.done_working()
         assert worker.state() == 'idle'
 
+    def test_prune_dead_workers(self):
+        worker = Worker(['basic']) # we haven't registered this worker, so the assertion below holds
+        assert self.redis.scard('resque:workers') == 0
+        self.redis.sadd('resque:workers',"%s:%s:%s" % (os.uname()[1],'1','basic'))
+        self.redis.sadd('resque:workers',"%s:%s:%s" % (os.uname()[1],'2','basic'))
+        self.redis.sadd('resque:workers',"%s:%s:%s" % (os.uname()[1],'3','basic'))
+        assert self.redis.scard('resque:workers') == 3
+        worker.prune_dead_workers()
+        assert self.redis.scard('resque:workers') == 0
+        self.redis.sadd('resque:workers',"%s:%s:%s" % ('host-that-does-not-exist','1','basic'))
+        self.redis.sadd('resque:workers',"%s:%s:%s" % ('host-that-does-not-exist','2','basic'))
+        self.redis.sadd('resque:workers',"%s:%s:%s" % ('host-that-does-not-exist','3','basic'))
+        worker.prune_dead_workers()
+        # the assertion below should hold, because the workers we registered above are on a
+        # different host, and thus should not be pruned by this process
+        assert self.redis.scard('resque:workers') == 3