1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114
|
"""
A simple logger object that consolidates messages incoming from ipcluster processes.
Authors:
* MinRK
"""
#-----------------------------------------------------------------------------
# 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
import sys
import zmq
from zmq.eventloop import ioloop, zmqstream
from IPython.config.configurable import LoggingConfigurable
from IPython.utils.traitlets import Int, Unicode, Instance, List
#-----------------------------------------------------------------------------
# Classes
#-----------------------------------------------------------------------------
class LogWatcher(LoggingConfigurable):
"""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.
"""
# configurables
topics = List([''], config=True,
help="The ZMQ topics to subscribe to. Default is to subscribe to all messages")
url = Unicode('tcp://127.0.0.1:20202', config=True,
help="ZMQ url on which to listen for log messages")
# internals
stream = Instance('zmq.eventloop.zmqstream.ZMQStream')
context = Instance(zmq.Context)
def _context_default(self):
return zmq.Context.instance()
loop = Instance(zmq.eventloop.ioloop.IOLoop)
def _loop_default(self):
return ioloop.IOLoop.instance()
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, '')
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)
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]:
self.log.error("Invalid log message: %s"%raw)
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]
self.log.log(level, "[%s] %s" % (topic, msg))
|