client = add_node(ec, client_name, slicename)
# Add resource managers for the linux applications
-app_server = add_app(ec, "cat ${SOURCES}/big_buck_bunny_240p_mpeg4_lq.ts | pv -fbt 2> \
+app_server = add_app(ec, "cat ${SRC}/big_buck_bunny_240p_mpeg4_lq.ts | pv -fbt 2> \
bw.txt | nc %s 1234" % client_name, server, video=video, depends=depends_server)
# Note: is important to add the -d option in nc command to not attempt to read from the
import traceback
err = traceback.format_exc()
self.error(err)
-
- self.set_released()
- self.debug("----- RELEASED ---- ")
+ self.set_released()
+ self.debug("----- RELEASED ---- ")
def fail(self):
""" Sets the RM to state FAILED.
self.set_ready()
def do_release(self):
- pass
+ self.set_released()
+ self.debug("----- RELEASED ---- ")
def do_fail(self):
self.set_failed()
# timestamp of last state check of the application
self._last_state_check = tnow()
-
+
def log_message(self, msg):
return " guid %d - host %s - %s " % (self.guid,
self.node.get("hostname"), msg)
return out
def do_provision(self):
+ # take a snapshot of the system if user is root
+ # to assure cleanProcess kill every nepi process
+ if self.node.get("username") == 'root':
+ ps_aux = "ps aux |awk '{print $2}' |sort -u"
+ (out, err), proc = self.node.execute(ps_aux)
+ self.node._pids = out.split()
+
# create run dir for application
self.node.mkdir(self.run_home)
"releasing the resource",
flags = Flags.ExecReadOnly)
+ gateway_user = Attribute("gatewayUser", "Gateway account username",
+ flags = Flags.ExecReadOnly)
+
+ gateway = Attribute("gateway", "Hostname of the gateway machine",
+ flags = Flags.ExecReadOnly)
+
cls._register_attribute(hostname)
cls._register_attribute(username)
cls._register_attribute(port)
cls._register_attribute(clean_experiment)
cls._register_attribute(clean_processes)
cls._register_attribute(tear_down)
+ cls._register_attribute(gateway_user)
+ cls._register_attribute(gateway)
def __init__(self, ec, guid):
super(LinuxNode, self).__init__(ec, guid)
self._os = None
# home directory at Linux host
self._home_dir = ""
+
+ # list of pids before running the app if the user is root
+ self._pids = []
# lock to prevent concurrent applications on the same node,
# to execute commands at the same time. There are potential
def clean_processes(self):
self.info("Cleaning up processes")
-
- cmd = ("sudo -S killall tcpdump || /bin/true ; " +
- "sudo -S kill $(ps aux | grep '[n]epi' | awk '{print $2}') || /bin/true ; " +
- "sudo -S killall -u %s || /bin/true ; " % self.get("username"))
+
+ if self.get("username") != 'root':
+ cmd = ("sudo -S killall tcpdump || /bin/true ; " +
+ "sudo -S kill $(ps aux | grep '[n]epi' | awk '{print $2}') || /bin/true ; " +
+ "sudo -S killall -u %s || /bin/true ; " % self.get("username"))
+ else:
+ pids_temp = []
+ if self.state >= ResourceState.READY:
+ ps_aux = "ps aux |awk '{print $2}' |sort -u"
+ (out, err), proc = self.execute(ps_aux)
+ pids_temp = out.split()
+ kill_pids = list(set(pids_temp) - set(self._pids))
+ kill_pids = ' '.join(kill_pids)
+
+ cmd = ("killall tcpdump || /bin/true ; " +
+ "kill $(ps aux | grep '[n]epi' | awk '{print $2}') || /bin/true ; " +
+ "kill %s || /bin/true ; " % kill_pids)
+ else:
+ cmd = ("killall tcpdump || /bin/true ; " +
+ "kill $(ps aux | grep '[n]epi' | awk '{print $2}') || /bin/true ; ")
out = err = ""
(out, err), proc = self.execute(cmd, retry = 1, with_lock = True)
host = self.get("hostname"),
user = self.get("username"),
port = self.get("port"),
+ gwuser = self.get("gatewayUser"),
+ gw = self.get("gateway"),
agent = True,
sudo = sudo,
stdin = stdin,
host = self.get("hostname"),
user = self.get("username"),
port = self.get("port"),
+ gwuser = self.get("gatewayUser"),
+ gw = self.get("gateway"),
agent = True,
sudo = sudo,
stdin = stdin,
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"),
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")
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")
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"),
(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,
self._slicenode = False
self._hostname = False
+ if self.get("gateway") or self.get("gatewayUser"):
+ self.set("gateway", None)
+ self.set("gatewayUser", None)
+
def _skip_provision(self):
pl_user = self.get("pluser")
pl_pass = self.get("plpassword")
self.warn(" Could not SSH login ")
self._blacklist_node(node)
#self._delete_node_from_slice(node)
- #self.set('hostname', None)
self.do_discover()
continue
self.warn(" Could not find directory /proc ")
self._blacklist_node(node)
#self._delete_node_from_slice(node)
- #self.set('hostname', None)
self.do_discover()
continue
self._set_hostname_attr(node_id)
self.warn(" Node not responding PING ")
self._blacklist_node(node_id)
- #self.set('hostname', None)
else:
# discovered node for provision, added to provision list
self._put_node_in_provision(node_id)
ip = self._get_ip(node_id)
if not ip: return ping_ok
- command = "ping -c4 %s" % ip
+ command = ['ping', '-c4']
+ command.append(ip)
(out, err) = lexec(command)
if not out.find("2 received") or not out.find("3 received") or not \
command = "su %s ; %s " % (user, command)
- p = subprocess.Popen(command, shell=True,
+ p = subprocess.Popen(command,
stdout = subprocess.PIPE,
- stderr = subprocess.PIPE)
- #stdin = stdin)
+ stderr = subprocess.PIPE,
+ stdin = stdin)
out, err = p.communicate()
return (out, err)
return rv
def rexec(command, host, user,
- port = None,
+ port = None,
+ gwuser = None,
+ gw = None,
agent = True,
sudo = False,
stdin = None,
"""
tmp_known_hosts = None
- hostip = gethostbyname(host)
+ if not gw:
+ hostip = gethostbyname(host)
+ else: hostip = None
args = ['ssh', '-C',
# Don't bother with localhost. Makes test easier
'-o', 'ConnectionAttempts=3',
'-o', 'ServerAliveInterval=30',
'-o', 'TCPKeepAlive=yes',
+ '-o', 'Batchmode=yes',
'-l', user, hostip or host]
if persistent and openssh_has_persist():
# Do not check for Host key. Unsafe.
args.extend(['-o', 'StrictHostKeyChecking=no'])
+ if gw:
+ if gwuser:
+ proxycommand = 'ProxyCommand=ssh %s@%s -W %%h:%%p' % (gwuser, gw)
+ else:
+ proxycommand = 'ProxyCommand=ssh %%r@%s -W %%h:%%p' % gw
+ args.extend(['-o', proxycommand])
+
if agent:
args.append('-A')
stdout = subprocess.PIPE,
stdin = subprocess.PIPE,
stderr = subprocess.PIPE)
-
+
# attach tempfile object to the process, to make sure the file stays
# alive until the process is finished with it
proc._known_hosts = tmp_known_hosts
return ((out, err), proc)
def rcopy(source, dest,
- port = None,
+ port = None,
+ gwuser = None,
+ gw = None,
agent = True,
recursive = False,
identity = None,
user,host = remspec.rsplit('@',1)
tmp_known_hosts = None
- hostip = gethostbyname(host)
+ if not gw:
+ hostip = gethostbyname(host)
+ else: hostip = None
args = ['ssh', '-l', user, '-C',
# Don't bother with localhost. Makes test easier
'-o', 'ControlPath=%s' % (make_control_path(agent, False),),
'-o', 'ControlPersist=60' ])
+ if gw:
+ if gwuser:
+ proxycommand = 'ProxyCommand=ssh %s@%s -W %%h:%%p' % (gwuser, gw)
+ else:
+ proxycommand = 'ProxyCommand=ssh %%r@%s -W %%h:%%p' % gw
+ args.extend(['-o', proxycommand])
+
if port:
args.append('-P%d' % port)
if port:
args.append('-P%d' % port)
+ if gw:
+ if gwuser:
+ proxycommand = 'ProxyCommand=ssh %s@%s -W %%h:%%p' % (gwuser, gw)
+ else:
+ proxycommand = 'ProxyCommand=ssh %%r@%s -W %%h:%%p' % gw
+ args.extend(['-o', proxycommand])
+
if recursive:
args.append('-r')
args.append(source)
args.append(dest)
-
+
for x in xrange(retry):
# connects to the remote host and starts a remote connection
proc = subprocess.Popen(args,
host = None,
port = None,
user = None,
+ gwuser = None,
+ gw = None,
agent = None,
identity = None,
server_key = None,
host = host,
port = port,
user = user,
+ gwuser = gwuser,
+ gw = gw,
agent = agent,
identity = identity,
server_key = server_key,
host = None,
port = None,
user = None,
+ gwuser = None,
+ gw = None,
agent = None,
identity = None,
server_key = None):
host = host,
port = port,
user = user,
+ gwuser = gwuser,
+ gw = gw,
agent = agent,
identity = identity,
server_key = server_key
host = None,
port = None,
user = None,
+ gwuser = None,
+ gw = None,
agent = None,
identity = None,
server_key = None):
host = host,
port = port,
user = user,
+ gwuser = gwuser,
+ gw = gw,
agent = agent,
identity = identity,
server_key = server_key
host = None,
port = None,
user = None,
+ gwuser = None,
+ gw = None,
agent = None,
sudo = False,
identity = None,
host = host,
port = port,
user = user,
+ gwuser = gwuser,
+ gw = gw,
agent = agent,
identity = identity,
server_key = server_key
def test_creation_phase(self):
self.assertEquals(PlanetlabNode._rtype, "PlanetlabNode")
- self.assertEquals(len(PlanetlabNode._attributes), 30)
+ self.assertEquals(len(PlanetlabNode._attributes), 32)
class PLNodeTestCase(unittest.TestCase):
"""
git://git.onelab.eu / nepi.git / commitdiff