-
-
Notifications
You must be signed in to change notification settings - Fork 4.4k
/
logwatcher.py
98 lines (76 loc) · 3.26 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
#!/usr/bin/env python
"""A simple logger object that consolidates messages incoming from ipclusterz processes."""
#-----------------------------------------------------------------------------
# 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.utils.traitlets import Int, Str, Instance, List
from .factory import LoggingFactory
#-----------------------------------------------------------------------------
# Classes
#-----------------------------------------------------------------------------
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.
This can subscribe to multiple topics, but defaults to all topics.
"""
# configurables
topics = List([''], config=True)
url = Str('tcp://127.0.0.1:20202', config=True)
# internals
context = Instance(zmq.Context, (), {})
stream = Instance('zmq.eventloop.zmqstream.ZMQStream')
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, '')
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]
logging.log(level, "[%s] %s" % (topic, msg))