1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23 """
24 ClusterShell Propagation module. Use the topology tree to send commands
25 through gateways and gather results.
26 """
27
28 from collections import deque
29 import logging
30
31 from ClusterShell.Defaults import DEFAULTS
32 from ClusterShell.NodeSet import NodeSet
33 from ClusterShell.Communication import Channel
34 from ClusterShell.Communication import ControlMessage, StdOutMessage
35 from ClusterShell.Communication import StdErrMessage, RetcodeMessage
36 from ClusterShell.Communication import StartMessage, EndMessage
37 from ClusterShell.Communication import RoutedMessageBase, ErrorMessage
38 from ClusterShell.Communication import ConfigurationMessage, TimeoutMessage
39 from ClusterShell.Topology import TopologyError
40
41
43 """error raised on invalid conditions during routing operations"""
44
46 """performs routes resolving operations within a propagation tree.
47 This object provides a next_hop method, that will look for the best
48 directly connected node to use to forward a message to a remote
49 node.
50
51 Upon instanciation, the router will parse the topology tree to
52 generate its routing table.
53 """
54 - def __init__(self, root, topology, fanout=0):
55 self.root = root
56 self.topology = topology
57 self.fanout = fanout
58 self.nodes_fanin = {}
59 self.table = None
60
61 self.table_generate(root, topology)
62 self._unreachable_hosts = NodeSet()
63
65 """The router relies on a routing table. The keys are the
66 destination nodes and the values are the next hop gateways to
67 use to reach these nodes.
68 """
69 self.table = {}
70 try:
71 root_group = topology.find_nodegroup(root)
72 except TopologyError:
73 msgfmt = "Invalid root or gateway node: %s"
74 raise RouteResolvingError(msgfmt % root)
75
76 for group in root_group.children():
77 self.table[group.nodeset] = NodeSet()
78 stack = [group]
79 while len(stack) > 0:
80 curr = stack.pop()
81 self.table[group.nodeset].add(curr.children_ns())
82 stack += curr.children()
83
84
85 self.table = dict((v, k) for k, v in self.table.iteritems())
86
88 """dispatch nodes from a target nodeset to the directly
89 connected gateways.
90
91 The method acts as an iterator, returning a gateway and the
92 associated hosts. It should provide a rather good load balancing
93 between the gateways.
94 """
95
96
97
98
99
100
101
102
103
104 for network in self.table.iterkeys():
105 dst_inter = network & dst
106 dst.difference_update(dst_inter)
107 for host in dst_inter.nsiter():
108 yield self.next_hop(host), host
109
110
111 if dst:
112 yield dst, dst
113
115 """perform the next hop resolution. If several hops are
116 available, then, the one with the least number of current jobs
117 will be returned
118 """
119 if dst in self._unreachable_hosts:
120 raise RouteResolvingError(
121 'Invalid destination: %s, host is unreachable' % dst)
122
123
124 if self.root == dst:
125 raise RouteResolvingError(
126 'Invalid resolution request: %s -> %s' % (self.root, dst))
127
128
129
130
131
132
133
134
135
136
137 for network, nexthops in self.table.iteritems():
138
139 if dst in network:
140 res = self._best_next_hop(nexthops)
141 if res is None:
142 raise RouteResolvingError('No route available to %s' % \
143 str(dst))
144 self.nodes_fanin[res] += len(dst)
145 return res
146
147
148 if dst in nexthops:
149 return dst
150
151 raise RouteResolvingError(
152 'No route from %s to host %s' % (self.root, dst))
153
155 """mark node dst as unreachable and don't advertise routes
156 through it anymore. The cache will be updated only when
157 necessary to avoid performing expensive traversals.
158 """
159
160
161 self._unreachable_hosts.add(dst)
162
164 """find out a good next hop gateway"""
165 backup = None
166 backup_connections = 1e400
167
168 candidates = candidates.difference(self._unreachable_hosts)
169
170 for host in candidates:
171
172
173 connections = self.nodes_fanin.setdefault(host, 0)
174
175
176
177
178 if backup_connections > connections:
179 backup = host
180 backup_connections = connections
181 return backup
182
183
185 """Admin node propagation logic. Instances are able to handle
186 incoming messages from a directly connected gateway, process them
187 and reply.
188
189 In order to take decisions, the instance acts as a finite states
190 machine, whose current state evolves according to received data.
191
192 -- INTERNALS --
193 Instance can be in one of the 4 different states:
194 - init (implicit)
195 This is the very first state. The instance enters the init
196 state at start() method, and will then send the configuration
197 to the remote node. Once the configuration is sent away, the
198 state changes to cfg.
199
200 - cfg
201 During this second state, the instance will wait for a valid
202 acknowledgement from the gateway to the previously sent
203 configuration message. If such a message is delivered, the
204 control message (the one that contains the actions to perform)
205 is sent, and the state is set to ctl.
206
207 - ctl
208 Third state, the instance is waiting for a valid ack for from
209 the gateway to the ctl packet. Then, the state switch to gtr
210 (gather).
211
212 - gtr
213 Final state: wait for results from the subtree and store them.
214 """
216 """
217 """
218 Channel.__init__(self)
219 self.task = task
220 self.gateway = gateway
221 self.workers = {}
222 self._cfg_write_hist = deque()
223 self._sendq = deque()
224 self._rc = None
225 self.logger = logging.getLogger(__name__)
226
228 """helper used to send a message, using msg queue if needed"""
229 if self.setup and not self._sendq:
230
231 self.send(ctl)
232 else:
233 self.logger.debug("send_queued: %d", len(self._sendq))
234 self._sendq.appendleft(ctl)
235
237 """helper used to send one queued message (if any)"""
238 if self._sendq:
239 ctl = self._sendq.pop()
240 self.logger.debug("dequeuing sendq: %s", ctl)
241 self.send(ctl)
242
251
252 - def recv(self, msg):
253 """process incoming messages"""
254 self.logger.debug("recv: %s", msg)
255 if msg.type == EndMessage.ident:
256
257 self.logger.debug("got EndMessage; closing")
258
259 self.worker.abort()
260 elif self.setup:
261 self.recv_ctl(msg)
262 elif self.opened:
263 self.recv_cfg(msg)
264 elif msg.type == StartMessage.ident:
265 self.opened = True
266 self.logger.debug('channel started (version %s on remote gateway)',
267 self._xml_reader.version)
268 else:
269 self.logger.error('unexpected message: %s', str(msg))
270
271 - def shell(self, nodes, command, worker, timeout, stderr, gw_invoke_cmd,
272 remote):
273 """command execution through channel"""
274 self.logger.debug("shell nodes=%s timeout=%s worker=%s remote=%s",
275 nodes, timeout, id(worker), remote)
276
277 self.workers[id(worker)] = worker
278
279 ctl = ControlMessage(id(worker))
280 ctl.action = 'shell'
281 ctl.target = nodes
282
283
284 info = dict((k, v) for k, v in self.task._info.items()
285 if k not in DEFAULTS._task_info_pkeys_bl)
286
287 ctl_data = {
288 'cmd': command,
289 'invoke_gateway': gw_invoke_cmd,
290 'taskinfo': info,
291 'stderr': stderr,
292 'timeout': timeout,
293 'remote': remote,
294 }
295 ctl.data_encode(ctl_data)
296 self.send_queued(ctl)
297
298 - def write(self, nodes, buf, worker):
299 """write buffer through channel to nodes on standard input"""
300 self.logger.debug("write buflen=%d", len(buf))
301 assert id(worker) in self.workers
302
303 ctl = ControlMessage(id(worker))
304 ctl.action = 'write'
305 ctl.target = nodes
306
307 ctl_data = {
308 'buf': buf,
309 }
310 ctl.data_encode(ctl_data)
311 self._cfg_write_hist.appendleft((ctl.msgid, nodes, len(buf), worker))
312 self.send_queued(ctl)
313
315 """send EOF through channel to specified nodes"""
316 self.logger.debug("set_write_eof")
317 assert id(worker) in self.workers
318
319 ctl = ControlMessage(id(worker))
320 ctl.action = 'eof'
321 ctl.target = nodes
322 self.send_queued(ctl)
323
325 """handle incoming messages for state 'propagate configuration'"""
326 self.logger.debug("recv_cfg")
327 if msg.type == 'ACK':
328 self.logger.debug("CTL - connection with gateway fully established")
329 self.setup = True
330 self.send_dequeue()
331 else:
332 self.logger.debug("_state_config error (msg=%s)", msg)
333
335 """handle incoming messages for state 'control'"""
336 if msg.type == 'ACK':
337 self.logger.debug("got ack (%s)", msg.type)
338
339 if self._cfg_write_hist and msg.ack == self._cfg_write_hist[-1][0]:
340 _, nodes, bytes_count, metaworker = self._cfg_write_hist.pop()
341 for node in nodes:
342
343
344 metaworker._on_written(node, bytes_count, 'stdin')
345 self.send_dequeue()
346 elif isinstance(msg, RoutedMessageBase):
347 metaworker = self.workers[msg.srcid]
348 if msg.type == StdOutMessage.ident:
349 nodeset = NodeSet(msg.nodes)
350 decoded = msg.data_decode() + '\n'
351 for line in decoded.splitlines():
352 for node in nodeset:
353 metaworker._on_remote_node_msgline(node, line, 'stdout',
354 self.gateway)
355 elif msg.type == StdErrMessage.ident:
356 nodeset = NodeSet(msg.nodes)
357 decoded = msg.data_decode() + '\n'
358 for line in decoded.splitlines():
359 for node in nodeset:
360 metaworker._on_remote_node_msgline(node, line, 'stderr',
361 self.gateway)
362 elif msg.type == RetcodeMessage.ident:
363 rc = msg.retcode
364 for node in NodeSet(msg.nodes):
365 metaworker._on_remote_node_rc(node, rc, self.gateway)
366 elif msg.type == TimeoutMessage.ident:
367 self.logger.debug("TimeoutMessage for %s", msg.nodes)
368 for node in NodeSet(msg.nodes):
369 metaworker._on_remote_node_timeout(node, self.gateway)
370 elif msg.type == ErrorMessage.ident:
371
372 raise TopologyError("%s: %s" % (self.gateway, msg.reason))
373 else:
374 self.logger.debug("recv_ctl: unhandled msg %s", msg)
375 """
376 return
377 if self.ptree.upchannel is not None:
378 self.logger.debug("_state_gather ->upchan %s" % msg)
379 self.ptree.upchannel.send(msg) # send to according event handler passed by shell()
380 else:
381 assert False
382 """
383
385 """Channel command is closing"""
386 self._rc = worker.current_rc
387
389 """Channel is closing"""
390
391
392 gateway = str(worker.nodes)
393 self.logger.debug("ev_close gateway=%s %s", gateway, self)
394 self.logger.debug("ev_close rc=%s", self._rc)
395
396 if self._rc:
397
398 self.logger.debug("unreachable gateway %s", gateway)
399 worker.task.router.mark_unreachable(gateway)
400 self.logger.debug("worker.task.gateways=%s", worker.task.gateways)
401
402