##// END OF EJS Templates
Merge pull request #8296 from minrk/rm-kernel...
Merge pull request #8296 from minrk/rm-kernel remove ipython_kernel

File last commit:

r20945:e4060ed8
r21224:f9254a22 merge
Show More
logwatcher.py
117 lines | 3.8 KiB | text/x-python | PythonLexer
MinRK
update recently changed modules with Authors in docstring
r4018 """
A simple logger object that consolidates messages incoming from ipcluster processes.
Authors:
* MinRK
"""
MinRK
Refactor newparallel to use Config system...
r3604
#-----------------------------------------------------------------------------
# Copyright (C) 2011 The IPython Development Team
#
# Distributed under the terms of the BSD License. The full license is in
# the file COPYING, distributed as part of this software.
#-----------------------------------------------------------------------------
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
import logging
MinRK
resort imports in a cleaner order
r3631 import sys
MinRK
Refactor newparallel to use Config system...
r3604
import zmq
from zmq.eventloop import ioloop, zmqstream
MinRK
resort imports in a cleaner order
r3631
MinRK
add LoggingConfigurable base class
r4016 from IPython.config.configurable import LoggingConfigurable
MinRK
avoid executing code in utils.localinterfaces at import time...
r12591 from IPython.utils.localinterfaces import localhost
MinRK
cleanup parallel traits...
r3988 from IPython.utils.traitlets import Int, Unicode, Instance, List
MinRK
Refactor newparallel to use Config system...
r3604
#-----------------------------------------------------------------------------
# Classes
#-----------------------------------------------------------------------------
MinRK
add LoggingConfigurable base class
r4016 class LogWatcher(LoggingConfigurable):
MinRK
Refactor newparallel to use Config system...
r3604 """A simple class that receives messages on a SUB socket, as published
by subclasses of `zmq.log.handlers.PUBHandler`, and logs them itself.
This can subscribe to multiple topics, but defaults to all topics.
"""
MinRK
reorganize Factory classes to follow relocation of Session object
r4007
MinRK
Refactor newparallel to use Config system...
r3604 # configurables
MinRK
re-enable log forwarding and iplogger
r3989 topics = List([''], config=True,
help="The ZMQ topics to subscribe to. Default is to subscribe to all messages")
MinRK
avoid executing code in utils.localinterfaces at import time...
r12591 url = Unicode(config=True,
MinRK
re-enable log forwarding and iplogger
r3989 help="ZMQ url on which to listen for log messages")
MinRK
avoid executing code in utils.localinterfaces at import time...
r12591 def _url_default(self):
return 'tcp://%s:20202' % localhost()
MinRK
Refactor newparallel to use Config system...
r3604
# internals
Sylvain Corlay
allow_none=False by default for Type and Instance
r20940 stream = Instance('zmq.eventloop.zmqstream.ZMQStream', allow_none=True)
MinRK
re-enable log forwarding and iplogger
r3989
Sylvain Corlay
remove unnecessary allow_none=True
r20943 context = Instance(zmq.Context)
MinRK
re-enable log forwarding and iplogger
r3989 def _context_default(self):
return zmq.Context.instance()
Sylvain Corlay
unnecessary allow_none=True
r20945 loop = Instance(zmq.eventloop.ioloop.IOLoop)
MinRK
Refactor newparallel to use Config system...
r3604 def _loop_default(self):
return ioloop.IOLoop.instance()
MinRK
rework logging connections
r3610 def __init__(self, **kwargs):
super(LogWatcher, self).__init__(**kwargs)
MinRK
Refactor newparallel to use Config system...
r3604 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')
MinRK
rework logging connections
r3610
def start(self):
MinRK
Refactor newparallel to use Config system...
r3604 self.stream.on_recv(self.log_message)
MinRK
rework logging connections
r3610 def stop(self):
self.stream.stop_on_recv()
MinRK
Refactor newparallel to use Config system...
r3604 def subscribe(self):
"""Update our SUB socket's subscriptions."""
self.stream.setsockopt(zmq.UNSUBSCRIBE, '')
MinRK
re-enable log forwarding and iplogger
r3989 if '' in self.topics:
self.log.debug("Subscribing to: everything")
self.stream.setsockopt(zmq.SUBSCRIBE, '')
else:
for topic in self.topics:
self.log.debug("Subscribing to: %r"%(topic))
self.stream.setsockopt(zmq.SUBSCRIBE, topic)
MinRK
Refactor newparallel to use Config system...
r3604
def _extract_level(self, topic_str):
"""Turn 'engine.0.INFO.extra' into (logging.INFO, 'engine.0.extra')"""
topics = topic_str.split('.')
for idx,t in enumerate(topics):
level = getattr(logging, t, None)
if level is not None:
break
if level is None:
level = logging.INFO
else:
topics.pop(idx)
return level, '.'.join(topics)
def log_message(self, raw):
"""receive and parse a message, then log it."""
if len(raw) != 2 or '.' not in raw[0]:
MinRK
rework logging connections
r3610 self.log.error("Invalid log message: %s"%raw)
MinRK
Refactor newparallel to use Config system...
r3604 return
else:
topic, msg = raw
# don't newline, since log messages always newline:
topic,level_name = topic.rsplit('.',1)
level,topic = self._extract_level(topic)
if msg[-1] == '\n':
msg = msg[:-1]
MinRK
re-enable log forwarding and iplogger
r3989 self.log.log(level, "[%s] %s" % (topic, msg))
MinRK
Refactor newparallel to use Config system...
r3604