#
# Copyright (C) 2011-2016 CEA/DAM
# Copyright (C) 2015-2017 Stephane Thiell <sthiell@stanford.edu>
#
# This file is part of ClusterShell.
#
# ClusterShell is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# ClusterShell is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with ClusterShell; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
# This file is part of the ClusterShell library.
"""
ClusterShell tree propagation worker
"""
import base64
import logging
import os
from os.path import basename, dirname, isfile, normpath
import sys
import tarfile
import tempfile
from ClusterShell.Event import EventHandler
from ClusterShell.NodeSet import NodeSet
from ClusterShell.Worker.EngineClient import EnginePort
from ClusterShell.Worker.Worker import DistantWorker, WorkerError
from ClusterShell.Worker.Worker import _eh_sigspec_invoke_compat
from ClusterShell.Worker.Exec import ExecWorker
from ClusterShell.Propagation import PropagationTreeRouter
# PEP 706: 'fully_trusted' preserves legacy extractall() (trusted gateway)
_TAR_EXTRACT_KWARGS = {'filter': 'fully_trusted'} \
if hasattr(tarfile, 'data_filter') else {}
class MetaWorkerEventHandler(EventHandler):
"""Handle events for the meta worker TreeWorker"""
def __init__(self, metaworker):
self.metaworker = metaworker
self.logger = logging.getLogger(__name__)
def ev_start(self, worker):
"""
Called to indicate that a worker has just started.
"""
self.logger.debug("MetaWorkerEventHandler: ev_start")
self.metaworker._start_count += 1
def ev_pickup(self, worker, node):
"""
Called for each node a child worker has just picked up.
Propagate the event to the meta worker.
"""
self.metaworker._emit_pickup(node)
def ev_read(self, worker, node, sname, msg):
"""
Called to indicate that a worker has data to read.
"""
self.metaworker._on_node_msgline(node, msg, sname)
def ev_written(self, worker, node, sname, size):
"""
Called to indicate that writing has been done.
"""
metaworker = self.metaworker
metaworker.current_node = node
metaworker.current_sname = sname
if metaworker.eh:
metaworker.eh.ev_written(metaworker, node, sname, size)
def ev_hup(self, worker, node, rc):
"""
Called to indicate that a worker's connection has been closed.
"""
self.metaworker._on_node_close(node, rc)
def ev_close(self, worker, timedout):
"""
Called to indicate that a worker has just finished. It may have failed
on timeout if timedout is set.
"""
self.logger.debug("MetaWorkerEventHandler: ev_close, timedout=%s",
timedout)
if timedout:
# WARNING!!! this is not possible as metaworker is changing task's
# shared timeout set!
#for node in worker.iter_keys_timeout():
# self.metaworker._on_node_timeout(node)
# we use NodeSet to copy set
for node in NodeSet._fromlist1(worker.iter_keys_timeout()):
self.metaworker._on_node_timeout(node)
self.metaworker._check_fini()
[docs]
class TreeWorker(DistantWorker):
"""
ClusterShell tree worker Class.
"""
# copy and rcopy tar command formats
# the choice of single or double quotes is essential
UNTAR_CMD_FMT = "tar -xf - -C '%s'"
TAR_CMD_FMT = "tar -cf - -C '%s' " \
"--transform \"s,^\\([^/]*\\)[/]*,\\1.$(hostname -s)/,\" " \
"'%s' | base64 -w 65536"
class _IOPortHandler(EventHandler):
"""
Special control port event handler used for:
* start the TreeWorker when the engine starts
* early write handling: write buffering and eof tracking
"""
def __init__(self, treeworker):
EventHandler.__init__(self)
self.treeworker = treeworker
def ev_port_start(self, port):
"""Event when port is registered."""
self.treeworker._start()
def ev_msg(self, port, msg):
"""
Message received: call appropriate worker method.
Used for TreeWorker.write() and set_write_eof().
"""
func, args = msg[0], msg[1:]
func(self.treeworker, *args)
[docs]
def __init__(self, nodes, handler, timeout, **kwargs):
"""
Initialize Tree worker instance.
:param nodes: Targeted nodeset.
:param handler: Worker EventHandler.
:param timeout: Timeout value for worker.
:param command: Command to execute.
:param topology: Force specific TopologyTree.
:param newroot: Root node of TopologyTree.
"""
DistantWorker.__init__(self, handler)
self.logger = logging.getLogger(__name__)
self.workers = []
self.nodes = NodeSet(nodes)
self.timeout = timeout
self.command = kwargs.get('command')
self.source = kwargs.get('source')
self.dest = kwargs.get('dest')
autoclose = kwargs.get('autoclose', False)
self.stderr = kwargs.get('stderr', False)
self.logger.debug("stderr=%s", self.stderr)
self.remote = kwargs.get('remote', True)
self.preserve = kwargs.get('preserve', None)
self.reverse = kwargs.get('reverse', False)
self._rcopy_bufs = {}
self._rcopy_tars = {}
self._close_count = 0
self._start_count = 0
self._child_count = 0
self._target_count = 0
self._has_timeout = False
self._started = False
self._initialized = False
self._aborted = False
# buffered pickup events queued during _launch() (before ev_start)
self._pending_pickups = []
if self.command is None and self.source is None:
raise ValueError("missing command or source parameter in "
"TreeWorker constructor")
# copy/rcopy: keep internal tar/scp errors on a separate stderr stream
# instead of merging them into stdout (which also carries rcopy data)
if self.source:
self.stderr = True
# build gateway invocation command
invoke_gw_args = []
for envname in ('PYTHONPATH',
'CLUSTERSHELL_GW_PYTHON_EXECUTABLE',
'CLUSTERSHELL_GW_LOG_DIR',
'CLUSTERSHELL_GW_LOG_LEVEL',
'CLUSTERSHELL_GW_B64_LINE_LENGTH'):
envval = os.getenv(envname)
if envval:
invoke_gw_args.append("%s=%s" % (envname, envval))
# It is critical to launch a remote Python executable with the same
# major version (ie. python or python3): the pickle protocol used
# for gateway messages is pinned (Communication.GW_PICKLE_PROTOCOL)
# but still differs between Python 2 and Python 3.
python_executable = os.getenv('CLUSTERSHELL_GW_PYTHON_EXECUTABLE',
basename(sys.executable or 'python'))
invoke_gw_args.append(python_executable)
invoke_gw_args.extend(['-m', 'ClusterShell.Gateway', '-Bu'])
self.invoke_gateway = ' '.join(invoke_gw_args)
self.topology = kwargs.get('topology')
if self.topology is not None:
self.newroot = kwargs.get('newroot') or \
str(self.topology.root.nodeset)
self.router = PropagationTreeRouter(self.newroot, self.topology)
else:
self.router = None
self.upchannel = None
self.metahandler = MetaWorkerEventHandler(self)
# gateway (string) -> active targets selection
self.gwtargets = {}
# IO port
self._port = EnginePort(handler=TreeWorker._IOPortHandler(self),
autoclose=True)
def _start(self):
# Engine has started: initialize router
self.topology = self.topology or self.task.topology
self.router = self.task._default_router(self.router)
self._launch(self.nodes)
self._check_ini()
self._started = True
def _launch(self, nodes):
self.logger.debug("TreeWorker._launch on %s (fanout=%d)", nodes,
self.task.info("fanout"))
# Prepare copy params if source is defined
destdir = None
if self.source:
if self.reverse:
self.logger.debug("rcopy source=%s, dest=%s", self.source,
self.dest)
# dest is a directory
destdir = self.dest
else:
self.logger.debug("copy source=%s, dest=%s", self.source,
self.dest)
# Special processing to determine best arcname and destdir for
# tar. The only case that we don't support is when source is a
# file and dest is a dir without a finishing / (in that case we
# cannot determine remotely whether it is a file or a
# directory).
if isfile(self.source):
# dest is not normalized here
arcname = basename(self.dest) or \
basename(normpath(self.source))
destdir = dirname(self.dest)
else:
# source is a directory: if dest has a trailing slash
# like in /tmp/ then arcname is basename(source)
# but if dest is /tmp/newname (without leading slash) then
# arcname becomes newname.
if self.dest[-1] == '/':
arcname = basename(self.source)
else:
arcname = basename(self.dest)
# dirname has not the same behavior when a leading slash is
# present, and we want that.
destdir = dirname(self.dest)
self.logger.debug("copy arcname=%s destdir=%s", arcname,
destdir)
# And launch stuffs
next_hops = self._distribute(self.task.info("fanout"), nodes.copy())
self.logger.debug("next_hops=%s" % [(str(n), str(v))
for n, v in next_hops])
for gw, targets in next_hops:
if gw == targets:
self.logger.debug('task.shell cmd=%s source=%s nodes=%s '
'timeout=%s remote=%s', self.command,
self.source, nodes, self.timeout, self.remote)
self._child_count += 1
self._target_count += len(targets)
if self.remote:
if self.source:
# Note: specific case where targets are not in topology
# as self.source is never used on remote gateways
# so we try a direct copy/rcopy:
self.logger.debug('_launch copy r=%s source=%s dest=%s',
self.reverse, self.source, self.dest)
worker = self.task.copy(self.source, self.dest, targets,
handler=self.metahandler,
stderr=self.stderr,
timeout=self.timeout,
preserve=self.preserve,
reverse=self.reverse,
tree=False)
else:
worker = self.task.shell(self.command,
nodes=targets,
timeout=self.timeout,
handler=self.metahandler,
stderr=self.stderr,
tree=False)
else:
assert self.source is None
workerclass = self.task.default('local_worker')
worker = workerclass(nodes=targets,
command=self.command,
handler=self.metahandler,
timeout=self.timeout,
stderr=self.stderr)
self.task.schedule(worker)
self.workers.append(worker)
self.logger.debug("added child worker %s count=%d", worker,
len(self.workers))
else:
self.logger.debug("trying gateway %s to reach %s", gw, targets)
if self.source:
self._copy_remote(self.source, destdir, targets, gw,
self.timeout, self.reverse)
else:
self._execute_remote(self.command, targets, gw,
self.timeout)
# Copy mode: send tar data after above workers have been initialized
if self.source and not self.reverse:
try:
# create temporary tar file with all source files
tmptar = tempfile.TemporaryFile()
tar = tarfile.open(fileobj=tmptar, mode='w:')
tar.add(self.source, arcname=arcname)
tar.close()
tmptar.flush()
# read generated tar file
tmptar.seek(0)
rbuf = tmptar.read(32768)
# send tar data to remote targets only
while len(rbuf) > 0:
self._write_remote(rbuf)
rbuf = tmptar.read(32768)
except OSError as exc:
raise WorkerError(exc)
def _distribute(self, fanout, dst_nodeset):
"""distribute target nodes between next hop gateways"""
self.router.fanout = fanout
distribution = {}
for gw, dstset in self.router.dispatch(dst_nodeset):
distribution.setdefault(str(gw), NodeSet()).add(dstset)
return tuple((NodeSet(k), v) for k, v in distribution.items())
def _copy_remote(self, source, dest, targets, gateway, timeout, reverse):
"""run a remote copy in tree mode (using gateway)"""
self.logger.debug("_copy_remote gateway=%s source=%s dest=%s "
"reverse=%s", gateway, source, dest, reverse)
self._target_count += len(targets)
self.gwtargets.setdefault(str(gateway), set()).update(targets)
# tar commands are built here and launched on targets
if reverse:
# these weird replace calls aim to escape single quotes ' within ''
srcdir = dirname(source).replace("'", '\'\"\'\"\'')
srcbase = basename(normpath(self.source)).replace("'", '\'\"\'\"\'')
cmd = self.TAR_CMD_FMT % (srcdir, srcbase)
else:
cmd = self.UNTAR_CMD_FMT % dest.replace("'", '\'\"\'\"\'')
self.logger.debug('_copy_remote: tar cmd: %s', cmd)
pchan = self.task._pchannel(gateway, self)
# ev_pickup is fired by PropagationChannel._send_ctl after the send
pchan.shell(nodes=targets, command=cmd, worker=self, timeout=timeout,
stderr=self.stderr, gw_invoke_cmd=self.invoke_gateway,
remote=self.remote)
def _execute_remote(self, cmd, targets, gateway, timeout):
"""run command against a remote node via a gateway"""
self.logger.debug("_execute_remote gateway=%s cmd=%s targets=%s",
gateway, cmd, targets)
self._target_count += len(targets)
# GH#560: use set instead of NodeSet to keep track of active targets
self.gwtargets.setdefault(str(gateway), set()).update(targets)
pchan = self.task._pchannel(gateway, self)
# ev_pickup is fired by PropagationChannel._send_ctl after the send
pchan.shell(nodes=targets, command=cmd, worker=self, timeout=timeout,
stderr=self.stderr, gw_invoke_cmd=self.invoke_gateway,
remote=self.remote)
def _relaunch(self, previous_gateway):
"""Redistribute and relaunch commands on targets that were running
on previous_gateway (which is probably marked unreachable by now)
NOTE: Relaunch is always called after failed remote execution, so
previous_gateway must be defined. However, it is not guaranteed that
the relaunch is going to be performed using gateways (that's a feature).
"""
targets = NodeSet.fromlist(self.gwtargets[previous_gateway])
self.logger.debug("_relaunch on targets %s from previous gateway %s",
targets, previous_gateway)
self.gwtargets[previous_gateway].difference_update(targets)
self._check_fini(previous_gateway)
self._target_count -= len(targets)
if self.eh is not None:
self.eh._ev_routing(self, { "event": "reroute",
"gateway": previous_gateway,
"targets": targets })
self._launch(targets)
def _engine_clients(self):
"""
Access underlying engine clients.
"""
return [self._port]
def _on_remote_node_msgline(self, node, msg, sname, gateway):
"""remote msg received"""
if not self.source or not self.reverse or sname != 'stdout':
DistantWorker._on_node_msgline(self, node, msg, sname)
return
# rcopy only: we expect base64 encoded tar content on stdout
encoded = self._rcopy_bufs.setdefault(node, b'') + msg
if node not in self._rcopy_tars:
self._rcopy_tars[node] = tempfile.TemporaryFile()
# partial base64 decoding requires a multiple of 4 characters
encoded_sz = (len(encoded) // 4) * 4
# write decoded binary msg to node temporary tarfile
self._rcopy_tars[node].write(base64.b64decode(encoded[0:encoded_sz]))
# keep trailing encoded chars for next time
self._rcopy_bufs[node] = encoded[encoded_sz:]
def _on_remote_node_close(self, node, rc, gateway):
"""remote node closing with return code"""
self.logger.debug("_on_remote_node_close %s %s via gw %s rc=%s", node,
self._close_count, gateway, rc)
# this must be done first to avoid recursion via event handlers
self.gwtargets[str(gateway)].remove(node)
DistantWorker._on_node_close(self, node, rc)
# finalize rcopy: extract tar data
if self.source and self.reverse:
if node in self._rcopy_bufs:
buf = self._rcopy_bufs[node]
tarfileobj = self._rcopy_tars[node]
if len(buf) > 0:
self.logger.debug("flushing node %s buf %d bytes", node,
len(buf))
tarfileobj.write(buf)
tarfileobj.flush()
tarfileobj.seek(0)
tmptar = None
try:
tmptar = tarfile.open(fileobj=tarfileobj)
self.logger.debug("%s extracting %d members in dest %s",
node, len(tmptar.getmembers()), self.dest)
tmptar.extractall(path=self.dest, **_TAR_EXTRACT_KWARGS)
# see PEP 3151
except (IOError, OSError, tarfile.TarError) as ex:
self._on_remote_node_msgline(node, str(ex).encode(),
'stderr', gateway)
finally:
if tmptar is not None:
tmptar.close()
del self._rcopy_bufs[node]
del self._rcopy_tars[node]
else:
self.logger.debug("no rcopy buffer received from %s", node)
self._close_count += 1
self._check_fini(gateway)
def _on_remote_node_timeout(self, node, gateway):
"""remote node timeout received"""
DistantWorker._on_node_timeout(self, node)
self.logger.debug("_on_remote_node_timeout %s via gw %s", node, gateway)
self._close_count += 1
self._has_timeout = True
self.gwtargets[str(gateway)].remove(node)
self._check_fini(gateway)
def _on_node_close(self, node, rc):
DistantWorker._on_node_close(self, node, rc)
self.logger.debug("_on_node_close %s %s (%s)", node, rc,
self._close_count)
self._close_count += 1
def _on_node_timeout(self, node):
DistantWorker._on_node_timeout(self, node)
self.logger.debug("_on_node_timeout %s (%s)", node, self._close_count)
self._close_count += 1
self._has_timeout = True
def _on_routing_event(self, arg):
self.logger.debug("_on_routing_event %s", arg)
self.eh._ev_routing(self, arg)
def _emit_pickup(self, node):
"""Fire ev_pickup for a node, or queue it until ev_start fires.
Direct child workers route pickups through MetaWorkerEventHandler.
Gateway-routed targets route pickups through PropagationChannel
._send_ctl, after the CTL is actually written on the wire.
Skipped when the worker has been aborted.
The 1:1 invariant between ev_pickup and ev_hup per node is held
structurally by the callers: Worker._on_start fires ev_pickup
at most once per child node-key (direct path); _send_ctl runs
at most once per actual channel.send (gateway path), and reroute
only happens when the previous CTL was never sent -- so a
rerouted target has not yet been emitted. See
test_tree_run_gw2f1_reroute and
test_tree_run_gw2f1_no_double_pickup_under_reroute.
"""
if self.eh is None or self._aborted:
return
if self._initialized:
_eh_sigspec_invoke_compat(self.eh.ev_pickup, 2, self, node)
else:
# ev_start has not fired yet; defer to preserve ordering
self._pending_pickups.append(node)
def _check_ini(self):
self.logger.debug("TreeWorker: _check_ini (%d, %d)", self._start_count,
self._child_count)
if (self.eh is not None and not self._initialized
and self._start_count >= self._child_count):
# this part is called once
self._initialized = True
self.eh.ev_start(self)
# Flush pickup events buffered during _launch(). If ev_start
# aborted the worker, no pickups are emitted (#594).
pending, self._pending_pickups = self._pending_pickups, []
for node in pending:
if self._aborted:
break
_eh_sigspec_invoke_compat(self.eh.ev_pickup, 2, self, node)
def _check_fini(self, gateway=None):
self.logger.debug("check_fini %s %s", self._close_count,
self._target_count)
if self._close_count >= self._target_count:
handler = self.eh
if handler:
# also use hasattr check because ev_timeout was missing in 1.8.0
if self._has_timeout and hasattr(handler, 'ev_timeout'):
handler.ev_timeout(self)
_eh_sigspec_invoke_compat(handler.ev_close, 2, self,
self._has_timeout)
# check completion of targets per gateway
if gateway:
targets = self.gwtargets[str(gateway)]
if not targets:
# no more active targets for this gateway
self.logger.debug("TreeWorker._check_fini %s call pchannel_"
"release for gw %s", self, gateway)
self.task._pchannel_release(gateway, self)
del self.gwtargets[str(gateway)]
def _write_remote(self, buf):
"""Write buf to remote clients only."""
for gateway, targets in self.gwtargets.items():
assert len(targets) > 0
self.task._pchannel(gateway, self).write(nodes=targets, buf=buf,
worker=self)
def _set_write_eof_remote(self):
for gateway, targets in self.gwtargets.items():
assert len(targets) > 0
self.task._pchannel(gateway, self).set_write_eof(nodes=targets,
worker=self)
[docs]
def write(self, buf):
"""Write to worker clients."""
if not self._started:
self._port.msg_send((TreeWorker.write, buf))
return
osexc = None
# Differentiate directly handled writes from remote ones
for worker in self.workers:
try:
worker.write(buf)
except OSError as exc:
osexc = exc
self._write_remote(buf)
if osexc:
raise osexc
[docs]
def set_write_eof(self):
"""
Tell worker to close its writer file descriptor once flushed. Do not
perform writes after this call.
"""
if not self._started:
self._port.msg_send((TreeWorker.set_write_eof, ))
return
# Differentiate directly handled EOFs from remote ones
for worker in self.workers:
worker.set_write_eof()
self._set_write_eof_remote()
def _gateway_abort(self, gateway):
"""Abort on gateway failure"""
if gateway not in self.gwtargets:
self.logger.warning("TreeWorker._gateway_abort %s not found",
gateway)
return
targets = self.gwtargets[gateway]
self.logger.debug("TreeWorker._gateway_abort %s found: targets=%s",
gateway, targets)
for target in NodeSet.fromlist(targets): # targets is a mutable list
self._on_remote_node_close(target, os.EX_PROTOCOL, gateway)
[docs]
def abort(self):
"""Abort processing any action by this worker."""
self.logger.debug("abort %s" % self)
# Stop pending and future pickup events (#594)
self._aborted = True
self._pending_pickups = []
for worker in self.workers:
worker.abort()
for gateway in self.gwtargets.copy():
self._gateway_abort(gateway)
# TreeWorker's former name (deprecated as of 1.8)
WorkerTree = TreeWorker