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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
"""The Landscape monitor plugin system."""
import os
from logging import info
from twisted.internet.defer import succeed
from landscape.lib.dbus_util import method
from landscape.lib.log import log_failure
from landscape.log import format_object
from landscape.plugin import BrokerClientPluginRegistry, Plugin, BrokerPlugin
BUS_NAME = "com.canonical.landscape.Monitor"
OBJECT_PATH = "/com/canonical/landscape/Monitor"
IFACE_NAME = BUS_NAME
class MonitorDBusObject(BrokerPlugin):
"""A DBUS object which provides an interface to the Landscape Monitor."""
bus_name = BUS_NAME
object_path = OBJECT_PATH
ping = method(IFACE_NAME)(BrokerPlugin.ping)
exit = method(IFACE_NAME)(BrokerPlugin.exit)
message = method(IFACE_NAME)(BrokerPlugin.message)
class MonitorPluginRegistry(BrokerClientPluginRegistry):
"""The central point of integration in the Landscape monitor."""
def __init__(self, broker, reactor, config, bus,
persist, persist_filename=None,
step_size=5*60):
super(MonitorPluginRegistry, self).__init__(broker)
self.reactor = reactor
self.config = config
self.persist = persist
self.persist_filename = persist_filename
if persist_filename and os.path.exists(persist_filename):
self.persist.load(persist_filename)
self._plugins = []
self.step_size = step_size
self.bus = bus
def flush(self):
"""Flush data to disk."""
if self.persist_filename:
self.persist.save(self.persist_filename)
def exchange(self):
"""Call C{exchange} on all plugins."""
super(MonitorPluginRegistry, self).exchange()
self.flush()
class MonitorPlugin(Plugin):
"""
@cvar persist_name: If specified as a string, a C{_persist} attribute
will be available after registration.
XXX This class is no longer very useful and should be cleaned out
at some point.
"""
persist_name = None
def register(self, registry):
super(MonitorPlugin, self).register(registry)
if self.persist_name is not None:
self._persist = registry.persist.root_at(self.persist_name)
def call_on_accepted(self, type, callable, *args, **kwargs):
def acceptance_changed(acceptance):
if acceptance:
return callable(*args, **kwargs)
self.registry.reactor.call_on(("message-type-acceptance-changed",
type), acceptance_changed)
class DataWatcher(MonitorPlugin):
"""
A utility for plugins which send data to the Landscape server
which does not constantly change. New messages will only be sent
when the result of get_data() has changed since the last time it
was called.
Subclasses should provide a get_data method, and message_type,
message_key, and persist_name class attributes.
"""
message_type = None
message_key = None
def get_message(self):
"""
Construct a message with the latest data, or None, if the data
has not changed since the last call.
"""
data = self.get_data()
if self._persist.get("data") != data:
self._persist.set("data", data)
return {"type": self.message_type, self.message_key: data}
def send_message(self, urgent):
message = self.get_message()
if message is not None:
info("Queueing a message with updated data watcher info "
"for %s.", format_object(self))
result = self.registry.broker.send_message(message, urgent=urgent)
def persist_data(message_id):
self.persist_data()
result.addCallback(persist_data)
result.addErrback(log_failure)
return result
return succeed(None)
def persist_data(self):
"""
Sub-classes that need to defer the saving of persistent data
should override this method.
"""
pass
def exchange(self, urgent=False):
"""
Conditionally add a message to the message store if new data
is available.
"""
return self.registry.broker.call_if_accepted(self.message_type,
self.send_message, urgent)
|