+#
+# NEPI, a framework to manage network experiments
+# Copyright (C) 2013 INRIA
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 as
+# published by the Free Software Foundation;
+#
+# This program 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 General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+# Author: Alina Quereilhac <alina.quereilhac@inria.fr>
+
+from nepi.execution.attribute import Attribute, Flags, Types
+from nepi.execution.resource import ResourceManager, clsinit_copy, \
+ ResourceState
+from nepi.resources.linux import rpmfuncs, debfuncs
+from nepi.util import sshfuncs, execfuncs
+from nepi.util.sshfuncs import ProcStatus
+
+import collections
+import os
+import random
+import re
+import tempfile
+import time
+import threading
+import traceback
+
+# TODO: Unify delays!!
+# TODO: Validate outcome of uploads!!
+
+[docs]class ExitCode:
+
"""
+
Error codes that the rexitcode function can return if unable to
+
check the exit code of a spawned process
+
"""
+
FILENOTFOUND = -1
+
CORRUPTFILE = -2
+
ERROR = -3
+
OK = 0
+
+[docs]class OSType:
+
"""
+
Supported flavors of Linux OS
+
"""
+
DEBIAN = 1
+
UBUNTU = 1 << 1
+
FEDORA = 1 << 2
+
FEDORA_8 = 1 << 3 | FEDORA
+
FEDORA_12 = 1 << 4 | FEDORA
+
FEDORA_14 = 1 << 5 | FEDORA
+
+@clsinit_copy
+[docs]class LinuxNode(ResourceManager):
+
"""
+
.. class:: Class Args :
+
+
:param ec: The Experiment controller
+
:type ec: ExperimentController
+
:param guid: guid of the RM
+
:type guid: int
+
+
.. note::
+
+
There are different ways in which commands can be executed using the
+
LinuxNode interface (i.e. 'execute' - blocking and non blocking, 'run',
+
'run_and_wait').
+
+
Brief explanation:
+
+
* 'execute' (blocking mode) :
+
+
HOW IT WORKS: 'execute', forks a process and run the
+
command, synchronously, attached to the terminal, in
+
foreground.
+
The execute method will block until the command returns
+
the result on 'out', 'err' (so until it finishes executing).
+
+
USAGE: short-lived commands that must be executed attached
+
to a terminal and in foreground, for which it IS necessary
+
to block until the command has finished (e.g. if you want
+
to run 'ls' or 'cat').
+
+
* 'execute' (NON blocking mode - blocking = False) :
+
+
HOW IT WORKS: Same as before, except that execute method
+
will return immediately (even if command still running).
+
+
USAGE: long-lived commands that must be executed attached
+
to a terminal and in foreground, but for which it is not
+
necessary to block until the command has finished. (e.g.
+
start an application using X11 forwarding)
+
+
* 'run' :
+
+
HOW IT WORKS: Connects to the host ( using SSH if remote)
+
and launches the command in background, detached from any
+
terminal (daemonized), and returns. The command continues to
+
run remotely, but since it is detached from the terminal,
+
its pipes (stdin, stdout, stderr) can't be redirected to the
+
console (as normal non detached processes would), and so they
+
are explicitly redirected to files. The pidfile is created as
+
part of the process of launching the command. The pidfile
+
holds the pid and ppid of the process forked in background,
+
so later on it is possible to check whether the command is still
+
running.
+
+
USAGE: long-lived commands that can run detached in background,
+
for which it is NOT necessary to block (wait) until the command
+
has finished. (e.g. start an application that is not using X11
+
forwarding. It can run detached and remotely in background)
+
+
* 'run_and_wait' :
+
+
HOW IT WORKS: Similar to 'run' except that it 'blocks' until
+
the command has finished execution. It also checks whether
+
errors occurred during runtime by reading the exitcode file,
+
which contains the exit code of the command that was run
+
(checking stderr only is not always reliable since many
+
commands throw debugging info to stderr and the only way to
+
automatically know whether an error really happened is to
+
check the process exit code).
+
+
Another difference with respect to 'run', is that instead
+
of directly executing the command as a bash command line,
+
it uploads the command to a bash script and runs the script.
+
This allows to use the bash script to debug errors, since
+
it remains at the remote host and can be run manually to
+
reproduce the error.
+
+
USAGE: medium-lived commands that can run detached in
+
background, for which it IS necessary to block (wait) until
+
the command has finished. (e.g. Package installation,
+
source compilation, file download, etc)
+
+
"""
+
_rtype = "linux::Node"
+
_help = "Controls Linux host machines ( either localhost or a host " \
+
"that can be accessed using a SSH key)"
+
_platform = "linux"
+
+
@classmethod
+
def _register_attributes(cls):
+
hostname = Attribute("hostname", "Hostname of the machine",
+
flags = Flags.Design)
+
+
username = Attribute("username", "Local account username",
+
flags = Flags.Credential)
+
+
port = Attribute("port", "SSH port", flags = Flags.Design)
+
+
home = Attribute("home",
+
"Experiment home directory to store all experiment related files",
+
flags = Flags.Design)
+
+
identity = Attribute("identity", "SSH identity file",
+
flags = Flags.Credential)
+
+
server_key = Attribute("serverKey", "Server public key",
+
flags = Flags.Design)
+
+
clean_home = Attribute("cleanHome", "Remove all nepi files and directories "
+
" from node home folder before starting experiment",
+
type = Types.Bool,
+
default = False,
+
flags = Flags.Design)
+
+
clean_experiment = Attribute("cleanExperiment", "Remove all files and directories "
+
" from a previous same experiment, before the new experiment starts",
+
type = Types.Bool,
+
default = False,
+
flags = Flags.Design)
+
+
clean_processes = Attribute("cleanProcesses",
+
"Kill all running processes before starting experiment",
+
type = Types.Bool,
+
default = False,
+
flags = Flags.Design)
+
+
tear_down = Attribute("tearDown", "Bash script to be executed before " + \
+
"releasing the resource",
+
flags = Flags.Design)
+
+
gateway_user = Attribute("gatewayUser", "Gateway account username",
+
flags = Flags.Design)
+
+
gateway = Attribute("gateway", "Hostname of the gateway machine",
+
flags = Flags.Design)
+
+
ip = Attribute("ip", "Linux host public IP address. "
+
"Must not be modified by the user unless hostname is 'localhost'",
+
flags = Flags.Design)
+
+
cls._register_attribute(hostname)
+
cls._register_attribute(username)
+
cls._register_attribute(port)
+
cls._register_attribute(home)
+
cls._register_attribute(identity)
+
cls._register_attribute(server_key)
+
cls._register_attribute(clean_home)
+
cls._register_attribute(clean_experiment)
+
cls._register_attribute(clean_processes)
+
cls._register_attribute(tear_down)
+
cls._register_attribute(gateway_user)
+
cls._register_attribute(gateway)
+
cls._register_attribute(ip)
+
+
def __init__(self, ec, guid):
+
super(LinuxNode, self).__init__(ec, guid)
+
self._os = None
+
# home directory at Linux host
+
self._home_dir = ""
+
+
# lock to prevent concurrent applications on the same node,
+
# to execute commands at the same time. There are potential
+
# concurrency issues when using SSH to a same host from
+
# multiple threads. There are also possible operational
+
# issues, e.g. an application querying the existence
+
# of a file or folder prior to its creation, and another
+
# application creating the same file or folder in between.
+
self._node_lock = threading.Lock()
+
+
[docs] def log_message(self, msg):
+
return " guid %d - host %s - %s " % (self.guid,
+
self.get("hostname"), msg)
+
+
@property
+
[docs] def home_dir(self):
+
home = self.get("home") or ""
+
if not home.startswith("/"):
+
home = os.path.join(self._home_dir, home)
+
return home
+
+
@property
+
[docs] def nepi_home(self):
+
return os.path.join(self.home_dir, ".nepi")
+
+
@property
+
[docs] def usr_dir(self):
+
return os.path.join(self.nepi_home, "nepi-usr")
+
+
@property
+
[docs] def lib_dir(self):
+
return os.path.join(self.usr_dir, "lib")
+
+
@property
+
[docs] def bin_dir(self):
+
return os.path.join(self.usr_dir, "bin")
+
+
@property
+
[docs] def src_dir(self):
+
return os.path.join(self.usr_dir, "src")
+
+
@property
+
[docs] def share_dir(self):
+
return os.path.join(self.usr_dir, "share")
+
+
@property
+
[docs] def exp_dir(self):
+
return os.path.join(self.nepi_home, "nepi-exp")
+
+
@property
+
[docs] def exp_home(self):
+
return os.path.join(self.exp_dir, self.ec.exp_id)
+
+
@property
+
[docs] def node_home(self):
+
return os.path.join(self.exp_home, "node-%d" % self.guid)
+
+
@property
+
[docs] def run_home(self):
+
return os.path.join(self.node_home, self.ec.run_id)
+
+
@property
+
[docs] def os(self):
+
if self._os:
+
return self._os
+
+
if not self.localhost and not self.get("username"):
+
msg = "Can't resolve OS, insufficient data "
+
self.error(msg)
+
raise RuntimeError, msg
+
+
out = self.get_os()
+
+
if out.find("Debian") == 0:
+
self._os = OSType.DEBIAN
+
elif out.find("Ubuntu") ==0:
+
self._os = OSType.UBUNTU
+
elif out.find("Fedora release") == 0:
+
self._os = OSType.FEDORA
+
if out.find("Fedora release 8") == 0:
+
self._os = OSType.FEDORA_8
+
elif out.find("Fedora release 12") == 0:
+
self._os = OSType.FEDORA_12
+
elif out.find("Fedora release 14") == 0:
+
self._os = OSType.FEDORA_14
+
else:
+
msg = "Unsupported OS"
+
self.error(msg, out)
+
raise RuntimeError, "%s - %s " %( msg, out )
+
+
return self._os
+
+
[docs] def get_os(self):
+
# The underlying SSH layer will sometimes return an empty
+
# output (even if the command was executed without errors).
+
# To work arround this, repeat the operation N times or
+
# until the result is not empty string
+
out = ""
+
try:
+
(out, err), proc = self.execute("cat /etc/issue",
+
with_lock = True,
+
blocking = True)
+
except:
+
trace = traceback.format_exc()
+
msg = "Error detecting OS: %s " % trace
+
self.error(msg, out, err)
+
+
return out
+
+
@property
+
[docs] def use_deb(self):
+
return (self.os & (OSType.DEBIAN|OSType.UBUNTU))
+
+
@property
+
[docs] def use_rpm(self):
+
return (self.os & OSType.FEDORA)
+
+
@property
+
[docs] def localhost(self):
+
return self.get("hostname") in ['localhost', '127.0.0.1', '::1']
+
+
[docs] def do_provision(self):
+
# check if host is alive
+
if not self.is_alive():
+
msg = "Deploy failed. Unresponsive node %s" % self.get("hostname")
+
self.error(msg)
+
raise RuntimeError, msg
+
+
self.find_home()
+
+
if self.get("cleanProcesses"):
+
self.clean_processes()
+
+
if self.get("cleanHome"):
+
self.clean_home()
+
+
if self.get("cleanExperiment"):
+
self.clean_experiment()
+
+
# Create shared directory structure and node home directory
+
paths = [self.lib_dir,
+
self.bin_dir,
+
self.src_dir,
+
self.share_dir,
+
self.node_home]
+
+
self.mkdir(paths)
+
+
# Get Public IP address if possible
+
if not self.get("ip"):
+
try:
+
ip = sshfuncs.gethostbyname(self.get("hostname"))
+
self.set("ip", ip)
+
except:
+
if self.get("gateway") is None:
+
msg = "Local DNS can not resolve hostname %s" % self.get("hostname")
+
self.error(msg)
+
+
super(LinuxNode, self).do_provision()
+
+
[docs] def do_deploy(self):
+
if self.state == ResourceState.NEW:
+
self.info("Deploying node")
+
self.do_discover()
+
self.do_provision()
+
+
# Node needs to wait until all associated interfaces are
+
# ready before it can finalize deployment
+
from nepi.resources.linux.interface import LinuxInterface
+
ifaces = self.get_connected(LinuxInterface.get_rtype())
+
for iface in ifaces:
+
if iface.state < ResourceState.READY:
+
self.ec.schedule(self.reschedule_delay, self.deploy)
+
return
+
+
super(LinuxNode, self).do_deploy()
+
+
[docs] def do_release(self):
+
rms = self.get_connected()
+
for rm in rms:
+
# Node needs to wait until all associated RMs are released
+
# before it can be released
+
if rm.state != ResourceState.RELEASED:
+
self.ec.schedule(self.reschedule_delay, self.release)
+
return
+
+
tear_down = self.get("tearDown")
+
if tear_down:
+
self.execute(tear_down)
+
+
self.clean_processes()
+
+
super(LinuxNode, self).do_release()
+
+
[docs] def valid_connection(self, guid):
+
# TODO: Validate!
+
return True
+
+
[docs] def clean_processes(self):
+
self.info("Cleaning up processes")
+
+
if self.localhost:
+
return
+
+
if self.get("username") != 'root':
+
cmd = ("sudo -S killall tcpdump || /bin/true ; " +
+
"sudo -S kill -9 $(ps aux | grep '[.]nepi' | awk '{print $2}') || /bin/true ; " +
+
"sudo -S killall -u %s || /bin/true ; " % self.get("username"))
+
else:
+
if self.state >= ResourceState.READY:
+
import pickle
+
pids = pickle.load(open("/tmp/save.proc", "rb"))
+
pids_temp = dict()
+
ps_aux = "ps aux |awk '{print $2,$11}'"
+
(out, err), proc = self.execute(ps_aux)
+
if len(out) != 0:
+
for line in out.strip().split("\n"):
+
parts = line.strip().split(" ")
+
pids_temp[parts[0]] = parts[1]
+
kill_pids = set(pids_temp.items()) - set(pids.items())
+
kill_pids = ' '.join(dict(kill_pids).keys())
+
+
cmd = ("killall tcpdump || /bin/true ; " +
+
"kill $(ps aux | grep '[.]nepi' | awk '{print $2}') || /bin/true ; " +
+
"kill %s || /bin/true ; " % kill_pids)
+
else:
+
cmd = ("killall tcpdump || /bin/true ; " +
+
"kill $(ps aux | grep '[.]nepi' | awk '{print $2}') || /bin/true ; ")
+
else:
+
cmd = ("killall tcpdump || /bin/true ; " +
+
"kill $(ps aux | grep '[.]nepi' | awk '{print $2}') || /bin/true ; ")
+
+
(out, err), proc = self.execute(cmd, retry = 1, with_lock = True)
+
+
[docs] def clean_home(self):
+
""" Cleans all NEPI related folders in the Linux host
+
"""
+
self.info("Cleaning up home")
+
+
cmd = "cd %s ; find . -maxdepth 1 -name \.nepi -execdir rm -rf {} + " % (
+
self.home_dir )
+
+
return self.execute(cmd, with_lock = True)
+
+
[docs] def clean_experiment(self):
+
""" Cleans all experiment related files in the Linux host.
+
It preserves NEPI files and folders that have a multi experiment
+
scope.
+
"""
+
self.info("Cleaning up experiment files")
+
+
cmd = "cd %s ; find . -maxdepth 1 -name '%s' -execdir rm -rf {} + " % (
+
self.exp_dir,
+
self.ec.exp_id )
+
+
return self.execute(cmd, with_lock = True)
+
+
[docs] def execute(self, command,
+
sudo = False,
+
env = None,
+
tty = False,
+
forward_x11 = False,
+
retry = 3,
+
connect_timeout = 30,
+
strict_host_checking = False,
+
persistent = True,
+
blocking = True,
+
with_lock = False
+
):
+
""" Notice that this invocation will block until the
+
execution finishes. If this is not the desired behavior,
+
use 'run' instead."""
+
+
if self.localhost:
+
(out, err), proc = execfuncs.lexec(command,
+
user = self.get("username"), # still problem with localhost
+
sudo = sudo,
+
env = env)
+
else:
+
if with_lock:
+
# If the execute command is blocking, we don't want to keep
+
# the node lock. This lock is used to avoid race conditions
+
# when creating the ControlMaster sockets. A more elegant
+
# solution is needed.
+
with self._node_lock:
+
(out, err), proc = sshfuncs.rexec(
+
command,
+
host = self.get("hostname"),
+
user = self.get("username"),
+
port = self.get("port"),
+
gwuser = self.get("gatewayUser"),
+
gw = self.get("gateway"),
+
agent = True,
+
sudo = sudo,
+
identity = self.get("identity"),
+
server_key = self.get("serverKey"),
+
env = env,
+
tty = tty,
+
forward_x11 = forward_x11,
+
retry = retry,
+
connect_timeout = connect_timeout,
+
persistent = persistent,
+
blocking = blocking,
+
strict_host_checking = strict_host_checking
+
)
+
else:
+
(out, err), proc = sshfuncs.rexec(
+
command,
+
host = self.get("hostname"),
+
user = self.get("username"),
+
port = self.get("port"),
+
gwuser = self.get("gatewayUser"),
+
gw = self.get("gateway"),
+
agent = True,
+
sudo = sudo,
+
identity = self.get("identity"),
+
server_key = self.get("serverKey"),
+
env = env,
+
tty = tty,
+
forward_x11 = forward_x11,
+
retry = retry,
+
connect_timeout = connect_timeout,
+
persistent = persistent,
+
blocking = blocking,
+
strict_host_checking = strict_host_checking
+
)
+
+
return (out, err), proc
+
+
[docs] def run(self, command, home,
+
create_home = False,
+
pidfile = 'pidfile',
+
stdin = None,
+
stdout = 'stdout',
+
stderr = 'stderr',
+
sudo = False,
+
tty = False,
+
strict_host_checking = False):
+
+
self.debug("Running command '%s'" % command)
+
+
if self.localhost:
+
(out, err), proc = execfuncs.lspawn(command, pidfile,
+
home = home,
+
create_home = create_home,
+
stdin = stdin or '/dev/null',
+
stdout = stdout or '/dev/null',
+
stderr = stderr or '/dev/null',
+
sudo = sudo)
+
else:
+
with self._node_lock:
+
(out, err), proc = sshfuncs.rspawn(
+
command,
+
pidfile = pidfile,
+
home = home,
+
create_home = create_home,
+
stdin = stdin or '/dev/null',
+
stdout = stdout or '/dev/null',
+
stderr = stderr or '/dev/null',
+
sudo = sudo,
+
host = self.get("hostname"),
+
user = self.get("username"),
+
port = self.get("port"),
+
gwuser = self.get("gatewayUser"),
+
gw = self.get("gateway"),
+
agent = True,
+
identity = self.get("identity"),
+
server_key = self.get("serverKey"),
+
tty = tty,
+
strict_host_checking = strict_host_checking
+
)
+
+
return (out, err), proc
+
+
[docs] def getpid(self, home, pidfile = "pidfile"):
+
if self.localhost:
+
pidtuple = execfuncs.lgetpid(os.path.join(home, pidfile))
+
else:
+
with self._node_lock:
+
pidtuple = sshfuncs.rgetpid(
+
os.path.join(home, pidfile),
+
host = self.get("hostname"),
+
user = self.get("username"),
+
port = self.get("port"),
+
gwuser = self.get("gatewayUser"),
+
gw = self.get("gateway"),
+
agent = True,
+
identity = self.get("identity"),
+
server_key = self.get("serverKey"),
+
strict_host_checking = False
+
)
+
+
return pidtuple
+
+
[docs] def status(self, pid, ppid):
+
if self.localhost:
+
status = execfuncs.lstatus(pid, ppid)
+
else:
+
with self._node_lock:
+
status = sshfuncs.rstatus(
+
pid, ppid,
+
host = self.get("hostname"),
+
user = self.get("username"),
+
port = self.get("port"),
+
gwuser = self.get("gatewayUser"),
+
gw = self.get("gateway"),
+
agent = True,
+
identity = self.get("identity"),
+
server_key = self.get("serverKey"),
+
strict_host_checking = False
+
)
+
+
return status
+
+
[docs] def kill(self, pid, ppid, sudo = False):
+
out = err = ""
+
proc = None
+
status = self.status(pid, ppid)
+
+
if status == sshfuncs.ProcStatus.RUNNING:
+
if self.localhost:
+
(out, err), proc = execfuncs.lkill(pid, ppid, sudo)
+
else:
+
with self._node_lock:
+
(out, err), proc = sshfuncs.rkill(
+
pid, ppid,
+
host = self.get("hostname"),
+
user = self.get("username"),
+
port = self.get("port"),
+
gwuser = self.get("gatewayUser"),
+
gw = self.get("gateway"),
+
agent = True,
+
sudo = sudo,
+
identity = self.get("identity"),
+
server_key = self.get("serverKey"),
+
strict_host_checking = False
+
)
+
+
return (out, err), proc
+
+
[docs] def copy(self, src, dst):
+
if self.localhost:
+
(out, err), proc = execfuncs.lcopy(src, dst,
+
recursive = True)
+
else:
+
with self._node_lock:
+
(out, err), proc = sshfuncs.rcopy(
+
src, dst,
+
port = self.get("port"),
+
gwuser = self.get("gatewayUser"),
+
gw = self.get("gateway"),
+
identity = self.get("identity"),
+
server_key = self.get("serverKey"),
+
recursive = True,
+
strict_host_checking = False)
+
+
return (out, err), proc
+
+
[docs] def upload(self, src, dst, text = False, overwrite = True,
+
raise_on_error = True):
+
""" Copy content to destination
+
+
src string with the content to copy. Can be:
+
- plain text
+
- a string with the path to a local file
+
- a string with a semi-colon separeted list of local files
+
- a string with a local directory
+
+
dst string with destination path on the remote host (remote is
+
always self.host)
+
+
text src is text input, it must be stored into a temp file before
+
uploading
+
"""
+
# If source is a string input
+
f = None
+
if text and not os.path.isfile(src):
+
# src is text input that should be uploaded as file
+
# create a temporal file with the content to upload
+
f = tempfile.NamedTemporaryFile(delete=False)
+
f.write(src)
+
f.close()
+
src = f.name
+
+
# If dst files should not be overwritten, check that the files do not
+
# exits already
+
if isinstance(src, str):
+
src = map(str.strip, src.split(";"))
+
+
if overwrite == False:
+
src = self.filter_existing_files(src, dst)
+
if not src:
+
return ("", ""), None
+
+
if not self.localhost:
+
# Build destination as <user>@<server>:<path>
+
dst = "%s@%s:%s" % (self.get("username"), self.get("hostname"), dst)
+
+
((out, err), proc) = self.copy(src, dst)
+
+
# clean up temp file
+
if f:
+
os.remove(f.name)
+
+
if err:
+
msg = " Failed to upload files - src: %s dst: %s" % (";".join(src), dst)
+
self.error(msg, out, err)
+
+
msg = "%s out: %s err: %s" % (msg, out, err)
+
if raise_on_error:
+
raise RuntimeError, msg
+
+
return ((out, err), proc)
+
+
[docs] def download(self, src, dst, raise_on_error = True):
+
if not self.localhost:
+
# Build destination as <user>@<server>:<path>
+
src = "%s@%s:%s" % (self.get("username"), self.get("hostname"), src)
+
+
((out, err), proc) = self.copy(src, dst)
+
+
if err:
+
msg = " Failed to download files - src: %s dst: %s" % (";".join(src), dst)
+
self.error(msg, out, err)
+
+
if raise_on_error:
+
raise RuntimeError, msg
+
+
return ((out, err), proc)
+
+
[docs] def install_packages_command(self, packages):
+
command = ""
+
if self.use_rpm:
+
command = rpmfuncs.install_packages_command(self.os, packages)
+
elif self.use_deb:
+
command = debfuncs.install_packages_command(self.os, packages)
+
else:
+
msg = "Error installing packages ( OS not known ) "
+
self.error(msg, self.os)
+
raise RuntimeError, msg
+
+
return command
+
+
[docs] def install_packages(self, packages, home, run_home = None,
+
raise_on_error = True):
+
""" Install packages in the Linux host.
+
+
'home' is the directory to upload the package installation script.
+
'run_home' is the directory from where to execute the script.
+
"""
+
command = self.install_packages_command(packages)
+
+
run_home = run_home or home
+
+
(out, err), proc = self.run_and_wait(command, run_home,
+
shfile = os.path.join(home, "instpkg.sh"),
+
pidfile = "instpkg_pidfile",
+
ecodefile = "instpkg_exitcode",
+
stdout = "instpkg_stdout",
+
stderr = "instpkg_stderr",
+
overwrite = False,
+
raise_on_error = raise_on_error)
+
+
return (out, err), proc
+
+
[docs] def remove_packages(self, packages, home, run_home = None,
+
raise_on_error = True):
+
""" Uninstall packages from the Linux host.
+
+
'home' is the directory to upload the package un-installation script.
+
'run_home' is the directory from where to execute the script.
+
"""
+
if self.use_rpm:
+
command = rpmfuncs.remove_packages_command(self.os, packages)
+
elif self.use_deb:
+
command = debfuncs.remove_packages_command(self.os, packages)
+
else:
+
msg = "Error removing packages ( OS not known ) "
+
self.error(msg)
+
raise RuntimeError, msg
+
+
run_home = run_home or home
+
+
(out, err), proc = self.run_and_wait(command, run_home,
+
shfile = os.path.join(home, "rmpkg.sh"),
+
pidfile = "rmpkg_pidfile",
+
ecodefile = "rmpkg_exitcode",
+
stdout = "rmpkg_stdout",
+
stderr = "rmpkg_stderr",
+
overwrite = False,
+
raise_on_error = raise_on_error)
+
+
return (out, err), proc
+
+
[docs] def mkdir(self, paths, clean = False):
+
""" Paths is either a single remote directory path to create,
+
or a list of directories to create.
+
"""
+
if clean:
+
self.rmdir(paths)
+
+
if isinstance(paths, str):
+
paths = [paths]
+
+
cmd = " ; ".join(map(lambda path: "mkdir -p %s" % path, paths))
+
+
return self.execute(cmd, with_lock = True)
+
+
[docs] def rmdir(self, paths):
+
""" Paths is either a single remote directory path to delete,
+
or a list of directories to delete.
+
"""
+
+
if isinstance(paths, str):
+
paths = [paths]
+
+
cmd = " ; ".join(map(lambda path: "rm -rf %s" % path, paths))
+
+
return self.execute(cmd, with_lock = True)
+
+
[docs] def run_and_wait(self, command, home,
+
shfile="cmd.sh",
+
env=None,
+
overwrite=True,
+
wait_run=True,
+
pidfile="pidfile",
+
ecodefile="exitcode",
+
stdin=None,
+
stdout="stdout",
+
stderr="stderr",
+
sudo=False,
+
tty=False,
+
raise_on_error=True):
+
"""
+
Uploads the 'command' to a bash script in the host.
+
Then runs the script detached in background in the host, and
+
busy-waites until the script finishes executing.
+
"""
+
+
if not shfile.startswith("/"):
+
shfile = os.path.join(home, shfile)
+
+
self.upload_command(command,
+
shfile = shfile,
+
ecodefile = ecodefile,
+
env = env,
+
overwrite = overwrite)
+
+
command = "bash %s" % shfile
+
# run command in background in remote host
+
(out, err), proc = self.run(command, home,
+
pidfile = pidfile,
+
stdin = stdin,
+
stdout = stdout,
+
stderr = stderr,
+
sudo = sudo,
+
tty = tty)
+
+
# check no errors occurred
+
if proc.poll():
+
msg = " Failed to run command '%s' " % command
+
self.error(msg, out, err)
+
if raise_on_error:
+
raise RuntimeError, msg
+
+
# Wait for pid file to be generated
+
pid, ppid = self.wait_pid(
+
home = home,
+
pidfile = pidfile,
+
raise_on_error = raise_on_error)
+
+
if wait_run:
+
# wait until command finishes to execute
+
self.wait_run(pid, ppid)
+
+
(eout, err), proc = self.check_errors(home,
+
ecodefile = ecodefile,
+
stderr = stderr)
+
+
# Out is what was written in the stderr file
+
if err:
+
msg = " Failed to run command '%s' " % command
+
self.error(msg, eout, err)
+
+
if raise_on_error:
+
raise RuntimeError, msg
+
+
(out, oerr), proc = self.check_output(home, stdout)
+
+
return (out, err), proc
+
+
[docs] def exitcode(self, home, ecodefile = "exitcode"):
+
"""
+
Get the exit code of an application.
+
Returns an integer value with the exit code
+
"""
+
(out, err), proc = self.check_output(home, ecodefile)
+
+
# Succeeded to open file, return exit code in the file
+
if proc.wait() == 0:
+
try:
+
return int(out.strip())
+
except:
+
# Error in the content of the file!
+
return ExitCode.CORRUPTFILE
+
+
# No such file or directory
+
if proc.returncode == 1:
+
return ExitCode.FILENOTFOUND
+
+
# Other error from 'cat'
+
return ExitCode.ERROR
+
+
[docs] def upload_command(self, command,
+
shfile="cmd.sh",
+
ecodefile="exitcode",
+
overwrite=True,
+
env=None):
+
""" Saves the command as a bash script file in the remote host, and
+
forces to save the exit code of the command execution to the ecodefile
+
"""
+
+
if not (command.strip().endswith(";") or command.strip().endswith("&")):
+
command += ";"
+
+
# The exit code of the command will be stored in ecodefile
+
command = " { %(command)s } ; echo $? > %(ecodefile)s ;" % {
+
'command': command,
+
'ecodefile': ecodefile,
+
}
+
+
# Export environment
+
environ = self.format_environment(env)
+
+
# Add environ to command
+
command = environ + command
+
+
return self.upload(command, shfile, text=True, overwrite=overwrite)
+
+
+
[docs] def check_errors(self, home,
+
ecodefile = "exitcode",
+
stderr = "stderr"):
+
""" Checks whether errors occurred while running a command.
+
It first checks the exit code for the command, and only if the
+
exit code is an error one it returns the error output.
+
+
"""
+
proc = None
+
err = ""
+
+
# get exit code saved in the 'exitcode' file
+
ecode = self.exitcode(home, ecodefile)
+
+
if ecode in [ ExitCode.CORRUPTFILE, ExitCode.ERROR ]:
+
err = "Error retrieving exit code status from file %s/%s" % (home, ecodefile)
+
elif ecode > 0 or ecode == ExitCode.FILENOTFOUND:
+
# The process returned an error code or didn't exist.
+
# Check standard error.
+
(err, eerr), proc = self.check_output(home, stderr)
+
+
# If the stderr file was not found, assume nothing bad happened,
+
# and just ignore the error.
+
# (cat returns 1 for error "No such file or directory")
+
if ecode == ExitCode.FILENOTFOUND and proc.poll() == 1:
+
err = ""
+
+
return ("", err), proc
+
+
[docs] def wait_pid(self, home, pidfile = "pidfile", raise_on_error = False):
+
""" Waits until the pid file for the command is generated,
+
and returns the pid and ppid of the process """
+
pid = ppid = None
+
delay = 1.0
+
+
for i in xrange(2):
+
pidtuple = self.getpid(home = home, pidfile = pidfile)
+
+
if pidtuple:
+
pid, ppid = pidtuple
+
break
+
else:
+
time.sleep(delay)
+
delay = delay * 1.5
+
else:
+
msg = " Failed to get pid for pidfile %s/%s " % (
+
home, pidfile )
+
self.error(msg)
+
+
if raise_on_error:
+
raise RuntimeError, msg
+
+
return pid, ppid
+
+
[docs] def wait_run(self, pid, ppid, trial = 0):
+
""" wait for a remote process to finish execution """
+
delay = 1.0
+
+
while True:
+
status = self.status(pid, ppid)
+
+
if status is ProcStatus.FINISHED:
+
break
+
elif status is not ProcStatus.RUNNING:
+
delay = delay * 1.5
+
time.sleep(delay)
+
# If it takes more than 20 seconds to start, then
+
# asume something went wrong
+
if delay > 20:
+
break
+
else:
+
# The app is running, just wait...
+
time.sleep(0.5)
+
+
[docs] def check_output(self, home, filename):
+
""" Retrives content of file """
+
(out, err), proc = self.execute("cat %s" %
+
os.path.join(home, filename), retry = 1, with_lock = True)
+
return (out, err), proc
+
+
[docs] def is_alive(self):
+
""" Checks if host is responsive
+
"""
+
if self.localhost:
+
return True
+
+
out = err = ""
+
msg = "Unresponsive host. Wrong answer. "
+
+
# The underlying SSH layer will sometimes return an empty
+
# output (even if the command was executed without errors).
+
# To work arround this, repeat the operation N times or
+
# until the result is not empty string
+
try:
+
(out, err), proc = self.execute("echo 'ALIVE'",
+
blocking = True,
+
with_lock = True)
+
+
if out.find("ALIVE") > -1:
+
return True
+
except:
+
trace = traceback.format_exc()
+
msg = "Unresponsive host. Error reaching host: %s " % trace
+
+
self.error(msg, out, err)
+
return False
+
+
[docs] def find_home(self):
+
""" Retrieves host home directory
+
"""
+
# The underlying SSH layer will sometimes return an empty
+
# output (even if the command was executed without errors).
+
# To work arround this, repeat the operation N times or
+
# until the result is not empty string
+
msg = "Impossible to retrieve HOME directory"
+
try:
+
(out, err), proc = self.execute("echo ${HOME}",
+
blocking = True,
+
with_lock = True)
+
+
if out.strip() != "":
+
self._home_dir = out.strip()
+
except:
+
trace = traceback.format_exc()
+
msg = "Impossible to retrieve HOME directory %s" % trace
+
+
if not self._home_dir:
+
self.error(msg)
+
raise RuntimeError, msg
+
+
[docs] def filter_existing_files(self, src, dst):
+
""" Removes files that already exist in the Linux host from src list
+
"""
+
# construct a dictionary with { dst: src }
+
dests = dict(map(lambda s: (os.path.join(dst, os.path.basename(s)), s), src)) \
+
if len(src) > 1 else dict({dst: src[0]})
+
+
command = []
+
for d in dests.keys():
+
command.append(" [ -f %(dst)s ] && echo '%(dst)s' " % {'dst' : d} )
+
+
command = ";".join(command)
+
+
(out, err), proc = self.execute(command, retry = 1, with_lock = True)
+
+
for d in dests.keys():
+
if out.find(d) > -1:
+
del dests[d]
+
+
if not dests:
+
return []
+
+
return dests.values()
+