/usr/share/pyshared/ClusterShell/CLI/Clush.py is in clustershell 1.6-4.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 | #!/usr/bin/env python
#
# Copyright CEA/DAM/DIF (2007, 2008, 2009, 2010, 2011, 2012)
# Contributor: Stephane THIELL <stephane.thiell@cea.fr>
#
# This file is part of the ClusterShell library.
#
# This software is governed by the CeCILL-C license under French law and
# abiding by the rules of distribution of free software. You can use,
# modify and/ or redistribute the software under the terms of the CeCILL-C
# license as circulated by CEA, CNRS and INRIA at the following URL
# "http://www.cecill.info".
#
# As a counterpart to the access to the source code and rights to copy,
# modify and redistribute granted by the license, users are provided only
# with a limited warranty and the software's author, the holder of the
# economic rights, and the successive licensors have only limited
# liability.
#
# In this respect, the user's attention is drawn to the risks associated
# with loading, using, modifying and/or developing or reproducing the
# software by the user in light of its specific status of free software,
# that may mean that it is complicated to manipulate, and that also
# therefore means that it is reserved for developers and experienced
# professionals having in-depth computer knowledge. Users are therefore
# encouraged to load and test the software's suitability as regards their
# requirements in conditions enabling the security of their systems and/or
# data to be ensured and, more generally, to use and operate it in the
# same conditions as regards security.
#
# The fact that you are presently reading this means that you have had
# knowledge of the CeCILL-C license and that you accept its terms.
"""
execute cluster commands in parallel
clush is an utility program to run commands on a cluster which benefits
from the ClusterShell library and its Ssh worker. It features an
integrated output results gathering system (dshbak-like), can get node
groups by running predefined external commands and can redirect lines
read on its standard input to the remote commands.
When no command are specified, clush runs interactively.
"""
import errno
import logging
import os
import resource
import sys
import signal
import threading
from ClusterShell.CLI.Config import ClushConfig, ClushConfigError
from ClusterShell.CLI.Display import Display
from ClusterShell.CLI.Display import VERB_QUIET, VERB_STD, VERB_VERB, VERB_DEBUG
from ClusterShell.CLI.OptionParser import OptionParser
from ClusterShell.CLI.Error import GENERIC_ERRORS, handle_generic_error
from ClusterShell.CLI.Utils import NodeSet, bufnodeset_cmp
from ClusterShell.Event import EventHandler
from ClusterShell.MsgTree import MsgTree
from ClusterShell.NodeSet import RESOLVER_NOGROUP, RESOLVER_STD_GROUP
from ClusterShell.NodeSet import NodeSetParseError
from ClusterShell.Task import Task, task_self
class UpdatePromptException(Exception):
"""Exception used by the signal handler"""
class StdInputHandler(EventHandler):
"""Standard input event handler class."""
def __init__(self, worker):
EventHandler.__init__(self)
self.master_worker = worker
def ev_msg(self, port, msg):
"""invoked when a message is received from port object"""
if not msg:
self.master_worker.set_write_eof()
return
# Forward messages to master worker
self.master_worker.write(msg)
class OutputHandler(EventHandler):
"""Base class for clush output handlers."""
def __init__(self):
EventHandler.__init__(self)
self._runtimer = None
def runtimer_init(self, task, ntotal):
"""Init timer for live command-completed progressmeter."""
self._runtimer = task.timer(2.0, RunTimer(task, ntotal),
interval=1./3., autoclose=True)
def _runtimer_clean(self):
"""Hide runtimer counter"""
if self._runtimer:
self._runtimer.eh.erase_line()
def _runtimer_set_dirty(self):
"""Force redisplay of counter"""
if self._runtimer:
self._runtimer.eh.set_dirty()
def _runtimer_finalize(self, worker):
"""Finalize display of runtimer counter"""
if self._runtimer:
self._runtimer.eh.finalize(worker.task.default("USER_interactive"))
def update_prompt(self, worker):
"""
If needed, notify main thread to update its prompt by sending
a SIGUSR1 signal. We use task-specific user-defined variable
to record current states (prefixed by USER_).
"""
worker.task.set_default("USER_running", False)
if worker.task.default("USER_handle_SIGUSR1"):
os.kill(os.getpid(), signal.SIGUSR1)
class DirectOutputHandler(OutputHandler):
"""Direct output event handler class."""
def __init__(self, display):
OutputHandler.__init__(self)
self._display = display
def ev_read(self, worker):
node = worker.current_node or worker.key
self._display.print_line(node, worker.current_msg)
def ev_error(self, worker):
node = worker.current_node or worker.key
self._display.print_line_error(node, worker.current_errmsg)
def ev_hup(self, worker):
node = worker.current_node or worker.key
rc = worker.current_rc
if rc > 0:
verb = VERB_QUIET
if self._display.maxrc:
verb = VERB_STD
self._display.vprint_err(verb, \
"clush: %s: exited with exit code %d" % (node, rc))
def ev_timeout(self, worker):
self._display.vprint_err(VERB_QUIET, "clush: %s: command timeout" % \
NodeSet._fromlist1(worker.iter_keys_timeout()))
def ev_close(self, worker):
self.update_prompt(worker)
class CopyOutputHandler(DirectOutputHandler):
"""Copy output event handler."""
def __init__(self, display, reverse=False):
DirectOutputHandler.__init__(self, display)
self.reverse = reverse
def ev_close(self, worker):
"""A copy worker has finished."""
for rc, nodes in worker.iter_retcodes():
if rc == 0:
if self.reverse:
self._display.vprint(VERB_VERB, "%s:`%s' -> `%s'" % \
(nodes, worker.source, worker.dest))
else:
self._display.vprint(VERB_VERB, "`%s' -> %s:`%s'" % \
(worker.source, nodes, worker.dest))
break
# multiple copy workers may be running (handled by this task's thread)
copies = worker.task.default("USER_copies") - 1
worker.task.set_default("USER_copies", copies)
if copies == 0:
self._runtimer_finalize(worker)
self.update_prompt(worker)
class GatherOutputHandler(OutputHandler):
"""Gathered output event handler class."""
def __init__(self, display):
OutputHandler.__init__(self)
self._display = display
def ev_read(self, worker):
if self._display.verbosity == VERB_VERB:
node = worker.current_node or worker.key
self._display.print_line(node, worker.current_msg)
def ev_error(self, worker):
self._runtimer_clean()
self._display.print_line_error(worker.current_node,
worker.current_errmsg)
self._runtimer_set_dirty()
def ev_close(self, worker):
# Worker is closing -- it's time to gather results...
self._runtimer_finalize(worker)
assert worker.current_node is not None, "cannot gather local command"
# Display command output, try to order buffers by rc
nodesetify = lambda v: (v[0], NodeSet._fromlist1(v[1]))
cleaned = False
for rc, nodelist in sorted(worker.iter_retcodes()):
# Then order by node/nodeset (see bufnodeset_cmp)
for buf, nodeset in sorted(map(nodesetify,
worker.iter_buffers(nodelist)),
cmp=bufnodeset_cmp):
if not cleaned:
# clean runtimer line before printing first result
self._runtimer_clean()
cleaned = True
self._display.print_gather(nodeset, buf)
self._display.flush()
self._close_common(worker)
# Notify main thread to update its prompt
self.update_prompt(worker)
def _close_common(self, worker):
verbexit = VERB_QUIET
if self._display.maxrc:
verbexit = VERB_STD
# Display return code if not ok ( != 0)
for rc, nodelist in worker.iter_retcodes():
if rc != 0:
ns = NodeSet._fromlist1(nodelist)
self._display.vprint_err(verbexit, \
"clush: %s: exited with exit code %d" % (ns, rc))
# Display nodes that didn't answer within command timeout delay
if worker.num_timeout() > 0:
self._display.vprint_err(verbexit, "clush: %s: command timeout" % \
NodeSet._fromlist1(worker.iter_keys_timeout()))
class LiveGatherOutputHandler(GatherOutputHandler):
"""Live line-gathered output event handler class."""
def __init__(self, display, nodes):
assert nodes is not None, "cannot gather local command"
GatherOutputHandler.__init__(self, display)
self._nodes = NodeSet(nodes)
self._nodecnt = dict.fromkeys(self._nodes, 0)
self._mtreeq = []
self._offload = 0
def ev_read(self, worker):
# Read new line from node
node = worker.current_node
self._nodecnt[node] += 1
cnt = self._nodecnt[node]
if len(self._mtreeq) < cnt:
self._mtreeq.append(MsgTree())
self._mtreeq[cnt - self._offload - 1].add(node, worker.current_msg)
self._live_line(worker)
def ev_hup(self, worker):
if self._mtreeq and worker.current_node not in self._mtreeq[0]:
# forget a node that doesn't answer to continue live line
# gathering anyway
self._nodes.remove(worker.current_node)
self._live_line(worker)
def _live_line(self, worker):
# if all nodes have replied, display gathered line
while self._mtreeq and len(self._mtreeq[0]) == len(self._nodes):
mtree = self._mtreeq.pop(0)
self._offload += 1
self._runtimer_clean()
nodesetify = lambda v: (v[0], NodeSet.fromlist(v[1]))
for buf, nodeset in sorted(map(nodesetify, mtree.walk()),
cmp=bufnodeset_cmp):
self._display.print_gather(nodeset, buf)
self._runtimer_set_dirty()
def ev_close(self, worker):
# Worker is closing -- it's time to gather results...
self._runtimer_finalize(worker)
for mtree in self._mtreeq:
nodesetify = lambda v: (v[0], NodeSet.fromlist(v[1]))
for buf, nodeset in sorted(map(nodesetify, mtree.walk()),
cmp=bufnodeset_cmp):
self._display.print_gather(nodeset, buf)
self._close_common(worker)
# Notify main thread to update its prompt
self.update_prompt(worker)
class RunTimer(EventHandler):
"""Running progress timer event handler"""
def __init__(self, task, total):
EventHandler.__init__(self)
self.task = task
self.total = total
self.cnt_last = -1
self.tslen = len(str(self.total))
self.wholelen = 0
self.started = False
def ev_timer(self, timer):
self.update()
def set_dirty(self):
self.cnt_last = -1
def erase_line(self):
if self.wholelen:
sys.stderr.write(' ' * self.wholelen + '\r')
def update(self):
cnt = len(self.task._engine.clients())
if cnt != self.cnt_last:
self.cnt_last = cnt
# display completed/total clients
towrite = 'clush: %*d/%*d\r' % (self.tslen, self.total - cnt,
self.tslen, self.total)
self.wholelen = len(towrite)
sys.stderr.write(towrite)
self.started = True
def finalize(self, force_cr):
"""finalize display of runtimer"""
if not self.started:
return
# display completed/total clients
fmt = 'clush: %*d/%*d'
if force_cr:
fmt += '\n'
else:
fmt += '\r'
sys.stderr.write(fmt % (self.tslen, self.total, self.tslen, self.total))
def signal_handler(signum, frame):
"""Signal handler used for main thread notification"""
if signum == signal.SIGUSR1:
signal.signal(signal.SIGUSR1, signal.SIG_IGN)
raise UpdatePromptException()
def get_history_file():
"""Turn the history file path"""
return os.path.join(os.environ["HOME"], ".clush_history")
def readline_setup():
"""
Configure readline to automatically load and save a history file
named .clush_history
"""
import readline
readline.parse_and_bind("tab: complete")
readline.set_completer_delims("")
try:
readline.read_history_file(get_history_file())
except IOError:
pass
def ttyloop(task, nodeset, timeout, display):
"""Manage the interactive prompt to run command"""
readline_avail = False
if task.default("USER_interactive"):
try:
import readline
readline_setup()
readline_avail = True
except ImportError:
pass
display.vprint(VERB_STD, \
"Enter 'quit' to leave this interactive mode")
rc = 0
ns = NodeSet(nodeset)
ns_info = True
cmd = ""
while task.default("USER_running") or cmd.lower() != 'quit':
try:
if task.default("USER_interactive") and \
not task.default("USER_running"):
if ns_info:
display.vprint(VERB_QUIET, \
"Working with nodes: %s" % ns)
ns_info = False
prompt = "clush> "
else:
prompt = ""
# Set SIGUSR1 handler if needed
if task.default("USER_handle_SIGUSR1"):
signal.signal(signal.SIGUSR1, signal_handler)
try:
cmd = raw_input(prompt)
finally:
signal.signal(signal.SIGUSR1, signal.SIG_IGN)
except EOFError:
print
return
except UpdatePromptException:
if task.default("USER_interactive"):
continue
return
except KeyboardInterrupt, kbe:
if display.gather:
# Suspend task, so we can safely access its data from
# the main thread
task.suspend()
print_warn = False
# Display command output, but cannot order buffers by rc
nodesetify = lambda v: (v[0], NodeSet._fromlist1(v[1]))
for buf, nodeset in sorted(map(nodesetify, task.iter_buffers()),
cmp=bufnodeset_cmp):
if not print_warn:
print_warn = True
display.vprint_err(VERB_STD, \
"Warning: Caught keyboard interrupt!")
display.print_gather(nodeset, buf)
# Return code handling
verbexit = VERB_QUIET
if display.maxrc:
verbexit = VERB_STD
ns_ok = NodeSet()
for rc, nodelist in task.iter_retcodes():
ns_ok.add(NodeSet._fromlist1(nodelist))
if rc != 0:
# Display return code if not ok ( != 0)
ns = NodeSet._fromlist1(nodelist)
display.vprint_err(verbexit, \
"clush: %s: exited with exit code %s" % (ns, rc))
# Add uncompleted nodeset to exception object
kbe.uncompleted_nodes = ns - ns_ok
# Display nodes that didn't answer within command timeout delay
if task.num_timeout() > 0:
display.vprint_err(verbexit, \
"clush: %s: command timeout" % \
NodeSet._fromlist1(task.iter_keys_timeout()))
raise kbe
if task.default("USER_running"):
ns_reg, ns_unreg = NodeSet(), NodeSet()
for c in task._engine.clients():
if c.registered:
ns_reg.add(c.key)
else:
ns_unreg.add(c.key)
if ns_unreg:
pending = "\nclush: pending(%d): %s" % (len(ns_unreg), ns_unreg)
else:
pending = ""
display.vprint_err(VERB_QUIET, "clush: interrupt (^C to " \
"abort task)\nclush: in progress(%d): %s%s" % (len(ns_reg), \
ns_reg, pending))
else:
cmdl = cmd.lower()
try:
ns_info = True
if cmdl.startswith('+'):
ns.update(cmdl[1:])
elif cmdl.startswith('-'):
ns.difference_update(cmdl[1:])
elif cmdl.startswith('@'):
ns = NodeSet(cmdl[1:])
elif cmdl == '=':
display.gather = not display.gather
if display.gather:
display.vprint(VERB_STD, \
"Switching to gathered output format")
else:
display.vprint(VERB_STD, \
"Switching to standard output format")
task.set_default("stdout_msgtree", \
display.gather or display.line_mode)
ns_info = False
continue
elif not cmdl.startswith('?'): # if ?, just print ns_info
ns_info = False
except NodeSetParseError:
display.vprint_err(VERB_QUIET, \
"clush: nodeset parse error (ignoring)")
if ns_info:
continue
if cmdl.startswith('!') and len(cmd.strip()) > 0:
run_command(task, cmd[1:], None, timeout, display)
elif cmdl != "quit":
if not cmd:
continue
if readline_avail:
readline.write_history_file(get_history_file())
run_command(task, cmd, ns, timeout, display)
return rc
def _stdin_thread_start(stdin_port):
"""Standard input reader thread entry point."""
# Note: read length should be larger and a multiple of 4096 for best
# performance to avoid excessive unreg/register of writer fd in
# engine; however, it shouldn't be too large.
bufsize = 4096 * 8
# thread loop: blocking read stdin + send messages to specified
# port object
buf = sys.stdin.read(bufsize)
while buf:
# send message to specified port object (with ack)
stdin_port.msg(buf)
buf = sys.stdin.read(bufsize)
# send a None message to indicate EOF
stdin_port.msg(None)
def bind_stdin(worker):
"""Create a stdin->port->worker binding: connect specified worker
to stdin with the help of a reader thread and a ClusterShell Port
object."""
assert not sys.stdin.isatty()
# Create a ClusterShell Port object bound to worker's task. This object
# is able to receive messages in a thread-safe manner and then will safely
# trigger ev_msg() on a specified event handler.
port = worker.task.port(handler=StdInputHandler(worker), autoclose=True)
# Launch a dedicated thread to read stdin in blocking mode. Indeed stdin
# can be a file, so we cannot use a WorkerSimple here as polling on file
# may result in different behaviors depending on selected engine.
threading.Thread(None, _stdin_thread_start, args=(port,)).start()
def run_command(task, cmd, ns, timeout, display):
"""
Create and run the specified command line, displaying
results in a dshbak way when gathering is used.
"""
task.set_default("USER_running", True)
if display.verbosity >= VERB_VERB and task.topology:
print Display.COLOR_RESULT_FMT % '-' * 15
print Display.COLOR_RESULT_FMT % task.topology,
print Display.COLOR_RESULT_FMT % '-' * 15
if (display.gather or display.line_mode) and ns is not None:
if display.gather and display.line_mode:
handler = LiveGatherOutputHandler(display, ns)
else:
handler = GatherOutputHandler(display)
if display.verbosity == VERB_STD or display.verbosity == VERB_VERB:
handler.runtimer_init(task, len(ns))
worker = task.shell(cmd, nodes=ns, handler=handler, timeout=timeout)
else:
worker = task.shell(cmd, nodes=ns, handler=DirectOutputHandler(display),
timeout=timeout)
if ns is None:
worker.set_key('LOCAL')
if task.default("USER_stdin_worker"):
bind_stdin(worker)
task.resume()
def run_copy(task, sources, dest, ns, timeout, preserve_flag, display):
"""
run copy command
"""
task.set_default("USER_running", True)
task.set_default("USER_copies", len(sources))
copyhandler = CopyOutputHandler(display)
if display.verbosity == VERB_STD or display.verbosity == VERB_VERB:
copyhandler.runtimer_init(task, len(ns) * len(sources))
# Sources check
for source in sources:
if not os.path.exists(source):
display.vprint_err(VERB_QUIET, "ERROR: file \"%s\" not found" % \
source)
clush_exit(1)
task.copy(source, dest, ns, handler=copyhandler, timeout=timeout,
preserve=preserve_flag)
task.resume()
def run_rcopy(task, sources, dest, ns, timeout, preserve_flag, display):
"""
run reverse copy command
"""
task.set_default("USER_running", True)
task.set_default("USER_copies", len(sources))
# Sanity checks
if not os.path.exists(dest):
display.vprint_err(VERB_QUIET, "ERROR: directory \"%s\" not found" % \
dest)
clush_exit(1)
if not os.path.isdir(dest):
display.vprint_err(VERB_QUIET, \
"ERROR: destination \"%s\" is not a directory" % dest)
clush_exit(1)
copyhandler = CopyOutputHandler(display, True)
if display.verbosity == VERB_STD or display.verbosity == VERB_VERB:
copyhandler.runtimer_init(task, len(ns) * len(sources))
for source in sources:
task.rcopy(source, dest, ns, handler=copyhandler, timeout=timeout,
preserve=preserve_flag)
task.resume()
def set_fdlimit(fd_max, display):
"""Make open file descriptors soft limit the max."""
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
if hard < fd_max:
display.vprint(VERB_DEBUG, "Warning: Consider increasing max open " \
"files hard limit (%d)" % hard)
rlim_max = min(hard, fd_max)
if soft != rlim_max:
display.vprint(VERB_DEBUG, "Modifying max open files soft limit: " \
"%d -> %d" % (soft, rlim_max))
resource.setrlimit(resource.RLIMIT_NOFILE, (rlim_max, hard))
def clush_exit(status):
"""Flush stdio buffers and exit script."""
for stream in [sys.stdout, sys.stderr]:
stream.flush()
# Use os._exit to avoid threads cleanup
os._exit(status)
def clush_excepthook(extype, value, traceback):
"""Exceptions hook for clush: this method centralizes exception
handling from main thread and from (possible) separate task thread.
This hook has to be previously installed on startup by overriding
sys.excepthook and task.excepthook."""
try:
raise extype, value
except ClushConfigError, econf:
print >> sys.stderr, "ERROR: %s" % econf
except KeyboardInterrupt, kbe:
uncomp_nodes = getattr(kbe, 'uncompleted_nodes', None)
if uncomp_nodes:
print >> sys.stderr, \
"Keyboard interrupt (%s did not complete)." % uncomp_nodes
else:
print >> sys.stderr, "Keyboard interrupt."
clush_exit(128 + signal.SIGINT)
except OSError, value:
print >> sys.stderr, "ERROR: %s" % value
if value.errno == errno.EMFILE:
print >> sys.stderr, "ERROR: current `nofile' limits: " \
"soft=%d hard=%d" % resource.getrlimit(resource.RLIMIT_NOFILE)
clush_exit(1)
except GENERIC_ERRORS, exc:
clush_exit(handle_generic_error(exc))
# Error not handled
task_self().default_excepthook(extype, value, traceback)
def main():
"""clush script entry point"""
sys.excepthook = clush_excepthook
# Default values
nodeset_base, nodeset_exclude = NodeSet(), NodeSet()
#
# Argument management
#
usage = "%prog [options] command"
parser = OptionParser(usage)
parser.add_option("--nostdin", action="store_true", dest="nostdin",
help="don't watch for possible input from stdin")
parser.install_nodes_options()
parser.install_display_options(verbose_options=True)
parser.install_filecopy_options()
parser.install_ssh_options()
(options, args) = parser.parse_args()
#
# Load config file and apply overrides
#
config = ClushConfig(options)
# Should we use ANSI colors for nodes?
if config.color == "auto":
color = sys.stdout.isatty() and (options.gatherall or \
sys.stderr.isatty())
else:
color = config.color == "always"
try:
# Create and configure display object.
display = Display(options, config, color)
except ValueError, exc:
parser.error("option mismatch (%s)" % exc)
#
# Compute the nodeset
#
if options.nodes:
nodeset_base = NodeSet.fromlist(options.nodes)
if options.exclude:
nodeset_exclude = NodeSet.fromlist(options.exclude)
if options.groupsource:
# Be sure -a/g -s source work as espected.
RESOLVER_STD_GROUP.default_sourcename = options.groupsource
# FIXME: add public API to enforce engine
Task._std_default['engine'] = options.engine
# Do we have nodes group?
task = task_self()
task.set_info("debug", config.verbosity >= VERB_DEBUG)
if config.verbosity == VERB_DEBUG:
RESOLVER_STD_GROUP.set_verbosity(1)
if options.nodes_all:
all_nodeset = NodeSet.fromall()
display.vprint(VERB_DEBUG, "Adding nodes from option -a: %s" % \
all_nodeset)
nodeset_base.add(all_nodeset)
if options.group:
grp_nodeset = NodeSet.fromlist(options.group,
resolver=RESOLVER_NOGROUP)
for grp in grp_nodeset:
addingrp = NodeSet("@" + grp)
display.vprint(VERB_DEBUG, \
"Adding nodes from option -g %s: %s" % (grp, addingrp))
nodeset_base.update(addingrp)
if options.exgroup:
grp_nodeset = NodeSet.fromlist(options.exgroup,
resolver=RESOLVER_NOGROUP)
for grp in grp_nodeset:
removingrp = NodeSet("@" + grp)
display.vprint(VERB_DEBUG, \
"Excluding nodes from option -X %s: %s" % (grp, removingrp))
nodeset_exclude.update(removingrp)
# Do we have an exclude list? (-x ...)
nodeset_base.difference_update(nodeset_exclude)
if len(nodeset_base) < 1:
parser.error('No node to run on.')
# Set open files limit.
set_fdlimit(config.fd_max, display)
#
# Task management
#
# check for clush interactive mode
interactive = not len(args) and \
not (options.copy or options.rcopy)
# check for foreground ttys presence (input)
stdin_isafgtty = sys.stdin.isatty() and \
os.tcgetpgrp(sys.stdin.fileno()) == os.getpgrp()
# check for special condition (empty command and stdin not a tty)
if interactive and not stdin_isafgtty:
# looks like interactive but stdin is not a tty:
# switch to non-interactive + disable ssh pseudo-tty
interactive = False
# SSH: disable pseudo-tty allocation (-T)
ssh_options = config.ssh_options or ''
ssh_options += ' -T'
config._set_main("ssh_options", ssh_options)
if options.nostdin and interactive:
parser.error("illegal option `--nostdin' in that case")
# Force user_interaction if Clush._f_user_interaction for test purposes
user_interaction = hasattr(sys.modules[__name__], '_f_user_interaction')
if not options.nostdin:
# Try user interaction: check for foreground ttys presence (ouput)
stdout_isafgtty = sys.stdout.isatty() and \
os.tcgetpgrp(sys.stdout.fileno()) == os.getpgrp()
user_interaction |= stdin_isafgtty and stdout_isafgtty
display.vprint(VERB_DEBUG, "User interaction: %s" % user_interaction)
if user_interaction:
# Standard input is a terminal and we want to perform some user
# interactions in the main thread (using blocking calls), so
# we run cluster commands in a new ClusterShell Task (a new
# thread is created).
task = Task()
# else: perform everything in the main thread
# Handle special signal only when user_interaction is set
task.set_default("USER_handle_SIGUSR1", user_interaction)
task.excepthook = sys.excepthook
task.set_default("USER_stdin_worker", not (sys.stdin.isatty() or \
options.nostdin or \
user_interaction))
display.vprint(VERB_DEBUG, "Create STDIN worker: %s" % \
task.default("USER_stdin_worker"))
if config.verbosity >= VERB_DEBUG:
task.set_info("debug", True)
logging.basicConfig(level=logging.DEBUG)
logging.debug("clush: STARTING DEBUG")
task.set_info("fanout", config.fanout)
if options.topofile:
if config.verbosity >= VERB_VERB:
print Display.COLOR_RESULT_FMT % \
"Enabling TREE MODE (technology preview)"
task.set_default("auto_tree", True)
task.set_topology(options.topofile)
if options.grooming_delay:
if config.verbosity >= VERB_VERB:
print Display.COLOR_RESULT_FMT % ("Grooming delay: %f" % \
options.grooming_delay)
task.set_info("grooming_delay", options.grooming_delay)
if config.ssh_user:
task.set_info("ssh_user", config.ssh_user)
if config.ssh_path:
task.set_info("ssh_path", config.ssh_path)
if config.ssh_options:
task.set_info("ssh_options", config.ssh_options)
# Set detailed timeout values
task.set_info("connect_timeout", config.connect_timeout)
command_timeout = config.command_timeout
task.set_info("command_timeout", command_timeout)
# Enable stdout/stderr separation
task.set_default("stderr", not options.gatherall)
# Disable MsgTree buffering if not gathering outputs
task.set_default("stdout_msgtree", display.gather or display.line_mode)
# Always disable stderr MsgTree buffering
task.set_default("stderr_msgtree", False)
# Set timeout at worker level when command_timeout is defined.
if command_timeout > 0:
timeout = command_timeout
else:
timeout = -1
# Configure task custom status
task.set_default("USER_interactive", interactive)
task.set_default("USER_running", False)
if (options.copy or options.rcopy) and not args:
parser.error("--[r]copy option requires at least one argument")
if options.copy:
if not options.dest_path:
options.dest_path = os.path.dirname(os.path.abspath(args[0]))
op = "copy sources=%s dest=%s" % (args, options.dest_path)
elif options.rcopy:
if not options.dest_path:
options.dest_path = os.path.dirname(os.path.abspath(args[0]))
op = "rcopy sources=%s dest=%s" % (args, options.dest_path)
else:
op = "command=\"%s\"" % ' '.join(args)
# print debug values (fanout value is get from the config object
# and not task itself as set_info() is an asynchronous call.
display.vprint(VERB_DEBUG, "clush: nodeset=%s fanout=%d [timeout " \
"conn=%.1f cmd=%.1f] %s" % (nodeset_base, config.fanout,
task.info("connect_timeout"),
task.info("command_timeout"),
op))
if not task.default("USER_interactive"):
if options.copy:
run_copy(task, args, options.dest_path, nodeset_base, 0,
options.preserve_flag, display)
elif options.rcopy:
run_rcopy(task, args, options.dest_path, nodeset_base, 0,
options.preserve_flag, display)
else:
run_command(task, ' '.join(args), nodeset_base, timeout, display)
if user_interaction:
ttyloop(task, nodeset_base, timeout, display)
elif task.default("USER_interactive"):
display.vprint_err(VERB_QUIET, \
"ERROR: interactive mode requires a tty")
clush_exit(1)
rc = 0
if options.maxrc:
# Instead of clush return code, return commands retcode
rc = task.max_retcode()
if task.num_timeout() > 0:
rc = 255
clush_exit(rc)
if __name__ == '__main__':
main()
|