updates to the horde package

Author: Matt George

pyres/horde.py

@@ -5,11 +5,14 @@
     sys.exit("multiprocessing was not available")
 
 import time, os, signal
+import datetime
+import logging
+
 from pyres.worker import Worker
 from pyres import ResQ
 from pyres.exceptions import NoQueueError
 from pyres.utils import OrderedDict
-import datetime
+
 
 class Minion(multiprocessing.Process, Worker):
     def __init__(self, queues, server, password):
@@ -18,6 +21,7 @@ def __init__(self, queues, server, password):
         self.validate_queues()
         self._shutdown = False
         self.child = None
+        self.hostname = os.uname()[1]
         if isinstance(server,basestring):
             self.resq = ResQ(server=server, password=password)
         elif isinstance(server, ResQ):
@@ -26,12 +30,14 @@ def __init__(self, queues, server, password):
             raise Exception("Bad server argument")
         #Worker.__init__(self, queues=queues, server="localhost:6379", password=None)
         #super(Minion,self).__init__(name='Minion')
+    def prune_dead_workers(self):
+        pass
 
     def work(self, interval=5):
         self.startup()
         while True:
             if self._shutdown:
-                print 'shutdown scheduled'
+                logging.info('shutdown scheduled')
                 break
             job = self.reserve()
             if job:
@@ -53,7 +59,6 @@ class Khan(object):
         'REMOVE': '_remove_minion',
         'SHUTDOWN': '_schedule_shutdown'
     }
-    _workers = OrderedDict()
     def __init__(self, pool_size=5, queues=[], server='localhost:6379', password=None):
         #super(Khan,self).__init__(queues=queues,server=server,password=password)
         self._shutdown = False
@@ -63,6 +68,7 @@ def __init__(self, pool_size=5, queues=[], server='localhost:6379', password=Non
         self.password = password
         self.pid = os.getpid()
         self.validate_queues()
+        self._workers = OrderedDict()
         if isinstance(server,basestring):
             self.resq = ResQ(server=server, password=password)
         elif isinstance(server, ResQ):
@@ -91,7 +97,7 @@ def _schedule_shutdown(self):
         self.schedule_shutdown(None, None)
 
     def schedule_shutdown(self, signum, frame):
-        print 'Shutdown scheduled'
+        logging.info('Khan Shutdown scheduled')
         self._shutdown = True
 
     def kill_child(self, signum, frame):
@@ -106,10 +112,10 @@ def register_khan(self):
 
     def _check_commands(self):
         if not self._shutdown:
-            print 'Checking commands'
+            logging.debug('Checking commands')
             command_key = 'resque:khan:%s' % self
-            command = self.resq.redis.pop(command_key)
-            print 'COMMAND', command
+            command = self.resq.redis.lpop(command_key)
+            logging.debug('COMMAND FOUND:', command)
             if command:
                 self.process_command(command)
                 self._check_commands()
@@ -124,16 +130,16 @@ def process_command(self, command):
                 fn()
 
     def add_minion(self):
-        print 'Adding minion'
         m = self._add_minion()
         m.start()
 
     def _add_minion(self):
-        print 'Adding mminion'
+        logging.info('Adding minion')
         #parent_conn, child_conn = multiprocessing.Pipe()
         m = Minion(self.queues, self.server, self.password)
         #m.start()
         self._workers[m.pid] = m
+        logging.info('minion added at %s' % m.pid)
         return m
         #self._workers.append(m)
 
@@ -153,12 +159,13 @@ def _remove_minion(self, pid=None):
         return m
 
     def register_worker(self):
+        logging.debug('registering khan')
         self.resq.redis.sadd('resque:khans',str(self))
         #self.resq._redis.add("worker:#{self}:started", Time.now.to_s)
         self.started = datetime.datetime.now()
 
     def unregister_worker(self):
-        print 'Unregistering'
+        logging.debug('unregistering khan')
         self.resq.redis.srem('resque:khans',str(self))
         self.started = None
 

pyres/worker.py

@@ -227,7 +227,7 @@ def worker_pids(self):
                                        grep pyres_worker").split("\n"))
 
     @classmethod
-    def run(cls, queues, server="localhost:6379", interval):
+    def run(cls, queues, server="localhost:6379", interval=None):
         worker = cls(queues=queues, server=server)
         if interval is not None:
             worker.work(interval)            

scripts/pyres_manager

@@ -0,0 +1,38 @@
+#!/usr/bin/env python
+
+import logging
+
+from optparse import OptionParser
+from pyres.horde import Khan
+
+
+
+def main():
+    
+    usage = "usage: %prog [options] arg1"
+    parser = OptionParser(usage=usage)
+    #parser.add_option("-q", dest="queue_list")
+    parser.add_option("--host", dest="host", default="localhost")
+    parser.add_option("--port",dest="port",type="int", default=6379)
+    parser.add_option("-i", '--interval', dest='interval', default=None, help='the default time interval to sleep between runs')    
+    parser.add_option('-l', '--log-level', dest='log_level', default='info', help='log level.  Valid values are "debug", "info", "warning", "error", "critical", in decreasing order of verbosity. Defaults to "info" if parameter not specified.')
+    parser.add_option("--pool", type="int", dest="pool_size", default=1, help="Number of minions to spawn under the manager.")
+    (options,args) = parser.parse_args()
+
+    if len(args) != 1:
+        parser.print_help()
+        parser.error("Argument must be a comma seperated list of queues")
+
+    log_level = getattr(logging, options.log_level.upper(), 'INFO')
+    logging.basicConfig(level=log_level, format="%(asctime)s: %(levelname)s: %(message)s")
+
+    interval = options.interval
+    if interval is not None: 
+        interval = float(interval)
+
+    queues = args[0].split(',')
+    server = '%s:%s' % (options.host,options.port)
+    Khan.run(pool_size=options.pool_size, queues=queues, server=server)
+
+if __name__ == "__main__":
+    main()