forked from ipython/ipython
-
Notifications
You must be signed in to change notification settings - Fork 0
/
logwatcher.py
114 lines (87 loc) · 3.68 KB
/
logwatcher.py
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
"""
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))