rework logging connections

IPython/zmq/parallel/clusterdir.py

@@ -480,12 +480,11 b' class ApplicationWithClusterDir(Application):'
             open_log_file = None
         else:
             open_log_file = sys.stdout
-        logger = logging.getLogger()
-        level = self.log_level
-        self.log = logger
-        # since we've reconnected the logger, we need to reconnect the log-level
-        self.log_level = level
-        if open_log_file is not None and self._log_handler not in self.log.handlers:
+        if open_log_file is not None:
+            self.log.removeHandler(self._log_handler)
+            self._log_handler = logging.StreamHandler(open_log_file)
+            self._log_formatter = logging.Formatter("[%(name)s] %(message)s")
+            self._log_handler.setFormatter(self._log_formatter)
             self.log.addHandler(self._log_handler)
         # log.startLogging(open_log_file)
 

IPython/zmq/parallel/controller.py

@@ -44,14 +44,12 b' class ControllerFactory(HubFactory):'
     children = List()
     mq_class = Str('zmq.devices.ProcessMonitoredQueue')
     
-    def _update_mq(self):
-        self.mq_class = 'zmq.devices.%sMonitoredQueue'%('Thread' if self.usethreads else 'Process')
+    def _usethreads_changed(self, name, old, new):
+        self.mq_class = 'zmq.devices.%sMonitoredQueue'%('Thread' if new else 'Process')
         
     def __init__(self, **kwargs):
         super(ControllerFactory, self).__init__(**kwargs)
         self.subconstructors.append(self.construct_schedulers)
-        self._update_mq()
-        self.on_trait_change(self._update_mq, 'usethreads')
     
     def start(self):
         super(ControllerFactory, self).start()
@@ -91,7 +89,7 b' class ControllerFactory(HubFactory):'
         children.append(q)
         # Task Queue (in a Process)
         if self.scheme == 'pure':
-            logging.warn("task::using pure XREQ Task scheduler")
+            self.log.warn("task::using pure XREQ Task scheduler")
             q = mq(zmq.XREP, zmq.XREQ, zmq.PUB, 'intask', 'outtask')
             q.bind_in(self.client_addrs['task'])
             q.bind_out(self.engine_addrs['task'])
@@ -99,10 +97,10 b' class ControllerFactory(HubFactory):'
             q.daemon=True
             children.append(q)
         elif self.scheme == 'none':
-            logging.warn("task::using no Task scheduler")
+            self.log.warn("task::using no Task scheduler")
             
         else:
-            logging.warn("task::using Python %s Task scheduler"%self.scheme)
+            self.log.warn("task::using Python %s Task scheduler"%self.scheme)
             sargs = (self.client_addrs['task'], self.engine_addrs['task'], self.monitor_url, self.client_addrs['notification'])
             q = Process(target=launch_scheduler, args=sargs, kwargs = dict(scheme=self.scheme))
             q.daemon=True

IPython/zmq/parallel/engine.py

@@ -25,8 +25,8 b' from streamkernel import Kernel'
 import heartmonitor
 
 def printer(*msg):
-    # print (logging.handlers, file=sys.__stdout__)
-    logging.info(str(msg))
+    # print (self.log.handlers, file=sys.__stdout__)
+    self.log.info(str(msg))
 
 class EngineFactory(RegistrationFactory):
     """IPython engine"""
@@ -54,7 +54,7 b' class EngineFactory(RegistrationFactory):'
     def register(self):
         """send the registration_request"""
         
-        logging.info("registering")
+        self.log.info("registering")
         content = dict(queue=self.ident, heartbeat=self.ident, control=self.ident)
         self.registrar.on_recv(self.complete_registration)
         # print (self.session.key)
@@ -112,10 +112,9 b' class EngineFactory(RegistrationFactory):'
                 sys.displayhook = self.display_hook_factory(self.session, iopub_stream)
                 sys.displayhook.topic = 'engine.%i.pyout'%self.id
             
-            self.kernel = Kernel(int_id=self.id, ident=self.ident, session=self.session, 
-                    control_stream=control_stream,
-                    shell_streams=shell_streams, iopub_stream=iopub_stream, loop=loop,
-                    user_ns = self.user_ns, config=self.config)
+            self.kernel = Kernel(config=self.config, int_id=self.id, ident=self.ident, session=self.session, 
+                    control_stream=control_stream, shell_streams=shell_streams, iopub_stream=iopub_stream, 
+                    loop=loop, user_ns = self.user_ns, logname=self.log.name)
             self.kernel.start()
             
             heart = heartmonitor.Heart(*map(str, hb_addrs), heart_id=identity)
@@ -124,10 +123,10 b' class EngineFactory(RegistrationFactory):'
             
             
         else:
-            logging.error("Registration Failed: %s"%msg)
+            self.log.error("Registration Failed: %s"%msg)
             raise Exception("Registration Failed: %s"%msg)
         
-        logging.info("Completed registration with id %i"%self.id)
+        self.log.info("Completed registration with id %i"%self.id)
     
     
     def unregister(self):

IPython/zmq/parallel/entry_point.py

@@ -79,8 +79,8 b' def integer_loglevel(loglevel):'
             loglevel = getattr(logging, loglevel)
     return loglevel
 
-def connect_logger(context, iface, root="ip", loglevel=logging.DEBUG):
-    logger = logging.getLogger()
+def connect_logger(logname, context, iface, root="ip", loglevel=logging.DEBUG):
+    logger = logging.getLogger(logname)
     if any([isinstance(h, handlers.PUBHandler) for h in logger.handlers]):
         # don't add a second PUBHandler
         return
@@ -106,9 +106,9 b' def connect_engine_logger(context, iface, engine, loglevel=logging.DEBUG):'
     logger.addHandler(handler)
     logger.setLevel(loglevel)
 
-def local_logger(loglevel=logging.DEBUG):
+def local_logger(logname, loglevel=logging.DEBUG):
     loglevel = integer_loglevel(loglevel)
-    logger = logging.getLogger()
+    logger = logging.getLogger(logname)
     if any([isinstance(h, logging.StreamHandler) for h in logger.handlers]):
         # don't add a second StreamHandler
         return

IPython/zmq/parallel/factory.py

@@ -29,9 +29,15 b' import IPython.zmq.parallel.streamsession as ss'
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
+class LoggingFactory(Configurable):
+    """A most basic class, that has a `log` (type:`Logger`) attribute, set via a `logname` Trait."""
+    log = Instance('logging.Logger', ('ZMQ', logging.WARN))
+    logname = CStr('ZMQ')
+    def _logname_changed(self, name, old, new):
+        self.log = logging.getLogger(new)
 
 
-class SessionFactory(Configurable):
+class SessionFactory(LoggingFactory):
     """The Base factory from which every factory in IPython.zmq.parallel inherits"""
     
     packer = Str('',config=True)
@@ -41,14 +47,14 b' class SessionFactory(Configurable):'
         return str(uuid.uuid4())
     username = Str(os.environ.get('USER','username'),config=True)
     exec_key = CUnicode('',config=True)
-    
     # not configurable:
     context = Instance('zmq.Context', (), {})
     session = Instance('IPython.zmq.parallel.streamsession.StreamSession')
-    loop = Instance('zmq.eventloop.ioloop.IOLoop')
+    loop = Instance('zmq.eventloop.ioloop.IOLoop', allow_none=False)
     def _loop_default(self):
         return IOLoop.instance()
     
+    
     def __init__(self, **kwargs):
         super(SessionFactory, self).__init__(**kwargs)
         

IPython/zmq/parallel/heartmonitor.py

@@ -13,6 +13,9 b' import zmq'
 from zmq.devices import ProcessDevice,ThreadDevice
 from zmq.eventloop import ioloop, zmqstream
 
+from IPython.utils.traitlets import Set, Instance, CFloat, Bool
+from factory import LoggingFactory
+
 class Heart(object):
     """A basic heart object for responding to a HeartMonitor.
     This is a simple wrapper with defaults for the most common
@@ -39,36 +42,35 b' class Heart(object):'
     def start(self):
         return self.device.start()
         
-class HeartMonitor(object):
+class HeartMonitor(LoggingFactory):
     """A basic HeartMonitor class
     pingstream: a PUB stream
     pongstream: an XREP stream
     period: the period of the heartbeat in milliseconds"""
-    loop=None
-    pingstream=None
-    pongstream=None
-    period=None
-    hearts=None
-    on_probation=None
-    last_ping=None
-    # debug=False
-    
-    def __init__(self, loop, pingstream, pongstream, period=1000):
-        self.loop = loop
-        self.period = period
-        
-        self.pingstream = pingstream
-        self.pongstream = pongstream
-        self.pongstream.on_recv(self.handle_pong)
         
-        self.hearts = set()
-        self.responses = set()
-        self.on_probation = set()
-        self.lifetime = 0
-        self.tic = time.time()
+    period=CFloat(1000, config=True) # in milliseconds
+    
+    pingstream=Instance('zmq.eventloop.zmqstream.ZMQStream')
+    pongstream=Instance('zmq.eventloop.zmqstream.ZMQStream')
+    loop = Instance('zmq.eventloop.ioloop.IOLoop')
+    def _loop_default(self):
+        return ioloop.IOLoop.instance()
+    debug=Bool(False)
         
-        self._new_handlers = set()
-        self._failure_handlers = set()
+    # not settable:
+    hearts=Set()
+    responses=Set()
+    on_probation=Set()
+    last_ping=CFloat(0)
+    _new_handlers = Set()
+    _failure_handlers = Set()
+    lifetime = CFloat(0)
+    tic = CFloat(0)
+    
+    def __init__(self, **kwargs):
+        super(HeartMonitor, self).__init__(**kwargs)
+        
+        self.pongstream.on_recv(self.handle_pong)
     
     def start(self):
         self.caller = ioloop.PeriodicCallback(self.beat, self.period, self.loop)
@@ -76,12 +78,12 b' class HeartMonitor(object):'
     
     def add_new_heart_handler(self, handler):
         """add a new handler for new hearts"""
-        logging.debug("heartbeat::new_heart_handler: %s"%handler)
+        self.log.debug("heartbeat::new_heart_handler: %s"%handler)
         self._new_handlers.add(handler)
         
     def add_heart_failure_handler(self, handler):
         """add a new handler for heart failure"""
-        logging.debug("heartbeat::new heart failure handler: %s"%handler)
+        self.log.debug("heartbeat::new heart failure handler: %s"%handler)
         self._failure_handlers.add(handler)
             
     def beat(self):
@@ -91,7 +93,7 b' class HeartMonitor(object):'
         toc = time.time()
         self.lifetime += toc-self.tic
         self.tic = toc
-        # logging.debug("heartbeat::%s"%self.lifetime)
+        # self.log.debug("heartbeat::%s"%self.lifetime)
         goodhearts = self.hearts.intersection(self.responses)
         missed_beats = self.hearts.difference(goodhearts)
         heartfailures = self.on_probation.intersection(missed_beats)
@@ -101,7 +103,7 b' class HeartMonitor(object):'
         self.on_probation = missed_beats.intersection(self.hearts)
         self.responses = set()
         # print self.on_probation, self.hearts
-        # logging.debug("heartbeat::beat %.3f, %i beating hearts"%(self.lifetime, len(self.hearts)))
+        # self.log.debug("heartbeat::beat %.3f, %i beating hearts"%(self.lifetime, len(self.hearts)))
         self.pingstream.send(str(self.lifetime))
     
     def handle_new_heart(self, heart):
@@ -109,7 +111,7 b' class HeartMonitor(object):'
             for handler in self._new_handlers:
                 handler(heart)
         else:
-            logging.info("heartbeat::yay, got new heart %s!"%heart)
+            self.log.info("heartbeat::yay, got new heart %s!"%heart)
         self.hearts.add(heart)
     
     def handle_heart_failure(self, heart):
@@ -118,10 +120,10 b' class HeartMonitor(object):'
                 try:
                     handler(heart)
                 except Exception as e:
-                    logging.error("heartbeat::Bad Handler! %s"%handler, exc_info=True)
+                    self.log.error("heartbeat::Bad Handler! %s"%handler, exc_info=True)
                     pass
         else:
-            logging.info("heartbeat::Heart %s failed :("%heart)
+            self.log.info("heartbeat::Heart %s failed :("%heart)
         self.hearts.remove(heart)
         
     
@@ -129,14 +131,14 b' class HeartMonitor(object):'
         "a heart just beat"
         if msg[1] == str(self.lifetime):
             delta = time.time()-self.tic
-            # logging.debug("heartbeat::heart %r took %.2f ms to respond"%(msg[0], 1000*delta))
+            # self.log.debug("heartbeat::heart %r took %.2f ms to respond"%(msg[0], 1000*delta))
             self.responses.add(msg[0])
         elif msg[1] == str(self.last_ping):
             delta = time.time()-self.tic + (self.lifetime-self.last_ping)
-            logging.warn("heartbeat::heart %r missed a beat, and took %.2f ms to respond"%(msg[0], 1000*delta))
+            self.log.warn("heartbeat::heart %r missed a beat, and took %.2f ms to respond"%(msg[0], 1000*delta))
             self.responses.add(msg[0])
         else:
-            logging.warn("heartbeat::got bad heartbeat (possibly old?): %s (current=%.3f)"%
+            self.log.warn("heartbeat::got bad heartbeat (possibly old?): %s (current=%.3f)"%
             (msg[1],self.lifetime))
 
 

IPython/zmq/parallel/hub.py

@@ -30,7 +30,7 b' from IPython.utils.traitlets import HasTraits, Instance, Int, Str, Dict, Set, Li'
 from IPython.utils.importstring import import_item
 
 from entry_point import select_random_ports
-from factory import RegistrationFactory
+from factory import RegistrationFactory, LoggingFactory
 
 from streamsession import Message, wrap_exception, ISO8601
 from heartmonitor import HeartMonitor
@@ -96,10 +96,6 b' class EngineConnector(HasTraits):'
     heartbeat=Str()
     pending=Set()
     
-    def __init__(self, **kwargs):
-        super(EngineConnector, self).__init__(**kwargs)
-        logging.info("engine::Engine Connected: %i"%self.id)
-
 class HubFactory(RegistrationFactory):
     """The Configurable for setting up a Hub."""
     
@@ -193,7 +189,7 b' class HubFactory(RegistrationFactory):'
     def start(self):
         assert self._constructed, "must be constructed by self.construct() first!"
         self.heartmonitor.start()
-        logging.info("Heartmonitor started")
+        self.log.info("Heartmonitor started")
     
     def construct_hub(self):
         """construct"""
@@ -206,10 +202,10 b' class HubFactory(RegistrationFactory):'
         # Registrar socket
         reg = ZMQStream(ctx.socket(zmq.XREP), loop)
         reg.bind(client_iface % self.regport)
-        logging.info("Hub listening on %s for registration."%(client_iface%self.regport))
+        self.log.info("Hub listening on %s for registration."%(client_iface%self.regport))
         if self.client_ip != self.engine_ip:
             reg.bind(engine_iface % self.regport)
-            logging.info("Hub listening on %s for registration."%(engine_iface%self.regport))
+            self.log.info("Hub listening on %s for registration."%(engine_iface%self.regport))
         
         ### Engine connections ###
 
@@ -218,8 +214,8 b' class HubFactory(RegistrationFactory):'
         hpub.bind(engine_iface % self.hb[0])
         hrep = ctx.socket(zmq.XREP)
         hrep.bind(engine_iface % self.hb[1])
-
-        self.heartmonitor = HeartMonitor(loop, ZMQStream(hpub,loop), ZMQStream(hrep,loop), self.ping)
+        self.heartmonitor = HeartMonitor(loop=loop, pingstream=ZMQStream(hpub,loop), pongstream=ZMQStream(hrep,loop), 
+                                period=self.ping, logname=self.log.name)
 
         ### Client connections ###
         # Clientele socket
@@ -259,14 +255,15 b' class HubFactory(RegistrationFactory):'
             'iopub' : client_iface%self.iopub[0],
             'notification': client_iface%self.notifier_port
             }
-        logging.debug("hub::Hub engine addrs: %s"%self.engine_addrs)
-        logging.debug("hub::Hub client addrs: %s"%self.client_addrs)
+        self.log.debug("hub::Hub engine addrs: %s"%self.engine_addrs)
+        self.log.debug("hub::Hub client addrs: %s"%self.client_addrs)
         self.hub = Hub(loop=loop, session=self.session, monitor=sub, heartmonitor=self.heartmonitor,
                 registrar=reg, clientele=c, notifier=n, db=self.db,
-                engine_addrs=self.engine_addrs, client_addrs=self.client_addrs)
+                engine_addrs=self.engine_addrs, client_addrs=self.client_addrs,
+                logname=self.log.name)
     
 
-class Hub(HasTraits):
+class Hub(LoggingFactory):
     """The IPython Controller Hub with 0MQ connections
     
     Parameters
@@ -371,7 +368,7 b' class Hub(HasTraits):'
                                 'connection_request': self.connection_request,
         }
         
-        logging.info("hub::created hub")
+        self.log.info("hub::created hub")
     
     @property
     def _next_id(self):
@@ -422,7 +419,7 b' class Hub(HasTraits):'
         try:
             msg = self.session.unpack_message(msg[1:], content=True)
         except:
-            logging.error("client::Invalid Message %s"%msg, exc_info=True)
+            self.log.error("client::Invalid Message %s"%msg, exc_info=True)
             return False
         
         msg_type = msg.get('msg_type', None)
@@ -439,15 +436,15 b' class Hub(HasTraits):'
     
     def dispatch_register_request(self, msg):
         """"""
-        logging.debug("registration::dispatch_register_request(%s)"%msg)
+        self.log.debug("registration::dispatch_register_request(%s)"%msg)
         idents,msg = self.session.feed_identities(msg)
         if not idents:
-            logging.error("Bad Queue Message: %s"%msg, exc_info=True)
+            self.log.error("Bad Queue Message: %s"%msg, exc_info=True)
             return
         try:
             msg = self.session.unpack_message(msg,content=True)
         except:
-            logging.error("registration::got bad registration message: %s"%msg, exc_info=True)
+            self.log.error("registration::got bad registration message: %s"%msg, exc_info=True)
             return
         
         msg_type = msg['msg_type']
@@ -455,38 +452,38 b' class Hub(HasTraits):'
         
         handler = self.registrar_handlers.get(msg_type, None)
         if handler is None:
-            logging.error("registration::got bad registration message: %s"%msg)
+            self.log.error("registration::got bad registration message: %s"%msg)
         else:
             handler(idents, msg)
     
     def dispatch_monitor_traffic(self, msg):
         """all ME and Task queue messages come through here, as well as
         IOPub traffic."""
-        logging.debug("monitor traffic: %s"%msg[:2])
+        self.log.debug("monitor traffic: %s"%msg[:2])
         switch = msg[0]
         idents, msg = self.session.feed_identities(msg[1:])
         if not idents:
-            logging.error("Bad Monitor Message: %s"%msg)
+            self.log.error("Bad Monitor Message: %s"%msg)
             return
         handler = self.monitor_handlers.get(switch, None)
         if handler is not None:
             handler(idents, msg)
         else:
-            logging.error("Invalid monitor topic: %s"%switch)
+            self.log.error("Invalid monitor topic: %s"%switch)
         
     
     def dispatch_client_msg(self, msg):
         """Route messages from clients"""
         idents, msg = self.session.feed_identities(msg)
         if not idents:
-            logging.error("Bad Client Message: %s"%msg)
+            self.log.error("Bad Client Message: %s"%msg)
             return
         client_id = idents[0]
         try:
             msg = self.session.unpack_message(msg, content=True)
         except:
             content = wrap_exception()
-            logging.error("Bad Client Message: %s"%msg, exc_info=True)
+            self.log.error("Bad Client Message: %s"%msg, exc_info=True)
             self.session.send(self.clientele, "hub_error", ident=client_id, 
                     content=content)
             return
@@ -494,13 +491,13 b' class Hub(HasTraits):'
         # print client_id, header, parent, content
         #switch on message type:
         msg_type = msg['msg_type']
-        logging.info("client:: client %s requested %s"%(client_id, msg_type))
+        self.log.info("client:: client %s requested %s"%(client_id, msg_type))
         handler = self.client_handlers.get(msg_type, None)
         try:
             assert handler is not None, "Bad Message Type: %s"%msg_type
         except:
             content = wrap_exception()
-            logging.error("Bad Message Type: %s"%msg_type, exc_info=True)
+            self.log.error("Bad Message Type: %s"%msg_type, exc_info=True)
             self.session.send(self.clientele, "hub_error", ident=client_id, 
                     content=content)
             return
@@ -521,9 +518,9 b' class Hub(HasTraits):'
         """handler to attach to heartbeater.
         Called when a new heart starts to beat.
         Triggers completion of registration."""
-        logging.debug("heartbeat::handle_new_heart(%r)"%heart)
+        self.log.debug("heartbeat::handle_new_heart(%r)"%heart)
         if heart not in self.incoming_registrations:
-            logging.info("heartbeat::ignoring new heart: %r"%heart)
+            self.log.info("heartbeat::ignoring new heart: %r"%heart)
         else:
             self.finish_registration(heart)
         
@@ -532,11 +529,11 b' class Hub(HasTraits):'
         """handler to attach to heartbeater.
         called when a previously registered heart fails to respond to beat request.
         triggers unregistration"""
-        logging.debug("heartbeat::handle_heart_failure(%r)"%heart)
+        self.log.debug("heartbeat::handle_heart_failure(%r)"%heart)
         eid = self.hearts.get(heart, None)
         queue = self.engines[eid].queue
         if eid is None:
-            logging.info("heartbeat::ignoring heart failure %r"%heart)
+            self.log.info("heartbeat::ignoring heart failure %r"%heart)
         else:
             self.unregister_engine(heart, dict(content=dict(id=eid, queue=queue)))
     
@@ -544,19 +541,19 b' class Hub(HasTraits):'
     
     def save_queue_request(self, idents, msg):
         if len(idents) < 2:
-            logging.error("invalid identity prefix: %s"%idents)
+            self.log.error("invalid identity prefix: %s"%idents)
             return
         queue_id, client_id = idents[:2]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
-            logging.error("queue::client %r sent invalid message to %r: %s"%(client_id, queue_id, msg), exc_info=True)
+            self.log.error("queue::client %r sent invalid message to %r: %s"%(client_id, queue_id, msg), exc_info=True)
             return
         
         eid = self.by_ident.get(queue_id, None)
         if eid is None:
-            logging.error("queue::target %r not registered"%queue_id)
-            logging.debug("queue::    valid are: %s"%(self.by_ident.keys()))
+            self.log.error("queue::target %r not registered"%queue_id)
+            self.log.debug("queue::    valid are: %s"%(self.by_ident.keys()))
             return
             
         header = msg['header']
@@ -573,21 +570,21 b' class Hub(HasTraits):'
     
     def save_queue_result(self, idents, msg):
         if len(idents) < 2:
-            logging.error("invalid identity prefix: %s"%idents)
+            self.log.error("invalid identity prefix: %s"%idents)
             return
             
         client_id, queue_id = idents[:2]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
-            logging.error("queue::engine %r sent invalid message to %r: %s"%(
+            self.log.error("queue::engine %r sent invalid message to %r: %s"%(
                     queue_id,client_id, msg), exc_info=True)
             return
         
         eid = self.by_ident.get(queue_id, None)
         if eid is None:
-            logging.error("queue::unknown engine %r is sending a reply: "%queue_id)
-            logging.debug("queue::       %s"%msg[2:])
+            self.log.error("queue::unknown engine %r is sending a reply: "%queue_id)
+            self.log.debug("queue::       %s"%msg[2:])
             return
         
         parent = msg['parent_header']
@@ -616,7 +613,7 b' class Hub(HasTraits):'
                 result['result_buffers'] = msg['buffers']
             self.db.update_record(msg_id, result)
         else:
-            logging.debug("queue:: unknown msg finished %s"%msg_id)
+            self.log.debug("queue:: unknown msg finished %s"%msg_id)
             
     #--------------------- Task Queue Traffic ------------------------------
     
@@ -627,7 +624,7 b' class Hub(HasTraits):'
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
-            logging.error("task::client %r sent invalid task message: %s"%(
+            self.log.error("task::client %r sent invalid task message: %s"%(
                     client_id, msg), exc_info=True)
             return
         record = init_record(msg)
@@ -646,7 +643,7 b' class Hub(HasTraits):'
         try:
             msg = self.session.unpack_message(msg, content=False)
         except:
-            logging.error("task::invalid task result message send to %r: %s"%(
+            self.log.error("task::invalid task result message send to %r: %s"%(
                     client_id, msg), exc_info=True)
             raise
             return
@@ -654,7 +651,7 b' class Hub(HasTraits):'
         parent = msg['parent_header']
         if not parent:
             # print msg
-            logging.warn("Task %r had no parent!"%msg)
+            self.log.warn("Task %r had no parent!"%msg)
             return
         msg_id = parent['msg_id']
         
@@ -687,13 +684,13 b' class Hub(HasTraits):'
             self.db.update_record(msg_id, result)
             
         else:
-            logging.debug("task::unknown task %s finished"%msg_id)
+            self.log.debug("task::unknown task %s finished"%msg_id)
     
     def save_task_destination(self, idents, msg):
         try:
             msg = self.session.unpack_message(msg, content=True)
         except:
-            logging.error("task::invalid task tracking message", exc_info=True)
+            self.log.error("task::invalid task tracking message", exc_info=True)
             return
         content = msg['content']
         print (content)
@@ -701,11 +698,11 b' class Hub(HasTraits):'
         engine_uuid = content['engine_id']
         eid = self.by_ident[engine_uuid]
         
-        logging.info("task::task %s arrived on %s"%(msg_id, eid))
+        self.log.info("task::task %s arrived on %s"%(msg_id, eid))
         # if msg_id in self.mia:
         #     self.mia.remove(msg_id)
         # else:
-        #     logging.debug("task::task %s not listed as MIA?!"%(msg_id))
+        #     self.log.debug("task::task %s not listed as MIA?!"%(msg_id))
         
         self.tasks[eid].append(msg_id)
         # self.pending[msg_id][1].update(received=datetime.now(),engine=(eid,engine_uuid))
@@ -726,12 +723,12 b' class Hub(HasTraits):'
         try:
             msg = self.session.unpack_message(msg, content=True)
         except:
-            logging.error("iopub::invalid IOPub message", exc_info=True)
+            self.log.error("iopub::invalid IOPub message", exc_info=True)
             return
         
         parent = msg['parent_header']
         if not parent:
-            logging.error("iopub::invalid IOPub message: %s"%msg)
+            self.log.error("iopub::invalid IOPub message: %s"%msg)
             return
         msg_id = parent['msg_id']
         msg_type = msg['msg_type']
@@ -741,7 +738,7 b' class Hub(HasTraits):'
         try:
             rec = self.db.get_record(msg_id)
         except:
-            logging.error("iopub::IOPub message has invalid parent", exc_info=True)
+            self.log.error("iopub::IOPub message has invalid parent", exc_info=True)
             return
         # stream
         d = {}
@@ -765,7 +762,7 b' class Hub(HasTraits):'
         
     def connection_request(self, client_id, msg):
         """Reply with connection addresses for clients."""
-        logging.info("client::client %s connected"%client_id)
+        self.log.info("client::client %s connected"%client_id)
         content = dict(status='ok')
         content.update(self.client_addrs)
         jsonable = {}
@@ -780,14 +777,14 b' class Hub(HasTraits):'
         try:
             queue = content['queue']
         except KeyError:
-            logging.error("registration::queue not specified", exc_info=True)
+            self.log.error("registration::queue not specified", exc_info=True)
             return
         heart = content.get('heartbeat', None)
         """register a new engine, and create the socket(s) necessary"""
         eid = self._next_id
         # print (eid, queue, reg, heart)
         
-        logging.debug("registration::register_engine(%i, %r, %r, %r)"%(eid, queue, reg, heart))
+        self.log.debug("registration::register_engine(%i, %r, %r, %r)"%(eid, queue, reg, heart))
         
         content = dict(id=eid,status='ok')
         content.update(self.engine_addrs)
@@ -797,12 +794,12 b' class Hub(HasTraits):'
                 raise KeyError("queue_id %r in use"%queue)
             except:
                 content = wrap_exception()
-                logging.error("queue_id %r in use"%queue, exc_info=True)
+                self.log.error("queue_id %r in use"%queue, exc_info=True)
         elif heart in self.hearts: # need to check unique hearts?
             try:
                 raise KeyError("heart_id %r in use"%heart)
             except:
-                logging.error("heart_id %r in use"%heart, exc_info=True)
+                self.log.error("heart_id %r in use"%heart, exc_info=True)
                 content = wrap_exception()
         else:
             for h, pack in self.incoming_registrations.iteritems():
@@ -810,14 +807,14 b' class Hub(HasTraits):'
                     try:
                         raise KeyError("heart_id %r in use"%heart)
                     except:
-                        logging.error("heart_id %r in use"%heart, exc_info=True)
+                        self.log.error("heart_id %r in use"%heart, exc_info=True)
                         content = wrap_exception()
                     break
                 elif queue == pack[1]:
                     try:
                         raise KeyError("queue_id %r in use"%queue)
                     except:
-                        logging.error("queue_id %r in use"%queue, exc_info=True)
+                        self.log.error("queue_id %r in use"%queue, exc_info=True)
                         content = wrap_exception()
                     break
         
@@ -836,7 +833,7 b' class Hub(HasTraits):'
                 dc.start()
                 self.incoming_registrations[heart] = (eid,queue,reg[0],dc)
         else:
-            logging.error("registration::registration %i failed: %s"%(eid, content['evalue']))
+            self.log.error("registration::registration %i failed: %s"%(eid, content['evalue']))
         return eid
     
     def unregister_engine(self, ident, msg):
@@ -844,9 +841,9 b' class Hub(HasTraits):'
         try:
             eid = msg['content']['id']
         except:
-            logging.error("registration::bad engine id for unregistration: %s"%ident, exc_info=True)
+            self.log.error("registration::bad engine id for unregistration: %s"%ident, exc_info=True)
             return
-        logging.info("registration::unregister_engine(%s)"%eid)
+        self.log.info("registration::unregister_engine(%s)"%eid)
         content=dict(id=eid, queue=self.engines[eid].queue)
         self.ids.remove(eid)
         self.keytable.pop(eid)
@@ -867,9 +864,9 b' class Hub(HasTraits):'
         try: 
             (eid,queue,reg,purge) = self.incoming_registrations.pop(heart)
         except KeyError:
-            logging.error("registration::tried to finish nonexistant registration", exc_info=True)
+            self.log.error("registration::tried to finish nonexistant registration", exc_info=True)
             return
-        logging.info("registration::finished registering engine %i:%r"%(eid,queue))
+        self.log.info("registration::finished registering engine %i:%r"%(eid,queue))
         if purge is not None:
             purge.stop()
         control = queue
@@ -885,11 +882,12 b' class Hub(HasTraits):'
         content = dict(id=eid, queue=self.engines[eid].queue)
         if self.notifier:
             self.session.send(self.notifier, "registration_notification", content=content)
+        self.log.info("engine::Engine Connected: %i"%eid)
     
     def _purge_stalled_registration(self, heart):
         if heart in self.incoming_registrations:
             eid = self.incoming_registrations.pop(heart)[0]
-            logging.info("registration::purging stalled registration: %i"%eid)
+            self.log.info("registration::purging stalled registration: %i"%eid)
         else:
             pass
             
@@ -910,7 +908,7 b' class Hub(HasTraits):'
         dc.start()
     
     def _shutdown(self):
-        logging.info("hub::hub shutting down.")
+        self.log.info("hub::hub shutting down.")
         time.sleep(0.1)
         sys.exit(0)
         

IPython/zmq/parallel/ipclusterapp.py

@@ -314,11 +314,12 b' class IPClusterApp(ApplicationWithClusterDir):'
         # and engine will be launched.
         el_class = import_item(config.Global.engine_launcher)
         self.engine_launcher = el_class(
-            work_dir=self.cluster_dir, config=config
+            work_dir=self.cluster_dir, config=config, logname=self.log.name
         )
         cl_class = import_item(config.Global.controller_launcher)
         self.controller_launcher = cl_class(
-            work_dir=self.cluster_dir, config=config
+            work_dir=self.cluster_dir, config=config,
+            logname=self.log.name
         )
 
         # Setup signals
@@ -348,11 +349,11 b' class IPClusterApp(ApplicationWithClusterDir):'
         return d
 
     def startup_message(self, r=None):
-        logging.info("IPython cluster: started")
+        self.log.info("IPython cluster: started")
         return r
         
     def start_controller(self, r=None):
-        # logging.info("In start_controller")
+        # self.log.info("In start_controller")
         config = self.master_config
         d = self.controller_launcher.start(
             cluster_dir=config.Global.cluster_dir
@@ -360,7 +361,7 b' class IPClusterApp(ApplicationWithClusterDir):'
         return d
             
     def start_engines(self, r=None):
-        # logging.info("In start_engines")
+        # self.log.info("In start_engines")
         config = self.master_config
         d = self.engine_launcher.start(
             config.Global.n,
@@ -369,12 +370,12 b' class IPClusterApp(ApplicationWithClusterDir):'
         return d
 
     def stop_controller(self, r=None):
-        # logging.info("In stop_controller")
+        # self.log.info("In stop_controller")
         if self.controller_launcher.running:
             return self.controller_launcher.stop()
 
     def stop_engines(self, r=None):
-        # logging.info("In stop_engines")
+        # self.log.info("In stop_engines")
         if self.engine_launcher.running:
             d = self.engine_launcher.stop()
             # d.addErrback(self.log_err)
@@ -383,16 +384,16 b' class IPClusterApp(ApplicationWithClusterDir):'
             return None
 
     def log_err(self, f):
-        logging.error(f.getTraceback())
+        self.log.error(f.getTraceback())
         return None
         
     def stop_launchers(self, r=None):
         if not self._stopping:
             self._stopping = True
             # if isinstance(r, failure.Failure):
-            #     logging.error('Unexpected error in ipcluster:')
-            #     logging.info(r.getTraceback())
-            logging.error("IPython cluster: stopping")
+            #     self.log.error('Unexpected error in ipcluster:')
+            #     self.log.info(r.getTraceback())
+            self.log.error("IPython cluster: stopping")
             # These return deferreds. We are not doing anything with them
             # but we are holding refs to them as a reminder that they 
             # do return deferreds.
@@ -462,7 +463,7 b' class IPClusterApp(ApplicationWithClusterDir):'
         try:
             self.loop.start()
         except:
-            logging.info("stopping...")
+            self.log.info("stopping...")
         self.remove_pid_file()
 
     def start_app_stop(self):

IPython/zmq/parallel/ipcontrollerapp.py

@@ -279,7 +279,7 b' class IPControllerApp(ApplicationWithClusterDir):'
             c.SessionFactory.exec_key = ''
         
         try:
-            self.factory = ControllerFactory(config=c)
+            self.factory = ControllerFactory(config=c, logname=self.log.name)
             self.start_logging()
             self.factory.construct()
         except:

IPython/zmq/parallel/ipengineapp.py

@@ -193,7 +193,7 b' class IPEngineApp(ApplicationWithClusterDir):'
         # Create the underlying shell class and EngineService
         # shell_class = import_item(self.master_config.Global.shell_class)
         try:
-            self.engine = EngineFactory(config=config)
+            self.engine = EngineFactory(config=config, logname=self.log.name)
         except:
             self.log.error("Couldn't start the Engine", exc_info=True)
             self.exit(1)

IPython/zmq/parallel/iploggerapp.py

@@ -107,7 +107,7 b' class IPLoggerApp(ApplicationWithClusterDir):'
         self.start_logging()
 
         try:
-            self.watcher = LogWatcher(config=self.master_config)
+            self.watcher = LogWatcher(config=self.master_config, logname=self.log.name)
         except:
             self.log.error("Couldn't start the LogWatcher", exc_info=True)
             self.exit(1)
@@ -115,6 +115,7 b' class IPLoggerApp(ApplicationWithClusterDir):'
 
     def start_app(self):
         try:
+            self.watcher.start()
             self.watcher.loop.start()
         except KeyboardInterrupt:
             self.log.critical("Logging Interrupted, shutting down...\n")

IPython/zmq/parallel/launcher.py

@@ -30,11 +30,12 b' from subprocess import Popen, PIPE'
 
 from zmq.eventloop import ioloop
 
-from IPython.config.configurable import Configurable
+# from IPython.config.configurable import Configurable
 from IPython.utils.traitlets import Str, Int, List, Unicode, Instance
 from IPython.utils.path import get_ipython_module_path
 from IPython.utils.process import find_cmd, pycmd2argv, FindCmdError
 
+from factory import LoggingFactory
 # from IPython.kernel.winhpcjob import (
 #     IPControllerTask, IPEngineTask,
 #     IPControllerJob, IPEngineSetJob
@@ -75,7 +76,7 b' class UnknownStatus(LauncherError):'
     pass
 
 
-class BaseLauncher(Configurable):
+class BaseLauncher(LoggingFactory):
     """An asbtraction for starting, stopping and signaling a process."""
 
     # In all of the launchers, the work_dir is where child processes will be
@@ -90,8 +91,8 b' class BaseLauncher(Configurable):'
     def _loop_default(self):
         return ioloop.IOLoop.instance()
 
-    def __init__(self, work_dir=u'.', config=None):
-        super(BaseLauncher, self).__init__(work_dir=work_dir, config=config)
+    def __init__(self, work_dir=u'.', config=None, **kwargs):
+        super(BaseLauncher, self).__init__(work_dir=work_dir, config=config, **kwargs)
         self.state = 'before' # can be before, running, after
         self.stop_callbacks = []
         self.start_data = None
@@ -163,7 +164,7 b' class BaseLauncher(Configurable):'
         a pass-through so it can be used as a callback.
         """
 
-        logging.info('Process %r started: %r' % (self.args[0], data))
+        self.log.info('Process %r started: %r' % (self.args[0], data))
         self.start_data = data
         self.state = 'running'
         return data
@@ -174,7 +175,7 b' class BaseLauncher(Configurable):'
         This logs the process stopping and sets the state to 'after'. Call
         this to trigger all the deferreds from :func:`observe_stop`."""
 
-        logging.info('Process %r stopped: %r' % (self.args[0], data))
+        self.log.info('Process %r stopped: %r' % (self.args[0], data))
         self.stop_data = data
         self.state = 'after'
         for i in range(len(self.stop_callbacks)):
@@ -212,9 +213,9 b' class LocalProcessLauncher(BaseLauncher):'
     cmd_and_args = List([])
     poll_frequency = Int(100) # in ms
 
-    def __init__(self, work_dir=u'.', config=None):
+    def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(LocalProcessLauncher, self).__init__(
-            work_dir=work_dir, config=config
+            work_dir=work_dir, config=config, **kwargs
         )
         self.process = None
         self.start_deferred = None
@@ -259,7 +260,7 b' class LocalProcessLauncher(BaseLauncher):'
         line = self.process.stdout.readline()
         # a stopped process will be readable but return empty strings
         if line:
-            logging.info(line[:-1])
+            self.log.info(line[:-1])
         else:
             self.poll()
     
@@ -267,7 +268,7 b' class LocalProcessLauncher(BaseLauncher):'
         line = self.process.stderr.readline()
         # a stopped process will be readable but return empty strings
         if line:
-            logging.error(line[:-1])
+            self.log.error(line[:-1])
         else:
             self.poll()
     
@@ -294,7 +295,7 b' class LocalControllerLauncher(LocalProcessLauncher):'
         """Start the controller by cluster_dir."""
         self.controller_args.extend(['--cluster-dir', cluster_dir])
         self.cluster_dir = unicode(cluster_dir)
-        logging.info("Starting LocalControllerLauncher: %r" % self.args)
+        self.log.info("Starting LocalControllerLauncher: %r" % self.args)
         return super(LocalControllerLauncher, self).start()
 
 
@@ -327,9 +328,9 b' class LocalEngineSetLauncher(BaseLauncher):'
     # launcher class
     launcher_class = LocalEngineLauncher
     
-    def __init__(self, work_dir=u'.', config=None):
+    def __init__(self, work_dir=u'.', config=None, **kwargs):
         super(LocalEngineSetLauncher, self).__init__(
-            work_dir=work_dir, config=config
+            work_dir=work_dir, config=config, **kwargs
         )
         self.launchers = {}
         self.stop_data = {}
@@ -339,14 +340,14 b' class LocalEngineSetLauncher(BaseLauncher):'
         self.cluster_dir = unicode(cluster_dir)
         dlist = []
         for i in range(n):
-            el = self.launcher_class(work_dir=self.work_dir, config=self.config)
+            el = self.launcher_class(work_dir=self.work_dir, config=self.config, logname=self.log.name)
             # Copy the engine args over to each engine launcher.
             import copy
             el.engine_args = copy.deepcopy(self.engine_args)
             el.on_stop(self._notice_engine_stopped)
             d = el.start(cluster_dir)
             if i==0:
-                logging.info("Starting LocalEngineSetLauncher: %r" % el.args)
+                self.log.info("Starting LocalEngineSetLauncher: %r" % el.args)
             self.launchers[i] = el
             dlist.append(d)
         self.notify_start(dlist)
@@ -431,7 +432,7 b' class MPIExecControllerLauncher(MPIExecLauncher):'
         """Start the controller by cluster_dir."""
         self.controller_args.extend(['--cluster-dir', cluster_dir])
         self.cluster_dir = unicode(cluster_dir)
-        logging.info("Starting MPIExecControllerLauncher: %r" % self.args)
+        self.log.info("Starting MPIExecControllerLauncher: %r" % self.args)
         return super(MPIExecControllerLauncher, self).start(1)
 
     def find_args(self):
@@ -453,7 +454,7 b' class MPIExecEngineSetLauncher(MPIExecLauncher):'
         self.engine_args.extend(['--cluster-dir', cluster_dir])
         self.cluster_dir = unicode(cluster_dir)
         self.n = n
-        logging.info('Starting MPIExecEngineSetLauncher: %r' % self.args)
+        self.log.info('Starting MPIExecEngineSetLauncher: %r' % self.args)
         return super(MPIExecEngineSetLauncher, self).start(n)
 
     def find_args(self):
@@ -572,7 +573,7 b' class SSHEngineSetLauncher(LocalEngineSetLauncher):'
 #         else:
 #             raise LauncherError("Job id couldn't be determined: %s" % output)
 #         self.job_id = job_id
-#         logging.info('Job started with job id: %r' % job_id)
+#         self.log.info('Job started with job id: %r' % job_id)
 #         return job_id
 # 
 #     @inlineCallbacks
@@ -584,7 +585,7 b' class SSHEngineSetLauncher(LocalEngineSetLauncher):'
 #             '/jobfile:%s' % self.job_file,
 #             '/scheduler:%s' % self.scheduler
 #         ]
-#         logging.info("Starting Win HPC Job: %s" % (self.job_cmd + ' ' + ' '.join(args),))
+#         self.log.info("Starting Win HPC Job: %s" % (self.job_cmd + ' ' + ' '.join(args),))
 #         # Twisted will raise DeprecationWarnings if we try to pass unicode to this
 #         output = yield getProcessOutput(str(self.job_cmd),
 #             [str(a) for a in args],
@@ -602,7 +603,7 b' class SSHEngineSetLauncher(LocalEngineSetLauncher):'
 #             self.job_id,
 #             '/scheduler:%s' % self.scheduler
 #         ]
-#         logging.info("Stopping Win HPC Job: %s" % (self.job_cmd + ' ' + ' '.join(args),))
+#         self.log.info("Stopping Win HPC Job: %s" % (self.job_cmd + ' ' + ' '.join(args),))
 #         try:
 #             # Twisted will raise DeprecationWarnings if we try to pass unicode to this
 #             output = yield getProcessOutput(str(self.job_cmd),
@@ -633,7 +634,7 b' class SSHEngineSetLauncher(LocalEngineSetLauncher):'
 #         t.controller_args.extend(self.extra_args)
 #         job.add_task(t)
 # 
-#         logging.info("Writing job description file: %s" % self.job_file)
+#         self.log.info("Writing job description file: %s" % self.job_file)
 #         job.write(self.job_file)
 # 
 #     @property
@@ -665,7 +666,7 b' class SSHEngineSetLauncher(LocalEngineSetLauncher):'
 #             t.engine_args.extend(self.extra_args)
 #             job.add_task(t)
 # 
-#         logging.info("Writing job description file: %s" % self.job_file)
+#         self.log.info("Writing job description file: %s" % self.job_file)
 #         job.write(self.job_file)
 # 
 #     @property
@@ -729,14 +730,14 b' class SSHEngineSetLauncher(LocalEngineSetLauncher):'
 #         else:
 #             raise LauncherError("Job id couldn't be determined: %s" % output)
 #         self.job_id = job_id
-#         logging.info('Job started with job id: %r' % job_id)
+#         self.log.info('Job started with job id: %r' % job_id)
 #         return job_id
 # 
 #     def write_batch_script(self, n):
 #         """Instantiate and write the batch script to the work_dir."""
 #         self.context['n'] = n
 #         script_as_string = Itpl.itplns(self.batch_template, self.context)
-#         logging.info('Writing instantiated batch script: %s' % self.batch_file)
+#         self.log.info('Writing instantiated batch script: %s' % self.batch_file)
 #         f = open(self.batch_file, 'w')
 #         f.write(script_as_string)
 #         f.close()
@@ -783,7 +784,7 b' class SSHEngineSetLauncher(LocalEngineSetLauncher):'
 #         # ${cluster_dir}
 #         self.context['cluster_dir'] = cluster_dir
 #         self.cluster_dir = unicode(cluster_dir)
-#         logging.info("Starting PBSControllerLauncher: %r" % self.args)
+#         self.log.info("Starting PBSControllerLauncher: %r" % self.args)
 #         return super(PBSControllerLauncher, self).start(1)
 # 
 # 
@@ -795,7 +796,7 b' class SSHEngineSetLauncher(LocalEngineSetLauncher):'
 #         """Start n engines by profile or cluster_dir."""
 #         self.program_args.extend(['--cluster-dir', cluster_dir])
 #         self.cluster_dir = unicode(cluster_dir)
-#         logging.info('Starting PBSEngineSetLauncher: %r' % self.args)
+#         self.log.info('Starting PBSEngineSetLauncher: %r' % self.args)
 #         return super(PBSEngineSetLauncher, self).start(n)
 
 
@@ -819,6 +820,6 b' class IPClusterLauncher(LocalProcessLauncher):'
             ['-n', repr(self.ipcluster_n)] + self.ipcluster_args
 
     def start(self):
-        logging.info("Starting ipcluster: %r" % self.args)
+        self.log.info("Starting ipcluster: %r" % self.args)
         return super(IPClusterLauncher, self).start()
 

IPython/zmq/parallel/logwatcher.py

@@ -18,15 +18,16 b' import logging'
 
 import zmq
 from zmq.eventloop import ioloop, zmqstream
-from IPython.config.configurable import Configurable
 from IPython.utils.traitlets import Int, Str, Instance, List
 
+from factory import LoggingFactory
+
 #-----------------------------------------------------------------------------
 # Classes
 #-----------------------------------------------------------------------------
 
 
-class LogWatcher(Configurable):
+class LogWatcher(LoggingFactory):
     """A simple class that receives messages on a SUB socket, as published
     by subclasses of `zmq.log.handlers.PUBHandler`, and logs them itself.
     
@@ -43,21 +44,25 b' class LogWatcher(Configurable):'
     def _loop_default(self):
         return ioloop.IOLoop.instance()
     
-    def __init__(self, config=None):
-        super(LogWatcher, self).__init__(config=config)
+    def __init__(self, **kwargs):
+        super(LogWatcher, self).__init__(**kwargs)
         s = self.context.socket(zmq.SUB)
         s.bind(self.url)
         self.stream = zmqstream.ZMQStream(s, self.loop)
         self.subscribe()
         self.on_trait_change(self.subscribe, 'topics')
         
+    def start(self):
         self.stream.on_recv(self.log_message)
     
+    def stop(self):
+        self.stream.stop_on_recv()
+    
     def subscribe(self):
         """Update our SUB socket's subscriptions."""
         self.stream.setsockopt(zmq.UNSUBSCRIBE, '')
         for topic in self.topics:
-            logging.debug("Subscribing to: %r"%topic)
+            self.log.debug("Subscribing to: %r"%topic)
             self.stream.setsockopt(zmq.SUBSCRIBE, topic)
     
     def _extract_level(self, topic_str):
@@ -79,7 +84,7 b' class LogWatcher(Configurable):'
     def log_message(self, raw):
         """receive and parse a message, then log it."""
         if len(raw) != 2 or '.' not in raw[0]:
-            logging.error("Invalid log message: %s"%raw)
+            self.log.error("Invalid log message: %s"%raw)
             return
         else:
             topic, msg = raw

IPython/zmq/parallel/scheduler.py

@@ -10,8 +10,9 b' Python Scheduler exists.'
 #----------------------------------------------------------------------
 
 from __future__ import print_function
-from random import randint,random
+import sys
 import logging
+from random import randint,random
 from types import FunctionType
 
 try:
@@ -24,7 +25,7 b' from zmq.eventloop import ioloop, zmqstream'
 
 # local imports
 from IPython.external.decorator import decorator
-from IPython.config.configurable import Configurable
+# from IPython.config.configurable import Configurable
 from IPython.utils.traitlets import Instance, Dict, List, Set
 
 import error
@@ -32,12 +33,13 b' from client import Client'
 from dependency import Dependency
 import streamsession as ss
 from entry_point import connect_logger, local_logger
+from factory import LoggingFactory
 
 
 @decorator
 def logged(f,self,*args,**kwargs):
     # print ("#--------------------")
-    logging.debug("scheduler::%s(*%s,**%s)"%(f.func_name, args, kwargs))
+    self.log.debug("scheduler::%s(*%s,**%s)"%(f.func_name, args, kwargs))
     # print ("#--")
     return f(self,*args, **kwargs)
 
@@ -108,7 +110,7 b' def leastload(loads):'
 # store empty default dependency:
 MET = Dependency([])
 
-class TaskScheduler(Configurable):
+class TaskScheduler(LoggingFactory):
     """Python TaskScheduler object.
     
     This is the simplest object that supports msg_id based
@@ -153,7 +155,7 b' class TaskScheduler(Configurable):'
             unregistration_notification = self._unregister_engine
         )
         self.notifier_stream.on_recv(self.dispatch_notification)
-        logging.info("Scheduler started...%r"%self)
+        self.log.info("Scheduler started...%r"%self)
     
     def resume_receiving(self):
         """Resume accepting jobs."""
@@ -180,7 +182,7 b' class TaskScheduler(Configurable):'
             try:
                 handler(str(msg['content']['queue']))
             except KeyError:
-                logging.error("task::Invalid notification msg: %s"%msg)
+                self.log.error("task::Invalid notification msg: %s"%msg)
     
     @logged
     def _register_engine(self, uid):
@@ -236,7 +238,7 b' class TaskScheduler(Configurable):'
         try:
             idents, msg = self.session.feed_identities(raw_msg, copy=False)
         except Exception as e:
-            logging.error("task::Invaid msg: %s"%msg)
+            self.log.error("task::Invaid msg: %s"%msg)
             return
         
         # send to monitor
@@ -277,7 +279,7 b' class TaskScheduler(Configurable):'
     def fail_unreachable(self, msg_id):
         """a message has become unreachable"""
         if msg_id not in self.depending:
-            logging.error("msg %r already failed!"%msg_id)
+            self.log.error("msg %r already failed!"%msg_id)
             return
         raw_msg, after, follow = self.depending.pop(msg_id)
         for mid in follow.union(after):
@@ -369,7 +371,7 b' class TaskScheduler(Configurable):'
         try:
             idents,msg = self.session.feed_identities(raw_msg, copy=False)
         except Exception as e:
-            logging.error("task::Invaid result: %s"%msg)
+            self.log.error("task::Invaid result: %s"%msg)
             return
         msg = self.session.unpack_message(msg, content=False, copy=False)
         header = msg['header']
@@ -470,7 +472,7 b' class TaskScheduler(Configurable):'
     
 
 
-def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, log_addr=None, loglevel=logging.DEBUG, scheme='weighted'):
+def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, logname='ZMQ', log_addr=None, loglevel=logging.DEBUG, scheme='weighted'):
     from zmq.eventloop import ioloop
     from zmq.eventloop.zmqstream import ZMQStream
     
@@ -490,13 +492,13 b' def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, log_addr=None, logle'
     scheme = globals().get(scheme, None)
     # setup logging
     if log_addr:
-        connect_logger(ctx, log_addr, root="scheduler", loglevel=loglevel)
+        connect_logger(logname, ctx, log_addr, root="scheduler", loglevel=loglevel)
     else:
-        local_logger(loglevel)
+        local_logger(logname, loglevel)
     
     scheduler = TaskScheduler(client_stream=ins, engine_stream=outs,
                             mon_stream=mons,notifier_stream=nots,
-                            scheme=scheme,io_loop=loop)
+                            scheme=scheme,io_loop=loop, logname=logname)
     
     try:
         loop.start()

IPython/zmq/parallel/streamkernel.py

@@ -110,7 +110,7 b' class Kernel(SessionFactory):'
         s = _Passer()
         content = dict(silent=True, user_variable=[],user_expressions=[])
         for line in self.exec_lines:
-            logging.debug("executing initialization: %s"%line)
+            self.log.debug("executing initialization: %s"%line)
             content.update({'code':line})
             msg = self.session.msg('execute_request', content)
             self.execute_request(s, [], msg)
@@ -139,8 +139,8 b' class Kernel(SessionFactory):'
                 
                 # assert self.reply_socketly_socket.rcvmore(), "Unexpected missing message part."
                 # msg = self.reply_socket.recv_json()
-            logging.info("Aborting:")
-            logging.info(str(msg))
+            self.log.info("Aborting:")
+            self.log.info(str(msg))
             msg_type = msg['msg_type']
             reply_type = msg_type.split('_')[0] + '_reply'
             # reply_msg = self.session.msg(reply_type, {'status' : 'aborted'}, msg)
@@ -148,7 +148,7 b' class Kernel(SessionFactory):'
             # self.reply_socket.send_json(reply_msg)
             reply_msg = self.session.send(stream, reply_type, 
                         content={'status' : 'aborted'}, parent=msg, ident=idents)[0]
-            logging.debug(str(reply_msg))
+            self.log.debug(str(reply_msg))
             # We need to wait a bit for requests to come in. This can probably
             # be set shorter for true asynchronous clients.
             time.sleep(0.05)
@@ -166,7 +166,7 b' class Kernel(SessionFactory):'
         content = dict(status='ok')
         reply_msg = self.session.send(stream, 'abort_reply', content=content, 
                 parent=parent, ident=ident)[0]
-        logging.debug(str(reply_msg))
+        self.log.debug(str(reply_msg))
     
     def shutdown_request(self, stream, ident, parent):
         """kill ourself.  This should really be handled in an external process"""
@@ -191,7 +191,7 b' class Kernel(SessionFactory):'
         try:
             msg = self.session.unpack_message(msg, content=True, copy=False)
         except:
-            logging.error("Invalid Message", exc_info=True)
+            self.log.error("Invalid Message", exc_info=True)
             return
         
         header = msg['header']
@@ -199,7 +199,7 b' class Kernel(SessionFactory):'
         
         handler = self.control_handlers.get(msg['msg_type'], None)
         if handler is None:
-            logging.error("UNKNOWN CONTROL MESSAGE TYPE: %r"%msg['msg_type'])
+            self.log.error("UNKNOWN CONTROL MESSAGE TYPE: %r"%msg['msg_type'])
         else:
             handler(self.control_stream, idents, msg)
     
@@ -240,11 +240,11 b' class Kernel(SessionFactory):'
         self._initial_exec_lines()
     
     def execute_request(self, stream, ident, parent):
-        logging.debug('execute request %s'%parent)
+        self.log.debug('execute request %s'%parent)
         try:
             code = parent[u'content'][u'code']
         except:
-            logging.error("Got bad msg: %s"%parent, exc_info=True)
+            self.log.error("Got bad msg: %s"%parent, exc_info=True)
             return
         self.session.send(self.iopub_stream, u'pyin', {u'code':code},parent=parent,
                             ident='%s.pyin'%self.prefix)
@@ -268,7 +268,7 b' class Kernel(SessionFactory):'
         
         reply_msg = self.session.send(stream, u'execute_reply', reply_content, parent=parent, 
                     ident=ident, subheader = dict(started=started))
-        logging.debug(str(reply_msg))
+        self.log.debug(str(reply_msg))
         if reply_msg['content']['status'] == u'error':
             self.abort_queues()
 
@@ -290,7 +290,7 b' class Kernel(SessionFactory):'
             msg_id = parent['header']['msg_id']
             bound = content.get('bound', False)
         except:
-            logging.error("Got bad msg: %s"%parent, exc_info=True)
+            self.log.error("Got bad msg: %s"%parent, exc_info=True)
             return
         # pyin_msg = self.session.msg(u'pyin',{u'code':code}, parent=parent)
         # self.iopub_stream.send(pyin_msg)
@@ -364,7 +364,7 b' class Kernel(SessionFactory):'
         try:
             msg = self.session.unpack_message(msg, content=True, copy=False)
         except:
-            logging.error("Invalid Message", exc_info=True)
+            self.log.error("Invalid Message", exc_info=True)
             return
             
         
@@ -379,7 +379,7 b' class Kernel(SessionFactory):'
             return
         handler = self.shell_handlers.get(msg['msg_type'], None)
         if handler is None:
-            logging.error("UNKNOWN MESSAGE TYPE: %r"%msg['msg_type'])
+            self.log.error("UNKNOWN MESSAGE TYPE: %r"%msg['msg_type'])
         else:
             handler(stream, idents, msg)
     

IPython/zmq/tunnel.py

@@ -158,7 +158,7 b" def openssh_tunnel(lport, rport, server, remoteip='127.0.0.1', keyfile=None, pas"
     ssh="ssh "
     if keyfile:
         ssh += "-i " + keyfile 
-    cmd = ssh + " -f -L 127.0.0.1:%i:127.0.0.1:%i %s sleep %i"%(lport, rport, server, timeout)
+    cmd = ssh + " -f -L 127.0.0.1:%i:%s:%i %s sleep %i"%(lport, remoteip, rport, server, timeout)
     tunnel = pexpect.spawn(cmd)
     failed = False
     while True: