#
# 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: Lucia Guevgeozian <lucia.guevgeozian_odizzio@inria.fr>
from nepi.execution.attribute import Attribute, Flags, Types
from nepi.execution.resource import ResourceManager, clsinit_copy, \
ResourceState
from nepi.resources.omf.node import OMFNode
from nepi.util.sfaapi import SFAAPIFactory
from nepi.util.execfuncs import lexec
from nepi.util import sshfuncs
from random import randint
import time
import re
import weakref
import socket
import threading
import datetime
@clsinit_copy
[docs]class WilabtSfaNode(OMFNode):
_rtype = "wilabt::sfa::Node"
_help = "Controls a Wilabt host accessible using a SSH key " \
"and provisioned using SFA"
_platform = "omf"
@classmethod
def _register_attributes(cls):
username = Attribute("username", "Local account username",
flags = Flags.Credential)
identity = Attribute("identity", "SSH identity file",
flags = Flags.Credential)
server_key = Attribute("serverKey", "Server public key",
flags = Flags.Design)
sfa_user = Attribute("sfauser", "SFA user",
flags = Flags.Credential)
sfa_private_key = Attribute("sfaPrivateKey", "SFA path to the private key \
used to generate the user credential",
flags = Flags.Credential)
slicename = Attribute("slicename", "SFA slice for the experiment",
flags = Flags.Credential)
gateway_user = Attribute("gatewayUser", "Gateway account username",
flags = Flags.Design)
gateway = Attribute("gateway", "Hostname of the gateway machine",
flags = Flags.Design)
host = Attribute("host", "Name of the physical machine",
flags = Flags.Design)
disk_image = Attribute("disk_image", "Specify a specific disk image for a node",
flags = Flags.Design)
cls._register_attribute(username)
cls._register_attribute(identity)
cls._register_attribute(server_key)
cls._register_attribute(sfa_user)
cls._register_attribute(sfa_private_key)
cls._register_attribute(slicename)
cls._register_attribute(gateway_user)
cls._register_attribute(gateway)
cls._register_attribute(host)
cls._register_attribute(disk_image)
def __init__(self, ec, guid):
super(WilabtSfaNode, self).__init__(ec, guid)
self._ecobj = weakref.ref(ec)
self._sfaapi = None
self._node_to_provision = None
self._slicenode = False
self._host = False
self._username = None
def _skip_provision(self):
sfa_user = self.get("sfauser")
if not sfa_user:
return True
else: return False
@property
[docs] def sfaapi(self):
"""
Property to instanciate the SFA API based in sfi client.
For each SFA method called this instance is used.
"""
if not self._sfaapi:
sfa_user = self.get("sfauser")
sfa_sm = "http://www.wilab2.ilabt.iminds.be:12369/protogeni/xmlrpc/am/3.0"
sfa_auth = '.'.join(sfa_user.split('.')[:2])
sfa_registry = "http://sfa3.planet-lab.eu:12345/"
sfa_private_key = self.get("sfaPrivateKey")
batch = True
_sfaapi = SFAAPIFactory.get_api(sfa_user, sfa_auth,
sfa_registry, sfa_sm, sfa_private_key, self._ecobj(), batch, WilabtSfaNode._rtype)
if not _sfaapi:
self.fail_sfaapi()
self._sfaapi = weakref.ref(_sfaapi)
return self._sfaapi()
[docs] def do_discover(self):
"""
Based on the attributes defined by the user, discover the suitable
node for provision.
"""
nodes = self.sfaapi.get_resources_hrn()
host = self._get_host()
if host:
# the user specified one particular node to be provisioned
self._host = True
host_hrn = nodes[host]
# check that the node is not blacklisted or being provisioned
# by other RM
if not self._blacklisted(host_hrn):
if not self._reserved(host_hrn):
if self._check_if_in_slice([host_hrn]):
self.debug("Node already in slice %s" % host_hrn)
self._slicenode = True
host = host + '.wilab2.ilabt.iminds.be'
self.set('host', host)
self._node_to_provision = host_hrn
super(WilabtSfaNode, self).do_discover()
[docs] def do_provision(self):
"""
Add node to user's slice and verifing that the node is functioning
correctly. Check ssh, omf rc running, hostname, file system.
"""
provision_ok = False
ssh_ok = False
proc_ok = False
timeout = 300
while not provision_ok:
node = self._node_to_provision
#if self._slicenode:
# self._delete_from_slice()
# self.debug("Waiting 480 sec for re-adding to slice")
# time.sleep(480) # Timout for the testbed to allow a new reservation
self._add_node_to_slice(node)
t = 0
while not self._check_if_in_slice([node]) and t < timeout \
and not self._ecobj().abort:
t = t + 5
time.sleep(t)
self.debug("Waiting 5 sec for resources to be added")
continue
if not self._check_if_in_slice([node]):
self.debug("Couldn't add node %s to slice" % node)
self.fail_node_not_available(node)
self._get_username()
ssh_ok = self._check_ssh_loop()
if not ssh_ok:
# the timeout was reach without establishing ssh connection
# the node is blacklisted, and a new
# node to provision is discovered
self._blacklist_node(node)
self.do_discover()
continue
# check /proc directory is mounted (ssh_ok = True)
# file system is not read only, hostname is correct
# and omf_rc process is up
else:
if not self._check_fs():
self.do_discover()
continue
if not self._check_omfrc():
self.do_discover()
continue
if not self._check_hostname():
self.do_discover()
continue
else:
provision_ok = True
if not self.get('host'):
self._set_host_attr(node)
self.info(" Node provisioned ")
super(WilabtSfaNode, self).do_provision()
[docs] def do_deploy(self):
if self.state == ResourceState.NEW:
self.info("Deploying w-iLab.t node")
self.do_discover()
self.do_provision()
super(WilabtSfaNode, self).do_deploy()
[docs] def do_release(self):
super(WilabtSfaNode, self).do_release()
if self.state == ResourceState.RELEASED and not self._skip_provision():
self.debug(" Releasing SFA API ")
self.sfaapi.release()
def _blacklisted(self, host_hrn):
"""
Check in the SFA API that the node is not in the blacklist.
"""
if self.sfaapi.blacklisted(host_hrn):
self.fail_node_not_available(host_hrn)
return False
def _reserved(self, host_hrn):
"""
Check in the SFA API that the node is not in the reserved
list.
"""
if self.sfaapi.reserved(host_hrn):
self.fail_node_not_available(host_hrn)
return False
def _get_username(self):
"""
Get the username for login in to the nodes from RSpec.
Wilabt username is not made out of any convention, it
has to be retrived from the manifest RSpec.
"""
slicename = self.get("slicename")
if self._username is None:
slice_info = self.sfaapi.get_slice_resources(slicename)
username = slice_info['resource'][0]['services'][0]['login'][0]['username']
self.set('username', username)
self.debug("Retriving username information from RSpec %s" % username)
self._username = username
def _check_ssh_loop(self):
"""
Check that the ssh login is possible. In wilabt is done
through the gateway because is private testbed.
"""
t = 0
timeout = 1200
ssh_ok = False
while t < timeout and not ssh_ok:
cmd = 'echo \'GOOD NODE\''
((out, err), proc) = self.execute(cmd)
if out.find("GOOD NODE") < 0:
self.debug( "No SSH connection, waiting 20s" )
t = t + 20
time.sleep(20)
continue
else:
self.debug( "SSH OK" )
ssh_ok = True
continue
return ssh_ok
def _check_fs(self):
"""
Check file system, /proc well mounted.
"""
cmd = 'mount |grep proc'
((out, err), proc) = self.execute(cmd)
if out.find("/proc type proc") < 0:
self.warning(" Corrupted file system ")
self._blacklist_node(node)
return False
return True
def _check_omfrc(self):
"""
Check that OMF 6 resource controller is running.
"""
cmd = 'ps aux|grep omf'
((out, err), proc) = self.execute(cmd)
if out.find("/usr/local/rvm/gems/ruby-1.9.3-p286@omf/bin/omf_rc") < 0:
return False
return True
def _check_hostname(self):
"""
Check that the hostname in the image is not set to localhost.
"""
cmd = 'hostname'
((out, err), proc) = self.execute(cmd)
if 'localhost' in out.lower():
return False
else:
self.set('hostname', out.strip())
return 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,
):
""" Notice that this invocation will block until the
execution finishes. If this is not the desired behavior,
use 'run' instead."""
(out, err), proc = sshfuncs.rexec(
command,
host = self.get("host"),
user = self.get("username"),
port = 22,
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
def _add_node_to_slice(self, host_hrn):
"""
Add node to slice, using SFA API. Actually Wilabt testbed
doesn't allow adding nodes, in fact in the API there is method
to group all the nodes instanciated as WilabtSfaNodes and the
Allocate and Provision is done with the last call at
sfaapi.add_resource_to_slice_batch.
"""
self.info(" Adding node to slice ")
slicename = self.get("slicename")
disk_image = self.get("disk_image")
if disk_image is not None:
properties = {'disk_image': disk_image}
else: properties = None
#properties = None
self.sfaapi.add_resource_to_slice_batch(slicename, host_hrn, properties=properties)
def _delete_from_slice(self):
"""
Delete every node from slice, using SFA API.
Wilabt doesn't allow to remove one sliver so this method
remove every slice from the slice.
"""
self.warning(" Deleting all slivers from slice ")
slicename = self.get("slicename")
self.sfaapi.remove_all_from_slice(slicename)
def _get_host(self):
"""
Get the attribute hostname.
"""
host = self.get("host")
if host:
return host
else:
return None
def _set_host_attr(self, node):
"""
Query SFAAPI for the hostname of a certain host hrn and sets the
attribute hostname, it will over write the previous value.
"""
hosts_hrn = self.sfaapi.get_resources_hrn()
for host, hrn in hosts_hrn.iteritems():
if hrn == node:
host = host + '.wilab2.ilabt.iminds.be'
self.set("host", host)
def _check_if_in_slice(self, hosts_hrn):
"""
Check using SFA API if any host hrn from hosts_hrn is in the user's
slice.
"""
slicename = self.get("slicename")
slice_nodes = self.sfaapi.get_slice_resources(slicename)['resource']
if slice_nodes:
if len(slice_nodes[0]['services']) != 0:
slice_nodes_hrn = self.sfaapi.get_resources_hrn(slice_nodes).values()
else: slice_nodes_hrn = []
nodes_inslice = list(set(hosts_hrn) & set(slice_nodes_hrn))
return nodes_inslice
def _blacklist_node(self, host_hrn):
"""
Add mal functioning node to blacklist (in SFA API).
"""
self.warning(" Blacklisting malfunctioning node ")
self.sfaapi.blacklist_resource(host_hrn)
if not self._host:
self.set('host', None)
else:
self.set('host', host_hrn.split('.').pop())
def _put_node_in_provision(self, host_hrn):
"""
Add node to the list of nodes being provisioned, in order for other RMs
to not try to provision the same one again.
"""
self.sfaapi.reserve_resource(host_hrn)
def _get_ip(self, host):
"""
Query cache for the IP of a node with certain hostname
"""
try:
ip = sshfuncs.gethostbyname(host)
except:
# Fail while trying to find the IP
return None
return ip
[docs] def fail_discovery(self):
msg = "Discovery failed. No candidates found for node"
self.error(msg)
raise RuntimeError, msg
[docs] def fail_node_not_alive(self, host=None):
msg = "Node %s not alive" % host
raise RuntimeError, msg
[docs] def fail_node_not_available(self, host):
msg = "Some nodes not available for provisioning"
raise RuntimeError, msg
[docs] def fail_not_enough_nodes(self):
msg = "Not enough nodes available for provisioning"
raise RuntimeError, msg
[docs] def fail_sfaapi(self):
msg = "Failing while trying to instanciate the SFA API."
raise RuntimeError, msg
[docs] def valid_connection(self, guid):
# TODO: Validate!
return True