Merge branch 'logging'

Conflicts:
	pyres/horde.py

Author: Matt George

pyres/__init__.py

@@ -4,9 +4,27 @@
 import pyres.json_parser as json
 
 import time, datetime
-
+import sys
 import logging
 
+def setup_logging(log_level=logging.INFO, filename=None, stream=sys.stderr):
+    if log_level == logging.NOTSET:
+        return
+    logger = logging.getLogger()
+    logger.setLevel(log_level)
+    if filename:
+        try:
+            from logging.handlers import WatchedFileHandler
+            handler = WatchedFileHandler(filename)
+        except:
+            from logging.handlers import RotatingFileHandler
+            handler = RotatingFileHandler(filename,maxBytes=52428800, backupCount=7)
+    else:
+        handler = logging.StreamHandler(strm=stream)
+    handler.setFormatter(logging.Formatter(
+        '%(asctime)s %(levelname)-8s %(message)s', '%Y-%m-%d %H:%M:%S'))
+    logger.addHandler(handler)
+
 def my_import(name):
     """Helper function for walking import calls when searching for classes by string names."""
     mod = __import__(name)    

pyres/horde.py

@@ -34,7 +34,7 @@ def setup_logging(namespace='', log_level=logging.INFO, log_file=None):
     return logger
 
 class Minion(multiprocessing.Process):
-    def __init__(self, queues, server, password, log_level=logging.INFO, log_file=None):
+    def __init__(self, queues, server, password, log_level=logging.INFO, log_path=None):
         multiprocessing.Process.__init__(self, name='Minion')
 
         #format = '%(asctime)s %(levelname)s %(filename)s-%(lineno)d: %(message)s'
@@ -51,14 +51,13 @@ def __init__(self, queues, server, password, log_level=logging.INFO, log_file=No
         self.password = password
 
         self.log_level = log_level
-        self.log_file = log_file
+        self.log_path = log_path
+        self.log_file = None
 
     def prune_dead_workers(self):
         pass
 
     def schedule_shutdown(self, signum, frame):
-        print signum,
-        print 'inside shutdown'
         self._shutdown = True
 
     def register_signal_handlers(self):
@@ -148,16 +147,25 @@ def work(self, interval=5):
                 time.sleep(interval)
         self.unregister_minion()
 
+    def clear_logger(self):
+        for handler in self.logger.handlers:
+            self.logger.removeHandler(handler)
+    
     def run(self):
         setproctitle('pyres_minion:%s: Starting' % (os.getppid(),))
+        if self.log_path:
+            self.log_file = os.path.join(self.log_path, 'minion-%s.log' % self.pid)
+        namespace = 'minion:%s' % self.pid
+        self.logger = setup_logging(namespace, self.log_level, self.log_file)
+        #self.clear_logger()
         if isinstance(self.server,basestring):
             self.resq = ResQ(server=self.server, password=self.password)
         elif isinstance(self.server, ResQ):
             self.resq = self.server
         else:
             raise Exception("Bad server argument")
-        namespace = 'minion:%s' % self.pid
-        self.logger = setup_logging(namespace, self.log_level, self.log_file)
+        
+        
         self.work()
         #while True:
         #    job = self.q.get()
@@ -184,10 +192,11 @@ def __init__(self, pool_size=5, queues=[], server='localhost:6379', password=Non
         self.password = password
         self.logging_level = logging_level
         self.log_file = log_file
-        self.logger = setup_logging('khan', self.logging_level, self.log_file)
+        
         #self._workers = list()
 
     def setup_resq(self):
+        self.logger.info('Connecting to redis server - %s' % self.server)
         if isinstance(self.server,basestring):
             self.resq = ResQ(server=self.server, password=self.password)
         elif isinstance(self.server, ResQ):
@@ -236,7 +245,6 @@ def _check_commands(self):
             command = self.resq.redis.lpop('resque:khan:%s' % str(self))
             self.logger.debug('COMMAND FOUND: %s ' % command)
             if command:
-                import pdb;pdb.set_trace()
                 self.process_command(command)
                 self._check_commands()
 
@@ -250,17 +258,23 @@ def process_command(self, command):
                 fn()
 
     def add_minion(self):
-        m = self._add_minion()
+        self._add_minion()
         self.resq.redis.srem('resque:khans',str(self))
         self.pool_size += 1
         self.resq.redis.sadd('resque:khans',str(self))
 
     def _add_minion(self):
-        self.logger.info('Adding minion')
-        m = Minion(self.queues, self.server, self.password, log_level=self.logging_level)
+        if hasattr(self,'logger'):
+            self.logger.info('Adding minion')
+        if self.log_file:
+            log_path = os.path.dirname(self.log_file)
+        else:
+            log_path = None
+        m = Minion(self.queues, self.server, self.password, log_level=self.logging_level, log_path=log_path)
         m.start()
         self._workers[m.pid] = m
-        self.logger.info('minion added at: %s' % m.pid)
+        if hasattr(self,'logger'):
+            self.logger.info('minion added at: %s' % m.pid)
         return m
 
     def _shutdown_minions(self):
@@ -295,9 +309,13 @@ def work(self, interval=2):
         setproctitle('pyres_manager: Starting')
         self.startup()
         self.setup_minions()
+        self.logger = setup_logging('khan', self.logging_level, self.log_file)
+        self.logger.info('Running as pid: %s' % self.pid)
+        self.logger.info('Added %s child processes' % self.pool_size)
+        self.logger.info('Setting up pyres connection')
         self.setup_resq()
         self.register_khan()
-        setproctitle('pyres_manager: running - %s' % self.queues)
+        setproctitle('pyres_manager: running %s' % self.queues)
         while True:
             self._check_commands()
             if self._shutdown:
@@ -314,8 +332,8 @@ def __str__(self):
         return '%s:%s:%s' % (hostname, self.pid, self.pool_size)
 
     @classmethod
-    def run(cls, pool_size=5, queues=[], server='localhost:6379', logging_level=logging.INFO):
-        worker = cls(pool_size=pool_size, queues=queues, server=server, logging_level=logging_level)
+    def run(cls, pool_size=5, queues=[], server='localhost:6379', logging_level=logging.INFO, log_file=None):
+        worker = cls(pool_size=pool_size, queues=queues, server=server, logging_level=logging_level, log_file=log_file)
         worker.work()
 
 #if __name__ == "__main__":

scripts/pyres_manager

@@ -17,6 +17,7 @@ def main():
     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.")
+    parser.add_option('-f', dest='logfile', help='If present, a logfile will be used.')
     (options,args) = parser.parse_args()
 
     if len(args) != 1:
@@ -32,7 +33,7 @@ def main():
 
     queues = args[0].split(',')
     server = '%s:%s' % (options.host,options.port)
-    Khan.run(pool_size=options.pool_size, queues=queues, server=server, logging_level=log_level)
+    Khan.run(pool_size=options.pool_size, queues=queues, server=server, logging_level=log_level, log_file=options.logfile)
 
 if __name__ == "__main__":
     main()

scripts/pyres_scheduler

@@ -1,4 +1,5 @@
 #!/usr/bin/env python
+from pyres import setup_logging
 from pyres.scheduler import Scheduler
 import logging
 
@@ -10,9 +11,11 @@ def main():
     parser.add_option("--host", dest="host", default="localhost")
     parser.add_option("--port",dest="port",type="int", default=6379)
     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('-f', dest='logfile', help='If present, a logfile will be used.')
     (options,args) = parser.parse_args()
     log_level = getattr(logging, options.log_level.upper(),'INFO')
-    logging.basicConfig(level=log_level, format="%(module)s: %(asctime)s: %(levelname)s: %(message)s")
+    #logging.basicConfig(level=log_level, format="%(module)s: %(asctime)s: %(levelname)s: %(message)s")
+    setup_logging(log_level=log_level, filename=options.logfile)
     server = '%s:%s' % (options.host, options.port)
     Scheduler.run(server)
 

scripts/pyres_worker

@@ -4,6 +4,7 @@ import logging
 
 from optparse import OptionParser
 from pyres.worker import Worker
+from pyres import setup_logging
 
 
 
@@ -16,15 +17,16 @@ def main():
     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('-f', dest='logfile', help='If present, a logfile will be used.')
     (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")
-
+    #logging.basicConfig(level=log_level, format="%(asctime)s: %(levelname)s: %(message)s")
+    setup_logging(log_level=log_level, filename=options.logfile)
     interval = options.interval
     if interval is not None: 
         interval = float(interval)

tests/__init__.py

@@ -8,7 +8,6 @@ class Basic(object):
     @staticmethod
     def perform(name):
         s = "name:%s" % name
-        print s
         return s
 
 class BasicMulti(object):