src/nepi/resources/linux/node.py

   1 #
   2 #    NEPI, a framework to manage network experiments
   3 #    Copyright (C) 2013 INRIA
   4 #
   5 #    This program is free software: you can redistribute it and/or modify
   6 #    it under the terms of the GNU General Public License as published by
   7 #    the Free Software Foundation, either version 3 of the License, or
   8 #    (at your option) any later version.
   9 #
  10 #    This program is distributed in the hope that it will be useful,
  11 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #    GNU General Public License for more details.
  14 #
  15 #    You should have received a copy of the GNU General Public License
  16 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  19
  20 from nepi.execution.attribute import Attribute, Flags
  21 from nepi.execution.resource import ResourceManager, clsinit, ResourceState
  22 from nepi.resources.linux import rpmfuncs, debfuncs
  23 from nepi.util import sshfuncs, execfuncs
  24 from nepi.util.sshfuncs import ProcStatus
  25
  26 import collections
  27 import os
  28 import random
  29 import re
  30 import tempfile
  31 import time
  32 import threading
  33
  34 # TODO: Verify files and dirs exists already
  35 # TODO: Blacklist nodes!
  36 # TODO: Unify delays!!
  37 # TODO: Validate outcome of uploads!!
  38
  39 reschedule_delay = "0.5s"
  40
  41 class ExitCode:
  42     """
  43     Error codes that the rexitcode function can return if unable to
  44     check the exit code of a spawned process
  45     """
  46     FILENOTFOUND = -1
  47     CORRUPTFILE = -2
  48     ERROR = -3
  49     OK = 0
  50
  51 class OSType:
  52     """
  53     Supported flavors of Linux OS
  54     """
  55     FEDORA_12 = "f12"
  56     FEDORA_14 = "f14"
  57     FEDORA = "fedora"
  58     UBUNTU = "ubuntu"
  59     DEBIAN = "debian"
  60
  61 @clsinit
  62 class LinuxNode(ResourceManager):
  63     """
  64     .. class:: Class Args :
  65
  66         :param ec: The Experiment controller
  67         :type ec: ExperimentController
  68         :param guid: guid of the RM
  69         :type guid: int
  70
  71     .. note::
  72
  73         There are different ways in which commands can be executed using the
  74         LinuxNode interface (i.e. 'execute' - blocking and non blocking, 'run',
  75         'run_and_wait').
  76
  77         Brief explanation:
  78
  79             * 'execute' (blocking mode) :
  80
  81                      HOW IT WORKS: 'execute', forks a process and run the
  82                      command, synchronously, attached to the terminal, in
  83                      foreground.
  84                      The execute method will block until the command returns
  85                      the result on 'out', 'err' (so until it finishes executing).
  86
  87                      USAGE: short-lived commands that must be executed attached
  88                      to a terminal and in foreground, for which it IS necessary
  89                      to block until the command has finished (e.g. if you want
  90                      to run 'ls' or 'cat').
  91
  92             * 'execute' (NON blocking mode - blocking = False) :
  93
  94                     HOW IT WORKS: Same as before, except that execute method
  95                     will return immediately (even if command still running).
  96
  97                     USAGE: long-lived commands that must be executed attached
  98                     to a terminal and in foreground, but for which it is not
  99                     necessary to block until the command has finished. (e.g.
 100                     start an application using X11 forwarding)
 101
 102              * 'run' :
 103
 104                    HOW IT WORKS: Connects to the host ( using SSH if remote)
 105                    and launches the command in background, detached from any
 106                    terminal (daemonized), and returns. The command continues to
 107                    run remotely, but since it is detached from the terminal,
 108                    its pipes (stdin, stdout, stderr) can't be redirected to the
 109                    console (as normal non detached processes would), and so they
 110                    are explicitly redirected to files. The pidfile is created as
 111                    part of the process of launching the command. The pidfile
 112                    holds the pid and ppid of the process forked in background,
 113                    so later on it is possible to check whether the command is still
 114                    running.
 115
 116                     USAGE: long-lived commands that can run detached in background,
 117                     for which it is NOT necessary to block (wait) until the command
 118                     has finished. (e.g. start an application that is not using X11
 119                     forwarding. It can run detached and remotely in background)
 120
 121              * 'run_and_wait' :
 122
 123                     HOW IT WORKS: Similar to 'run' except that it 'blocks' until
 124                     the command has finished execution. It also checks whether
 125                     errors occurred during runtime by reading the exitcode file,
 126                     which contains the exit code of the command that was run
 127                     (checking stderr only is not always reliable since many
 128                     commands throw debugging info to stderr and the only way to
 129                     automatically know whether an error really happened is to
 130                     check the process exit code).
 131
 132                     Another difference with respect to 'run', is that instead
 133                     of directly executing the command as a bash command line,
 134                     it uploads the command to a bash script and runs the script.
 135                     This allows to use the bash script to debug errors, since
 136                     it remains at the remote host and can be run manually to
 137                     reproduce the error.
 138
 139                     USAGE: medium-lived commands that can run detached in
 140                     background, for which it IS necessary to block (wait) until
 141                     the command has finished. (e.g. Package installation,
 142                     source compilation, file download, etc)
 143
 144     """
 145     _rtype = "LinuxNode"
 146
 147     @classmethod
 148     def _register_attributes(cls):
 149         hostname = Attribute("hostname", "Hostname of the machine",
 150                 flags = Flags.ExecReadOnly)
 151
 152         username = Attribute("username", "Local account username",
 153                 flags = Flags.Credential)
 154
 155         port = Attribute("port", "SSH port", flags = Flags.ExecReadOnly)
 156
 157         home = Attribute("home",
 158                 "Experiment home directory to store all experiment related files",
 159                 flags = Flags.ExecReadOnly)
 160
 161         identity = Attribute("identity", "SSH identity file",
 162                 flags = Flags.Credential)
 163
 164         server_key = Attribute("serverKey", "Server public key",
 165                 flags = Flags.ExecReadOnly)
 166
 167         clean_home = Attribute("cleanHome", "Remove all files and directories " + \
 168                 " from home folder before starting experiment",
 169                 flags = Flags.ExecReadOnly)
 170
 171         clean_processes = Attribute("cleanProcesses",
 172                 "Kill all running processes before starting experiment",
 173                 flags = Flags.ExecReadOnly)
 174
 175         tear_down = Attribute("tearDown", "Bash script to be executed before " + \
 176                 "releasing the resource",
 177                 flags = Flags.ExecReadOnly)
 178
 179         cls._register_attribute(hostname)
 180         cls._register_attribute(username)
 181         cls._register_attribute(port)
 182         cls._register_attribute(home)
 183         cls._register_attribute(identity)
 184         cls._register_attribute(server_key)
 185         cls._register_attribute(clean_home)
 186         cls._register_attribute(clean_processes)
 187         cls._register_attribute(tear_down)
 188
 189     def __init__(self, ec, guid):
 190         super(LinuxNode, self).__init__(ec, guid)
 191         self._os = None
 192
 193         # lock to avoid concurrency issues on methods used by applications
 194         self._lock = threading.Lock()
 195
 196     def log_message(self, msg):
 197         return " guid %d - host %s - %s " % (self.guid,
 198                 self.get("hostname"), msg)
 199
 200     @property
 201     def home(self):
 202         return self.get("home") or ""
 203
 204     @property
 205     def exp_home(self):
 206         return os.path.join(self.home, self.ec.exp_id)
 207
 208     @property
 209     def node_home(self):
 210         node_home = "node-%d" % self.guid
 211         return os.path.join(self.exp_home, node_home)
 212
 213     @property
 214     def os(self):
 215         if self._os:
 216             return self._os
 217
 218         if (not self.get("hostname") or not self.get("username")):
 219             msg = "Can't resolve OS, insufficient data "
 220             self.error(msg)
 221             raise RuntimeError, msg
 222
 223         (out, err), proc = self.execute("cat /etc/issue", with_lock = True)
 224
 225         if err and proc.poll():
 226             msg = "Error detecting OS "
 227             self.error(msg, out, err)
 228             raise RuntimeError, "%s - %s - %s" %( msg, out, err )
 229
 230         if out.find("Fedora release 12") == 0:
 231             self._os = OSType.FEDORA_12
 232         elif out.find("Fedora release 14") == 0:
 233             self._os = OSType.FEDORA_14
 234         elif out.find("Debian") == 0:
 235             self._os = OSType.DEBIAN
 236         elif out.find("Ubuntu") ==0:
 237             self._os = OSType.UBUNTU
 238         else:
 239             msg = "Unsupported OS"
 240             self.error(msg, out)
 241             raise RuntimeError, "%s - %s " %( msg, out )
 242
 243         return self._os
 244
 245     @property
 246     def localhost(self):
 247         return self.get("hostname") in ['localhost', '127.0.0.7', '::1']
 248
 249     def provision(self):
 250         if not self.is_alive():
 251             self._state = ResourceState.FAILED
 252             msg = "Deploy failed. Unresponsive node %s" % self.get("hostname")
 253             self.error(msg)
 254             raise RuntimeError, msg
 255
 256         if self.get("cleanProcesses"):
 257             self.clean_processes()
 258
 259         if self.get("cleanHome"):
 260             self.clean_home()
 261
 262         self.mkdir(self.node_home)
 263
 264         super(LinuxNode, self).provision()
 265
 266     def deploy(self):
 267         if self.state == ResourceState.NEW:
 268             try:
 269                 self.discover()
 270                 self.provision()
 271             except:
 272                 self._state = ResourceState.FAILED
 273                 raise
 274
 275         # Node needs to wait until all associated interfaces are
 276         # ready before it can finalize deployment
 277         from nepi.resources.linux.interface import LinuxInterface
 278         ifaces = self.get_connected(LinuxInterface.rtype())
 279         for iface in ifaces:
 280             if iface.state < ResourceState.READY:
 281                 self.ec.schedule(reschedule_delay, self.deploy)
 282                 return
 283
 284         super(LinuxNode, self).deploy()
 285
 286     def release(self):
 287         tear_down = self.get("tearDown")
 288         if tear_down:
 289             self.execute(tear_down)
 290
 291         super(LinuxNode, self).release()
 292
 293     def valid_connection(self, guid):
 294         # TODO: Validate!
 295         return True
 296
 297     def clean_processes(self, killer = False):
 298         self.info("Cleaning up processes")
 299
 300         if killer:
 301             # Hardcore kill
 302             cmd = ("sudo -S killall python tcpdump || /bin/true ; " +
 303                 "sudo -S killall python tcpdump || /bin/true ; " +
 304                 "sudo -S kill $(ps -N -T -o pid --no-heading | grep -v $PPID | sort) || /bin/true ; " +
 305                 "sudo -S killall -u root || /bin/true ; " +
 306                 "sudo -S killall -u root || /bin/true ; ")
 307         else:
 308             # Be gentler...
 309             cmd = ("sudo -S killall tcpdump || /bin/true ; " +
 310                 "sudo -S killall tcpdump || /bin/true ; " +
 311                 "sudo -S killall -u %s || /bin/true ; " % self.get("username") +
 312                 "sudo -S killall -u %s || /bin/true ; " % self.get("username"))
 313
 314         out = err = ""
 315         (out, err), proc = self.execute(cmd, retry = 1, with_lock = True)
 316
 317     def clean_home(self):
 318         self.info("Cleaning up home")
 319
 320         cmd = (
 321             # "find . -maxdepth 1  \( -name '.cache' -o -name '.local' -o -name '.config' -o -name 'nepi-*' \)" +
 322             "find . -maxdepth 1 -name 'nepi-*' " +
 323             " -execdir rm -rf {} + "
 324             )
 325
 326         if self.home:
 327             cmd = "cd %s ; " % self.home + cmd
 328
 329         out = err = ""
 330         (out, err), proc = self.execute(cmd, with_lock = True)
 331
 332     def upload(self, src, dst, text = False):
 333         """ Copy content to destination
 334
 335            src  content to copy. Can be a local file, directory or a list of files
 336
 337            dst  destination path on the remote host (remote is always self.host)
 338
 339            text src is text input, it must be stored into a temp file before uploading
 340         """
 341         # If source is a string input
 342         f = None
 343         if text and not os.path.isfile(src):
 344             # src is text input that should be uploaded as file
 345             # create a temporal file with the content to upload
 346             f = tempfile.NamedTemporaryFile(delete=False)
 347             f.write(src)
 348             f.close()
 349             src = f.name
 350
 351         if not self.localhost:
 352             # Build destination as <user>@<server>:<path>
 353             dst = "%s@%s:%s" % (self.get("username"), self.get("hostname"), dst)
 354         result = self.copy(src, dst)
 355
 356         # clean up temp file
 357         if f:
 358             os.remove(f.name)
 359
 360         return result
 361
 362     def download(self, src, dst):
 363         if not self.localhost:
 364             # Build destination as <user>@<server>:<path>
 365             src = "%s@%s:%s" % (self.get("username"), self.get("hostname"), src)
 366         return self.copy(src, dst)
 367
 368     def install_packages(self, packages, home):
 369         command = ""
 370         if self.os in [OSType.FEDORA_12, OSType.FEDORA_14, OSType.FEDORA]:
 371             command = rpmfuncs.install_packages_command(self.os, packages)
 372         elif self.os in [OSType.DEBIAN, OSType.UBUNTU]:
 373             command = debfuncs.install_packages_command(self.os, packages)
 374         else:
 375             msg = "Error installing packages ( OS not known ) "
 376             self.error(msg, self.os)
 377             raise RuntimeError, msg
 378
 379         out = err = ""
 380         (out, err), proc = self.run_and_wait(command, home,
 381             shfile = "instpkg.sh",
 382             pidfile = "instpkg_pidfile",
 383             ecodefile = "instpkg_exitcode",
 384             stdout = "instpkg_stdout",
 385             stderr = "instpkg_stderr",
 386             raise_on_error = True)
 387
 388         return (out, err), proc
 389
 390     def remove_packages(self, packages, home):
 391         command = ""
 392         if self.os in [OSType.FEDORA_12, OSType.FEDORA_14, OSType.FEDORA]:
 393             command = rpmfuncs.remove_packages_command(self.os, packages)
 394         elif self.os in [OSType.DEBIAN, OSType.UBUNTU]:
 395             command = debfuncs.remove_packages_command(self.os, packages)
 396         else:
 397             msg = "Error removing packages ( OS not known ) "
 398             self.error(msg)
 399             raise RuntimeError, msg
 400
 401         out = err = ""
 402         (out, err), proc = self.run_and_wait(command, home,
 403             shfile = "rmpkg.sh",
 404             pidfile = "rmpkg_pidfile",
 405             ecodefile = "rmpkg_exitcode",
 406             stdout = "rmpkg_stdout",
 407             stderr = "rmpkg_stderr",
 408             raise_on_error = True)
 409
 410         return (out, err), proc
 411
 412     def mkdir(self, path, clean = False):
 413         if clean:
 414             self.rmdir(path)
 415
 416         return self.execute("mkdir -p %s" % path, with_lock = True)
 417
 418     def rmdir(self, path):
 419         return self.execute("rm -rf %s" % path, with_lock = True)
 420
 421     def run_and_wait(self, command, home,
 422             shfile = "cmd.sh",
 423             env = None,
 424             pidfile = "pidfile",
 425             ecodefile = "exitcode",
 426             stdin = None,
 427             stdout = "stdout",
 428             stderr = "stderr",
 429             sudo = False,
 430             tty = False,
 431             raise_on_error = False):
 432         """
 433         runs a command in background on the remote host, busy-waiting
 434         until the command finishes execution.
 435         This is more robust than doing a simple synchronized 'execute',
 436         since in the remote host the command can continue to run detached
 437         even if network disconnections occur
 438         """
 439         self.upload_command(command, home,
 440             shfile = shfile,
 441             ecodefile = ecodefile,
 442             env = env)
 443
 444         command = "bash ./%s" % shfile
 445         # run command in background in remote host
 446         (out, err), proc = self.run(command, home,
 447                 pidfile = pidfile,
 448                 stdin = stdin,
 449                 stdout = stdout,
 450                 stderr = stderr,
 451                 sudo = sudo,
 452                 tty = tty)
 453
 454         # check no errors occurred
 455         if proc.poll():
 456             msg = " Failed to run command '%s' " % command
 457             self.error(msg, out, err)
 458             if raise_on_error:
 459                 raise RuntimeError, msg
 460
 461         # Wait for pid file to be generated
 462         pid, ppid = self.wait_pid(
 463                 home = home,
 464                 pidfile = pidfile,
 465                 raise_on_error = raise_on_error)
 466
 467         # wait until command finishes to execute
 468         self.wait_run(pid, ppid)
 469
 470         (out, err), proc = self.check_errors(home,
 471             ecodefile = ecodefile,
 472             stdout = stdout,
 473             stderr= stderr)
 474
 475         # Out is what was written in the stderr file
 476         if err:
 477             msg = " Failed to run command '%s' " % command
 478             self.error(msg, out, err)
 479
 480             if raise_on_error:
 481                 raise RuntimeError, msg
 482
 483         return (out, err), proc
 484
 485     def exitcode(self, home, ecodefile = "exitcode"):
 486         """
 487         Get the exit code of an application.
 488         Returns an integer value with the exit code
 489         """
 490         (out, err), proc = self.check_output(home, ecodefile)
 491
 492         # Succeeded to open file, return exit code in the file
 493         if proc.wait() == 0:
 494             try:
 495                 return int(out.strip())
 496             except:
 497                 # Error in the content of the file!
 498                 return ExitCode.CORRUPTFILE
 499
 500         # No such file or directory
 501         if proc.returncode == 1:
 502             return ExitCode.FILENOTFOUND
 503
 504         # Other error from 'cat'
 505         return ExitCode.ERROR
 506
 507     def upload_command(self, command, home,
 508             shfile = "cmd.sh",
 509             ecodefile = "exitcode",
 510             env = None):
 511         """ Saves the command as a bash script file in the remote host, and
 512         forces to save the exit code of the command execution to the ecodefile
 513         """
 514
 515         if not (command.strip().endswith(";") or command.strip().endswith("&")):
 516             command += ";"
 517
 518         # The exit code of the command will be stored in ecodefile
 519         command = " { %(command)s } ; echo $? > %(ecodefile)s ;" % {
 520                 'command': command,
 521                 'ecodefile': ecodefile,
 522                 }
 523
 524         # Export environment
 525         environ = self.format_environment(env)
 526
 527         # Add environ to command
 528         command = environ + command
 529
 530         dst = os.path.join(home, shfile)
 531         return self.upload(command, dst, text = True)
 532
 533     def format_environment(self, env, inline = False):
 534         """Format environmental variables for command to be executed either
 535         as an inline command
 536         (i.e. export PYTHONPATH=src/..; export LALAL= ..;python script.py) or
 537         as a bash script (i.e. export PYTHONPATH=src/.. \n export LALA=.. \n)
 538         """
 539         if not env: return ""
 540         env = env.strip()
 541
 542         sep = ";" if inline else "\n"
 543         return sep.join(map(lambda e: " export %s" % e, env.split(" "))) + sep
 544
 545     def check_errors(self, home,
 546             ecodefile = "exitcode",
 547             stdout = "stdout",
 548             stderr = "stderr"):
 549         """
 550         Checks whether errors occurred while running a command.
 551         It first checks the exit code for the command, and only if the
 552         exit code is an error one it returns the error output.
 553
 554         """
 555         proc = None
 556         err = ""
 557         # retrive standard output from the file
 558         (out, oerr), oproc = self.check_output(home, stdout)
 559
 560         # get exit code saved in the 'exitcode' file
 561         ecode = self.exitcode(home, ecodefile)
 562
 563         if ecode in [ ExitCode.CORRUPTFILE, ExitCode.ERROR ]:
 564             err = "Error retrieving exit code status from file %s/%s" % (home, ecodefile)
 565         elif ecode > 0 or ecode == ExitCode.FILENOTFOUND:
 566             # The process returned an error code or didn't exist.
 567             # Check standard error.
 568             (err, eerr), proc = self.check_output(home, stderr)
 569
 570             # If the stderr file was not found, assume nothing bad happened,
 571             # and just ignore the error.
 572             # (cat returns 1 for error "No such file or directory")
 573             if ecode == ExitCode.FILENOTFOUND and proc.poll() == 1:
 574                 err = ""
 575
 576         return (out, err), proc
 577
 578     def wait_pid(self, home, pidfile = "pidfile", raise_on_error = False):
 579         """ Waits until the pid file for the command is generated,
 580             and returns the pid and ppid of the process """
 581         pid = ppid = None
 582         delay = 1.0
 583
 584         for i in xrange(4):
 585             pidtuple = self.getpid(home = home, pidfile = pidfile)
 586
 587             if pidtuple:
 588                 pid, ppid = pidtuple
 589                 break
 590             else:
 591                 time.sleep(delay)
 592                 delay = delay * 1.5
 593         else:
 594             msg = " Failed to get pid for pidfile %s/%s " % (
 595                     home, pidfile )
 596             self.error(msg)
 597
 598             if raise_on_error:
 599                 raise RuntimeError, msg
 600
 601         return pid, ppid
 602
 603     def wait_run(self, pid, ppid, trial = 0):
 604         """ wait for a remote process to finish execution """
 605         start_delay = 1.0
 606
 607         while True:
 608             status = self.status(pid, ppid)
 609
 610             if status is ProcStatus.FINISHED:
 611                 break
 612             elif status is not ProcStatus.RUNNING:
 613                 delay = delay * 1.5
 614                 time.sleep(delay)
 615                 # If it takes more than 20 seconds to start, then
 616                 # asume something went wrong
 617                 if delay > 20:
 618                     break
 619             else:
 620                 # The app is running, just wait...
 621                 time.sleep(0.5)
 622
 623     def check_output(self, home, filename):
 624         """ Retrives content of file """
 625         (out, err), proc = self.execute("cat %s" %
 626             os.path.join(home, filename), retry = 1, with_lock = True)
 627         return (out, err), proc
 628
 629     def is_alive(self):
 630         if self.localhost:
 631             return True
 632
 633         out = err = ""
 634         try:
 635             # TODO: FIX NOT ALIVE!!!!
 636             (out, err), proc = self.execute("echo 'ALIVE' || (echo 'NOTALIVE') >&2", retry = 5,
 637                     with_lock = True)
 638         except:
 639             import traceback
 640             trace = traceback.format_exc()
 641             msg = "Unresponsive host  %s " % err
 642             self.error(msg, out, trace)
 643             return False
 644
 645         if out.strip().startswith('ALIVE'):
 646             return True
 647         else:
 648             msg = "Unresponsive host "
 649             self.error(msg, out, err)
 650             return False
 651
 652     def copy(self, src, dst):
 653         if self.localhost:
 654             (out, err), proc = execfuncs.lcopy(source, dest,
 655                     recursive = True,
 656                     strict_host_checking = False)
 657         else:
 658             with self._lock:
 659                 (out, err), proc = sshfuncs.rcopy(
 660                     src, dst,
 661                     port = self.get("port"),
 662                     identity = self.get("identity"),
 663                     server_key = self.get("serverKey"),
 664                     recursive = True,
 665                     strict_host_checking = False)
 666
 667         return (out, err), proc
 668
 669     def execute(self, command,
 670             sudo = False,
 671             stdin = None,
 672             env = None,
 673             tty = False,
 674             forward_x11 = False,
 675             timeout = None,
 676             retry = 3,
 677             err_on_timeout = True,
 678             connect_timeout = 30,
 679             strict_host_checking = False,
 680             persistent = True,
 681             blocking = True,
 682             with_lock = False
 683             ):
 684         """ Notice that this invocation will block until the
 685         execution finishes. If this is not the desired behavior,
 686         use 'run' instead."""
 687
 688         if self.localhost:
 689             (out, err), proc = execfuncs.lexec(command,
 690                     user = user,
 691                     sudo = sudo,
 692                     stdin = stdin,
 693                     env = env)
 694         else:
 695             if with_lock:
 696                 with self._lock:
 697                     (out, err), proc = sshfuncs.rexec(
 698                         command,
 699                         host = self.get("hostname"),
 700                         user = self.get("username"),
 701                         port = self.get("port"),
 702                         agent = True,
 703                         sudo = sudo,
 704                         stdin = stdin,
 705                         identity = self.get("identity"),
 706                         server_key = self.get("serverKey"),
 707                         env = env,
 708                         tty = tty,
 709                         forward_x11 = forward_x11,
 710                         timeout = timeout,
 711                         retry = retry,
 712                         err_on_timeout = err_on_timeout,
 713                         connect_timeout = connect_timeout,
 714                         persistent = persistent,
 715                         blocking = blocking,
 716                         strict_host_checking = strict_host_checking
 717                         )
 718             else:
 719                 (out, err), proc = sshfuncs.rexec(
 720                     command,
 721                     host = self.get("hostname"),
 722                     user = self.get("username"),
 723                     port = self.get("port"),
 724                     agent = True,
 725                     sudo = sudo,
 726                     stdin = stdin,
 727                     identity = self.get("identity"),
 728                     server_key = self.get("serverKey"),
 729                     env = env,
 730                     tty = tty,
 731                     forward_x11 = forward_x11,
 732                     timeout = timeout,
 733                     retry = retry,
 734                     err_on_timeout = err_on_timeout,
 735                     connect_timeout = connect_timeout,
 736                     persistent = persistent,
 737                     blocking = blocking,
 738                     strict_host_checking = strict_host_checking
 739                     )
 740
 741         return (out, err), proc
 742
 743     def run(self, command, home,
 744             create_home = False,
 745             pidfile = 'pidfile',
 746             stdin = None,
 747             stdout = 'stdout',
 748             stderr = 'stderr',
 749             sudo = False,
 750             tty = False):
 751
 752         self.debug("Running command '%s'" % command)
 753
 754         if self.localhost:
 755             (out, err), proc = execfuncs.lspawn(command, pidfile,
 756                     stdout = stdout,
 757                     stderr = stderr,
 758                     stdin = stdin,
 759                     home = home,
 760                     create_home = create_home,
 761                     sudo = sudo,
 762                     user = user)
 763         else:
 764             with self._lock:
 765                 (out, err), proc = sshfuncs.rspawn(
 766                     command,
 767                     pidfile = pidfile,
 768                     home = home,
 769                     create_home = create_home,
 770                     stdin = stdin if stdin is not None else '/dev/null',
 771                     stdout = stdout if stdout else '/dev/null',
 772                     stderr = stderr if stderr else '/dev/null',
 773                     sudo = sudo,
 774                     host = self.get("hostname"),
 775                     user = self.get("username"),
 776                     port = self.get("port"),
 777                     agent = True,
 778                     identity = self.get("identity"),
 779                     server_key = self.get("serverKey"),
 780                     tty = tty
 781                     )
 782
 783         return (out, err), proc
 784
 785     def getpid(self, home, pidfile = "pidfile"):
 786         if self.localhost:
 787             pidtuple =  execfuncs.lgetpid(os.path.join(home, pidfile))
 788         else:
 789             with self._lock:
 790                 pidtuple = sshfuncs.rgetpid(
 791                     os.path.join(home, pidfile),
 792                     host = self.get("hostname"),
 793                     user = self.get("username"),
 794                     port = self.get("port"),
 795                     agent = True,
 796                     identity = self.get("identity"),
 797                     server_key = self.get("serverKey")
 798                     )
 799
 800         return pidtuple
 801
 802     def status(self, pid, ppid):
 803         if self.localhost:
 804             status = execfuncs.lstatus(pid, ppid)
 805         else:
 806             with self._lock:
 807                 status = sshfuncs.rstatus(
 808                         pid, ppid,
 809                         host = self.get("hostname"),
 810                         user = self.get("username"),
 811                         port = self.get("port"),
 812                         agent = True,
 813                         identity = self.get("identity"),
 814                         server_key = self.get("serverKey")
 815                         )
 816
 817         return status
 818
 819     def kill(self, pid, ppid, sudo = False):
 820         out = err = ""
 821         proc = None
 822         status = self.status(pid, ppid)
 823
 824         if status == sshfuncs.ProcStatus.RUNNING:
 825             if self.localhost:
 826                 (out, err), proc = execfuncs.lkill(pid, ppid, sudo)
 827             else:
 828                 with self._lock:
 829                     (out, err), proc = sshfuncs.rkill(
 830                         pid, ppid,
 831                         host = self.get("hostname"),
 832                         user = self.get("username"),
 833                         port = self.get("port"),
 834                         agent = True,
 835                         sudo = sudo,
 836                         identity = self.get("identity"),
 837                         server_key = self.get("serverKey")
 838                         )
 839
 840         return (out, err), proc
 841