commit f67665e9c1f2840c6077048bbe45ae832f6c525e
Author: Radek Pazdera <rpazdera(a)redhat.com>
Date: Thu Apr 18 19:51:56 2013 +0200
NetTestController: Major Refactoring
This big commit contains major refactoring of the internals of the
NetTestController class.
Multiple changes have been done here including the changes in the XML
format. All these changes are very tightly connected so I am sending
them in a single patch.
Much of the code from the NetTestController related to handling slave
machines has been moved to two new classes Machine and Interface.
Another big change was done to the recipe variable within the
NetTestController class that used to contain a variety of values related
to the recipe execution. Right now, it only contains values that have
been parsed out from the recipe file nothing more. The other things were
moved elsewhere.
There were some changes in the system config handling as well. Most of
the functionality was moved to the slaves, so they can now better
control their own configuration (and clean it up in case the controller
becomes irresponsive).
Signed-off-by: Radek Pazdera <rpazdera(a)redhat.com>
lnst/Common/XmlProcessing.py | 6 +-
lnst/Common/XmlTemplates.py | 36 ++-
lnst/Controller/Machine.py | 520 ++++++++++++++++++++++++++++++++++
lnst/Controller/NetTestController.py | 442 +++++------------------------
lnst/Controller/SlavePool.py | 51 +++-
5 files changed, 654 insertions(+), 401 deletions(-)
---
diff --git a/lnst/Common/XmlProcessing.py b/lnst/Common/XmlProcessing.py
index 83bd630..8e28915 100644
--- a/lnst/Common/XmlProcessing.py
+++ b/lnst/Common/XmlProcessing.py
@@ -237,9 +237,9 @@ class LnstParser(XmlParser):
def set_target(self, data_dict):
self._data = data_dict
- # TODO: This should be removed and done differently after we
- # figure out the new design of template functions
- self._template_proc.set_machines(data_dict["machines"])
+
+ def set_machines(self, machines):
+ self._template_proc.set_machines(machines)
def set_definitions(self, defs):
self._template_proc.set_definitions(defs)
diff --git a/lnst/Common/XmlTemplates.py b/lnst/Common/XmlTemplates.py
index 29d90d9..3664c1e 100644
--- a/lnst/Common/XmlTemplates.py
+++ b/lnst/Common/XmlTemplates.py
@@ -226,19 +226,21 @@ class XmlTemplates:
raise XmlTemplateError(msg)
machine = machines[m_id]
-
- if if_id not in machine['netconfig']:
+ try:
+ iface = machine.get_interface(if_id)
+ except:
msg = "Second parameter of function ip() is invalid: "\
"Interface %s does not exist." % if_id
raise XmlTemplateError(msg)
- if ip_id >= len(machine['netconfig'][if_id]['addresses']) or ip_id < 0:
- msg = "Third parameter of function ip() is invalid: "\
- "Address %s does not exist." % ip_id
- raise XmlTemplateError(msg)
- ip_addr = machine['netconfig'][if_id]['addresses'][ip_id]
+ #try:
+ addr = iface.get_address(ip_id)
+ #except:
+ # msg = "Third parameter of function ip() is invalid: "\
+ # "Address %s does not exist." % ip_id
+ # raise XmlTemplateError(msg)
- return ip_addr.split('/')[0]
+ return addr.split('/')[0]
def _hwaddr_func(self, params):
self._validate_func_params("hwaddr", params, 2, 0)
@@ -252,13 +254,14 @@ class XmlTemplates:
raise XmlTemplateError(msg)
machine = machines[m_id]
- if if_id not in machine['netconfig']:
- msg = "Second parameter of function hwaddr() is invalid: "\
+ try:
+ iface = machine.get_interface(if_id)
+ except:
+ msg = "Second parameter of function ip() is invalid: "\
"Interface %s does not exist." % if_id
raise XmlTemplateError(msg)
- mac_addr = machine['netconfig'][if_id]['hwaddr']
- return mac_addr
+ return iface.get_hwaddr()
def _devname_func(self, params):
@@ -273,13 +276,14 @@ class XmlTemplates:
raise XmlTemplateError(msg)
machine = machines[m_id]
- if if_id not in machine['netconfig']:
- msg = "Second parameter of function devname() is invalid: "\
+ try:
+ iface = machine.get_interface(if_id)
+ except:
+ msg = "Second parameter of function ip() is invalid: "\
"Interface %s does not exist." % if_id
raise XmlTemplateError(msg)
- dev_name = machine['netconfig'][if_id]['name']
- return dev_name
+ return iface.get_devname()
@staticmethod
def _validate_func_params(name, params, mandatory, optional):
diff --git a/lnst/Controller/Machine.py b/lnst/Controller/Machine.py
new file mode 100644
index 0000000..ef716ec
--- /dev/null
+++ b/lnst/Controller/Machine.py
@@ -0,0 +1,520 @@
+"""
+This file containst classes for representing and handling
+a Machine and an Interface in LNST
+
+Copyright 2013 Red Hat, Inc.
+Licensed under the GNU General Public License, version 2 as
+published by the Free Software Foundation; see COPYING for details.
+"""
+
+__author__ = """
+rpazdera(a)redhat.com (Radek Pazdera)
+"""
+
+import logging
+import socket
+import os
+import re
+import pickle
+import tempfile
+from time import sleep
+from xmlrpclib import Binary
+from pprint import pprint, pformat
+from lnst.Common.Logs import log_exc_traceback
+from lnst.Common.XmlRpc import ServerProxy, ServerException
+from lnst.Common.NetUtils import MacPool
+from lnst.Common.VirtUtils import VirtNetCtl, VirtDomainCtl, BridgeCtl
+from lnst.Common.Utils import wait_for, md5sum, dir_md5sum, create_tar_archive
+from lnst.Common.ConnectionHandler import send_data, recv_data
+from lnst.Common.ConnectionHandler import ConnectionHandler
+
+class MachineError(Exception):
+ pass
+
+class Machine(object):
+ """ Slave machine abstraction
+
+ A machine object represents a handle using which the controller can
+ manipulate the machine. This includes tasks such as, configuration,
+ deconfiguration, and running commands.
+ """
+
+ def __init__(self, m_id, hostname=None, libvirt_domain=None):
+ self._id = m_id
+ self._hostname = hostname
+ self._connection = None
+ self._configured = False
+ self._system_config = {}
+
+ self._domain_ctl = None
+ self._libvirt_domain = libvirt_domain
+ if libvirt_domain:
+ self._domain_ctl = VirtDomainCtl(libvirt_domain)
+
+ self._msg_dispatcher = None
+ self._mac_pool = None
+
+ self._interfaces = []
+
+ def _add_interface(self, if_id, if_type, cls):
+ if if_id != None:
+ for iface in self._interfaces:
+ if if_id == iface.get_id():
+ msg = "Interface '%s' already exists on machine '%s'" \
+ % (if_id, self._id)
+ raise MachineError(msg)
+
+ iface = cls(self, if_id, if_type)
+ self._interfaces.append(iface)
+ return iface
+
+ #
+ # Factory methods for constructing interfaces on this machine. The
+ # types of interfaces are explained with the classes below.
+ #
+ def new_static_interface(self, if_id, if_type):
+ return self._add_interface(if_id, if_type, StaticInterface)
+
+ def new_unused_interface(self, if_type):
+ return self._add_interface(if_type, None, UnusedInterface)
+
+ def new_virtual_interface(self, if_id, if_type):
+ return self._add_interface(if_id, if_type, VirtualInterface)
+
+ def new_soft_interface(self, if_id, if_type):
+ return self._add_interface(if_id, if_type, SoftInterface)
+
+ def get_interface(self, if_id):
+ for iface in self._interfaces:
+ if iface.get_id != None and if_id == iface.get_id():
+ return iface
+
+ msg = "Interface '%s' not found on machine '%s'" % (if_id, self._id)
+ raise MachineError(msg)
+
+ def _rpc_call(self, method_name, *args):
+ data = {"type": "command", "method_name": method_name, "args": args}
+
+ self._msg_dispatcher.send_message(self._id, data)
+ result = self._msg_dispatcher.wait_for_result(self._id)
+
+ return result
+
+ def configure(self, recipe_name, do_cleanup=False):
+ """ Prepare the machine
+
+ Calling this method will initialize the rpc connection to the
+ machine and initialize all the interfaces. Note, that it will
+ *not* configure the interfaces. They need to be configured
+ individually later on.
+ """
+ hostname = self._hostname
+ port = self._port
+ m_id = self._id
+
+ logging.info("Connecting to RPC on machine %s (%s)", m_id, hostname)
+ connection = socket.create_connection((hostname, port))
+ self._msg_dispatcher.add_slave(self._id, connection)
+
+ hello = self._rpc_call("hello", recipe_name)
+ if hello != "hello":
+ msg = "Unable to establish RPC connection " \
+ "to machine %s, handshake failed!" % hostname
+ raise Machine(msg)
+
+ if do_cleanup:
+ self._rpc_call("machine_cleanup")
+
+ for iface in self._interfaces:
+ iface.initialize()
+
+ self._configured = True
+
+ def is_configured(self):
+ """ Test if the machine was configured """
+
+ return self._configured
+
+ def cleanup(self, deconfigure=True):
+ """ Clean the machine up
+
+ This is the counterpart of the configure() method. It will
+ stop any still active commands on the machine, deconfigure
+ all the interfaces that have been configured on the machine,
+ and finalize and close the rpc connection to the machine.
+ """
+ if not self._configured:
+ return
+
+ self._rpc_call("kill_cmds")
+
+ for iface in reversed(self._interfaces):
+ iface.deconfigure()
+ iface.cleanup()
+
+ self._rpc_call("bye")
+ self._msg_dispatcher.disconnect_slave(self.get_id())
+
+ self._configured = False
+
+ def run_command(self, command):
+ """ Run a command on the machine """
+
+ if "timeout" in command:
+ timeout = command["timeout"]
+ logging.debug("Setting socket timeout to \"%d\"", timeout)
+ socket.setdefaulttimeout(timeout)
+ try:
+ cmd_res = self._rpc_call("run_command", command)
+ except socket.timeout:
+ msg = "RPC connection to machine %s timed out" % self.get_id()
+ raise Machine(msg)
+ finally:
+ if "timeout" in command:
+ logging.debug("Setting socket timeout to default value")
+ socket.setdefaulttimeout(None)
+
+ return cmd_res
+
+ def get_hostname(self):
+ """ Get hostname/ip of the machine
+
+ This will return the hostname/ip of the machine's controller
+ interface.
+ """
+ return self._hostname
+
+ def get_id(self):
+ """ Returns machine's id as defined in the recipe """
+ return self._id
+
+ def set_rpc(self, dispatcher, port):
+ self._msg_dispatcher = dispatcher
+ self._port = port
+
+ def get_mac_pool(self):
+ if self._mac_pool:
+ return self._mac_pool
+ else:
+ raise MachineError("Mac pool not available.")
+
+ def set_mac_pool(self, mac_pool):
+ self._mac_pool = mac_pool
+
+ def restore_system_config(self):
+ return self._rpc_call("restore_system_config")
+
+ def set_network_bridges(self, bridges):
+ self._network_bridges = bridges
+
+ def get_network_bridges(self, bridges):
+ if self._network_bridges:
+ return self._network_bridges
+ else:
+ raise MachineError("Network bridges not available.")
+
+ def start_packet_capture(self):
+ return self._rpc_call("start_packet_capture", "")
+
+ def stop_packet_capture(self):
+ self._rpc_call("stop_packet_capture")
+
+ def copy_file_to_machine(self, local_path, remote_path=None):
+ remote_path = self._rpc_call("start_copy_to", remote_path)
+ f = open(local_path, "rb")
+
+ while True:
+ data = f.read(1024*1024) # 1MB buffer
+ if len(data) == 0:
+ break
+
+ self._rpc_call("copy_part_to", remote_path, Binary(data))
+
+ self._rpc_call("finish_copy_to", remote_path)
+ return remote_path
+
+ def copy_file_from_machine(self, remote_path, local_path):
+ status = self._rpc_call("start_copy_from", remote_path)
+ if not status:
+ raise MachineError("The requested file cannot be transfered." \
+ "It does not exist on machine %s" % self.get_id())
+
+ local_file = open(local_path, "wb")
+
+ buf_size = 1024*1024 # 1MB buffer
+ binary = "next"
+ while binary != "":
+ binary = self._rpc_call("copy_part_from", remote_path, buf_size)
+ local_file.write(binary.data)
+
+ local_file.close()
+ self._rpc_call("finish_copy_from", remote_path)
+
+ def sync_resources(self, required):
+ self._rpc_call("clear_resource_table")
+
+ for res_type, resources in required.iteritems():
+ for res_name, res in resources.iteritems():
+ has_resource = self._rpc_call("has_resource", res["hash"])
+ if not has_resource:
+ msg = "Transfering %s %s to machine %s" % \
+ (res_name, res_type, self.get_id())
+ logging.info(msg)
+
+ local_path = required[res_type][res_name]["path"]
+
+ if res_type == "tools":
+ archive = tempfile.NamedTemporaryFile(delete=False)
+ archive_path = archive.name
+ archive.close()
+
+ create_tar_archive(local_path, archive_path, True)
+ local_path = archive_path
+
+ remote_path = self.copy_file_to_machine(local_path)
+ self._rpc_call("add_resource_to_cache", res["hash"],
+ remote_path, res_name, res["path"], res_type)
+
+ if res_type == "tools":
+ os.unlink(archive_path)
+
+ self._rpc_call("map_resource", res["hash"], res_type, res_name)
+
+ def __str__(self):
+ return "[Machine hostname(%s) libvirt_domain(%s) interfaces(%d)]" % \
+ (self._hostname, self._libvirt_domain, len(self._interfaces))
+
+class Interface(object):
+ """ Abstraction of a test network interface on a slave machine
+
+ This is a base class for object that represent test interfaces
+ on a test machine.
+ """
+ def __init__(self, machine, if_id, if_type):
+ self._machine = machine
+ self._configured = False
+
+ self._id = if_id
+ self._type = if_type
+
+ self._hwaddr = None
+ self._devname = None
+ self._network = None
+
+ self._slaves = {}
+ self._addresses = []
+ self._options = []
+
+ def get_id(self):
+ return self._id
+
+ def set_hwaddr(self, hwaddr):
+ self._hwaddr = hwaddr
+
+ def get_hwaddr(self):
+ if not self._hwaddr:
+ msg = "Hardware address is not available for interface ''" \
+ % self.get_id()
+ raise MachineError(msg)
+ return self._hwaddr
+
+ def set_devname(self, devname):
+ self._devname = devname
+
+ def get_devname(self):
+ if not self._devname:
+ msg = "Device name is not available for interface ''" \
+ % self.get_id()
+ raise MachineError(msg)
+ return self._devname
+
+ def set_network(self, network):
+ self._network = network
+
+ def get_network(self):
+ if not self._network:
+ msg = "Network segment is not available for interface ''" \
+ % self.get_id()
+ raise MachineError(msg)
+ return self._network
+
+ def set_option(self, name, value):
+ self._options.append((name, value))
+
+ def add_slave(self, iface):
+ self._slaves[iface.get_id()] = iface
+
+ def add_address(self, addr):
+ self._addresses.append(addr)
+
+ def get_address(self, num):
+ return self._addresses[num]
+
+ def _get_config(self):
+ config = {"hwaddr": self._hwaddr, "type": self._type,
+ "addresses": self._addresses, "slaves": self._slaves.keys(),
+ "options": self._options}
+ if self._type == "eth":
+ config["phys_id"] = self.get_id()
+ return config
+
+ def down(self):
+ self._machine._rpc_call("set_device_down", self._hwaddr)
+
+ def initialize(self):
+ phys_devs = self._machine._rpc_call("get_devices_by_hwaddr",
+ self._hwaddr)
+ if len(phys_devs) == 1:
+ pass
+ elif len(phys_devs) < 1:
+ msg = "Device %s not found on machine %s" \
+ % (self.get_id(), self._machine.get_id())
+ raise MachineError(msg)
+ elif len(phys_devs) > 1:
+ msg = "Multiple interfaces with same address %s on machine %s" \
+ % (self._hwaddr, self._machine.get_id())
+ raise MachineError(msg)
+
+ self.down()
+
+ def cleanup(self):
+ pass
+
+ def configure(self):
+ if self._configured:
+ msg = "Unable to configure interface %s on machine %s. " \
+ "It has been configured already." % (self.get_id(),
+ self._machine.get_id())
+ raise MachineError(msg)
+
+ logging.info("Configuring interface %s on machine %s", self.get_id(),
+ self._machine.get_id())
+
+ self._machine._rpc_call("configure_interface", self.get_id(),
+ self._get_config())
+ self._configured = True
+
+ if_info = self._machine._rpc_call("get_interface_info", self.get_id())
+ if "name" in if_info:
+ self._devname = if_info["name"]
+ self._hwaddr = if_info["hwaddr"]
+
+ def deconfigure(self):
+ if not self._configured:
+ return
+
+ self._machine._rpc_call("deconfigure_interface", self.get_id())
+ self._configured = False
+
+class StaticInterface(Interface):
+ """ Static interface
+
+ This class represents interfaces that are present on the
+ machine. LNST will only use them for testing without performing
+ any special actions.
+
+ This type is suitable for physical interfaces.
+ """
+ def __init__(self, machine, if_id, if_type):
+ super(StaticInterface, self).__init__(machine, if_id, if_type)
+
+class VirtualInterface(Interface):
+ """ Dynamically created interface
+
+ This class represents interfaces in libvirt virtual machines
+ that were created dynamically by LNST just for this test.
+
+ This requires some special handling and communication with
+ libvirt.
+ """
+ def __init__(self, machine, if_id, if_type):
+ super(VirtualInterface, self).__init__(machine, if_id, if_type)
+
+ def initialize(self):
+ if not self._domain_ctl:
+ msg = "Cannot create an interface. " \
+ "Machine '%s' is not virtual." % machine_id
+ raise MachineError(msg)
+
+ if self._hwaddr:
+ query = self._machine._rpc_call('get_devices_by_hwaddr',
+ self._hwaddr)
+ if len(query):
+ msg = "Device with hwaddr %s already exists" % self._hwaddr
+ raise MachineError(msg)
+ else:
+ while True:
+ self._hwaddr = self._machine.get_mac_pool().get_addr()
+ query = self._machine._rpc_call('get_devices_by_hwaddr',
+ self._hwaddr)
+ if not len(query_result):
+ break
+
+ bridges = self._machine.get_network_bridges()
+ if self._network in bridges:
+ brctl = bridges[network]
+ else:
+ bridges["network"] = brctl = BridgeCtl()
+
+ br_name = brctl.get_name()
+ brctl.init()
+
+ logging.info("Creating interface %s (%s) on machine %s",
+ self.get_id(), self._hwaddr, self._machine.get_id())
+
+ domain_ctl = self._machine.get_domain_ctl()
+ domain_ctl.attach_interface(self._hwaddr, br_name)
+
+ ready = wait_for(self._ready, timeout=10)
+ if not ready:
+ msg = "Netdevice initialization failed." \
+ "Unable to create device %s (%s) on machine %s" \
+ % (self._get_id, self._hwaddr, self._machine.get_id())
+ raise MachineError(msg)
+
+ super(VirtualInterface, self).initialize()
+
+ def cleanup(self):
+ domain_ctl = self._machine.get_domain_ctl()
+ domain_ctl.detach_interface(self._hwaddr)
+
+ def is_ready(self):
+ ifaces = self._rpc_call('get_devices_by_hwaddr', self._hwaddr)
+ return len(ifaces) > 0
+
+class SoftInterface(Interface):
+ """ Software interface abstraction
+
+ This type of interface represents interfaces created in the kernel
+ during the runtime. This includes devices such as bonds and teams.
+ """
+
+ def __init__(self, machine, if_id, if_type):
+ super(SoftInterface, self).__init__(machine, if_id, if_type)
+
+ def initialize(self):
+ pass
+
+class UnusedInterface(Interface):
+ """ Unused interface for this test
+
+ This class represents interfaces that will not be used in the
+ current test setup. This applies when a slave machine from a
+ pool has more interfaces then the machine it was matched to
+ from the recipe.
+
+ LNST still needs to know about these interfaces so it can turn
+ them off.
+ """
+
+ def __init__(self, machine, if_id, if_type):
+ super(UnusedInterface, self).__init__(machine, if_id, if_type)
+
+ def initialize(self):
+ self.down()
+
+ def configure(self):
+ pass
+
+ def deconfigure(self):
+ pass
diff --git a/lnst/Controller/NetTestController.py b/lnst/Controller/NetTestController.py
index d51e66f..bf80fbe 100644
--- a/lnst/Controller/NetTestController.py
+++ b/lnst/Controller/NetTestController.py
@@ -30,6 +30,7 @@ from lnst.Common.NetTestCommand import NetTestCommandContext, NetTestCommand
from lnst.Common.NetTestCommand import str_command, CommandException
from lnst.Controller.RecipeParse import RecipeParse
from lnst.Controller.SlavePool import SlavePool
+from lnst.Controller.Machine import Machine, MachineError
from lnst.Common.ConnectionHandler import send_data, recv_data
from lnst.Common.ConnectionHandler import ConnectionHandler
@@ -54,24 +55,24 @@ class NetTestController:
check_process_running("libvirtd"), config)
self._slave_pool = sp
+ self._machines = {}
+ self._network_bridges = {}
+
self._recipe = recipe = {}
recipe["networks"] = {}
recipe["machines"] = {}
- recipe["provisioning"] = {}
recipe["switches"] = {}
mac_pool_range = config.get_option('environment', 'mac_pool_range')
- self._mac_pool = MacPool(mac_pool_range[0],
- mac_pool_range[1])
+ self._mac_pool = MacPool(mac_pool_range[0], mac_pool_range[1])
- ntparse = RecipeParse(recipe_path)
- ntparse.set_target(self._recipe)
+ parser = RecipeParse(recipe_path)
+ parser.set_target(self._recipe)
+ parser.set_machines(self._machines)
- ntparse.register_event_handler("provisioning_requirements_ready",
+ parser.register_event_handler("provisioning_requirements_ready",
self._prepare_provisioning)
- ntparse.register_event_handler("machine_ready",
- self._prepare_slave)
- ntparse.register_event_handler("interface_config_ready",
+ parser.register_event_handler("interface_config_ready",
self._prepare_interface)
modules_dirs = config.get_option('environment', 'module_dirs')
@@ -81,7 +82,7 @@ class NetTestController:
self._resource_table["module"] = self._load_test_modules(modules_dirs)
self._resource_table["tools"] = self._load_test_tools(tools_dirs)
- self._ntparse = ntparse
+ self._parser = parser
def _get_machineinfo(self, machine_id):
try:
@@ -99,317 +100,104 @@ class NetTestController:
raise NetTestError(msg)
def _prepare_provisioning(self):
- provisioning = self._recipe["provisioning"]
- if len(provisioning["setup_requirements"]) <= 0:
+ machines = self._recipe["machines"]
+ if len(machines) <= 0:
return
sp = self._slave_pool
- machines = sp.provision_setup(provisioning["setup_requirements"])
+ machines = sp.provision_machines(machines)
if machines == None:
msg = "This setup cannot be provisioned with the current pool."
raise NetTestError(msg)
for m_id, machine in machines.iteritems():
- self._recipe["machines"][m_id] = machine
- provisioning["map"] = {}
+ self._machines[m_id] = machine
logging.info("Provisioning initialized")
for m_id in machines.keys():
provisioner = sp.get_provisioner_id(m_id)
- provisioning["map"][m_id] = provisioner
logging.info(" machine %s uses %s" % (m_id, provisioner))
- machines[m_id]["params"]["system_config"] = {}
-
- def _prepare_device(self, machine_id, dev_id):
- info = self._get_machineinfo(machine_id)
- dev = self._recipe["machines"][machine_id]["interfaces"][dev_id]
-
- dev_net_name = dev["network"]
- networks = self._recipe["networks"]
- if not dev_net_name in networks:
- networks[dev_net_name] = {"members": []}
-
- dev_net = networks[dev_net_name]
- dev_net["members"].append((machine_id, dev_id))
-
- if dev["create"] == "libvirt":
- if not "virt_domain_ctl" in info:
- msg = "Cannot create device. " \
- "Machine '%s' is not virtual." % (machine_id)
- raise NetTestError(msg)
-
- if "hwaddr" in dev:
- query_result = self._rpc_call(machine_id,
- 'get_devices_by_hwaddr', dev["hwaddr"])
- if query_result:
- msg = "Device with hwaddr %s already exists" \
- % dev["hwaddr"]
- raise NetTestError(msg)
- else:
- while True:
- dev["hwaddr"] = self._mac_pool.get_addr()
- query_result = self._rpc_call(machine_id,
- 'get_devices_by_hwaddr', dev["hwaddr"])
- if not len(query_result):
- break
-
- if "libvirt_bridge" in dev:
- brctl = BridgeCtl(dev["libvirt_bridge"])
- else:
- if "default_bridge" in dev_net:
- brctl = dev_net["default_bridge"]
- else:
- brctl = BridgeCtl()
- dev_net["default_bridge"] = brctl
-
- br_name = brctl.get_name()
- brctl.init()
-
- logging.info("Creating interface %s (%s) on machine %s",
- dev_id, dev["hwaddr"], machine_id)
-
- domain_ctl = info["virt_domain_ctl"]
- domain_ctl.attach_interface(dev["hwaddr"], br_name)
-
- ready_check_func = lambda: self._device_ready(machine_id, dev_id)
- ready = wait_for(ready_check_func, timeout=10)
-
- if not ready:
- msg = "Netdevice initialization failed." \
- "Unable to create device %s (%s) on machine %s" \
- % (dev_id, dev["hwaddr"], machine_id)
- raise NetTestError(msg)
-
- if 'created_devices' not in info:
- info['created_devices'] = []
- info['created_devices'].append((dev_id, dev))
-
- phys_devs = self._rpc_call(machine_id,
- 'get_devices_by_hwaddr', dev["hwaddr"])
- if len(phys_devs) == 1:
- pass
- elif len(phys_devs) < 1:
- msg = "Device %s not found on machine %s" \
- % (dev_id, machine_id)
- raise NetTestError(msg)
- elif len(phys_devs) > 1:
- msg = "Multiple interfaces with same address %s on machine %s" \
- % (dev["hwaddr"], machine_id)
- raise NetTestError(msg)
-
- def _device_ready(self, machine_id, dev_id):
- dev = self._recipe["machines"][machine_id]["interfaces"][dev_id]
-
- devs = self._rpc_call(machine_id,
- 'get_devices_by_hwaddr', dev["hwaddr"])
- return len(devs) > 0
-
- def _prepare_interface(self, machine_id, netdev_config_id):
- info = self._get_machineinfo(machine_id)
- logging.info("Configuring interface %s on %s", netdev_config_id,
- info["hostname"])
-
- self._configure_interface(machine_id, netdev_config_id)
-
- if_info = self._rpc_call(machine_id,
- 'get_interface_info', netdev_config_id)
- machine = self._recipe["machines"][machine_id]
- if "name" in if_info:
- machine["netconfig"][netdev_config_id]["name"] = if_info["name"]
-
- info["configured_interfaces"].append(netdev_config_id)
-
- def _configure_interface(self, machine_id, netdev_config_id):
- netconfig = self._recipe["machines"][machine_id]["netconfig"]
- dev_config = netconfig[netdev_config_id]
-
- self._rpc_call(machine_id,
- 'configure_interface', netdev_config_id, dev_config)
-
- def _deconfigure_interface(self, machine_id, netdev_config_id):
- self._rpc_call(machine_id, 'deconfigure_interface', netdev_config_id)
-
- def _prepare_slave(self, machine_id):
- logging.info("Preparing machine %s", machine_id)
- info = self._get_machineinfo(machine_id)
+ for m_id in machines.keys():
+ self._prepare_machine(m_id)
- if "libvirt_domain" in info:
- domain_ctl = VirtDomainCtl(info["libvirt_domain"])
- info["virt_domain_ctl"] = domain_ctl
+ def _prepare_machine(self, m_id):
+ machine = self._machines[m_id]
+ address = socket.gethostbyname(machine.get_hostname())
- self._init_slave_logging(machine_id)
- self._init_slave_rpc(machine_id)
+ self._log_ctl.add_slave(m_id, address)
+ port = self._config.get_option('environment', 'rpcport')
+ machine.set_rpc(self._msg_dispatcher, port)
+ machine.set_mac_pool(self._mac_pool)
- info["configured_interfaces"] = []
+ recipe_name = os.path.basename(self._recipe_path)
+ machine.configure(recipe_name, self._docleanup)
+ machine.sync_resources(self._resource_table)
- self._rpc_call(machine_id, "clear_resource_table")
- required = self._resource_table
+ def _prepare_interface(self, machine_id, if_id):
+ machine = self._machines[machine_id]
+ ifconfig = self._recipe["machines"][machine_id]["interfaces"][if_id]
+ if_type = ifconfig["type"]
- if self._docleanup and not info["skip_cleanup"]:
- self._rpc_call(machine_id, 'machine_cleanup')
- else:
- logging.info("Skipping cleanup on machine %s" % machine_id)
-
- for res_type, resources in self._resource_table.iteritems():
- for res_name, res in resources.iteritems():
- has_resource = self._rpc_call(machine_id, "has_resource",
- res["hash"])
- if not has_resource:
- msg = "Transfering %s %s to machine %s" % \
- (res_name, res_type, machine_id)
- logging.info(msg)
-
- local_path = required[res_type][res_name]["path"]
-
- if res_type == "tools":
- archive = tempfile.NamedTemporaryFile(delete=False)
- archive_path = archive.name
- archive.close()
-
- create_tar_archive(local_path, archive_path, True)
- local_path = archive_path
-
- remote_path = self._copy_to_slave(local_path, machine_id)
- self._rpc_call(machine_id, "add_resource_to_cache",
- res["hash"], remote_path, res_name,
- res["path"], res_type)
-
- if res_type == "tools":
- os.unlink(archive_path)
-
- self._rpc_call(machine_id, "map_resource",
- res["hash"], res_type, res_name)
-
- # Some additional initialization is necessary in case the
- # underlying machine is provisioned from the pool
- prov_id = self._slave_pool.get_provisioner_id(machine_id)
- if prov_id:
- provisioner = self._slave_pool.get_provisioner(machine_id)
- logging.info("Initializing provisioned system (%s)" % prov_id)
- for device in provisioner["interfaces"].itervalues():
- self._rpc_call(machine_id, 'set_device_down', device["hwaddr"])
-
- machine = self._recipe["machines"][machine_id]
- for dev_id in machine["interfaces"].iterkeys():
- self._prepare_device(machine_id, dev_id)
-
- def _init_slave_rpc(self, machine_id):
- info = self._get_machineinfo(machine_id)
- hostname = info["hostname"]
- if "rpcport" in info:
- port = info["rpcport"]
- else:
- port = self._config.get_option('environment', 'rpcport')
- logging.info("Connecting to RPC on machine %s", hostname)
+ try:
+ iface = machine.get_interface(if_id)
+ except MachineError:
+ iface = machine.new_soft_interface(if_id, if_type)
- rpc = socket.create_connection((hostname, port))
- self._msg_dispatcher.add_slave(machine_id, rpc, info)
+ if "slaves" in ifconfig:
+ for slave_id in ifconfig["slaves"]:
+ iface.add_slave(machine.get_interface(slave_id))
- if self._rpc_call(machine_id, 'hello', self._recipe_path) != "hello":
- msg = "Unable to establish RPC connection to machine %s. " \
- % hostname
- msg += "Handshake failed"
- raise NetTestError(msg)
+ if "addresses" in ifconfig:
+ for addr in ifconfig["addresses"]:
+ iface.add_address(addr)
- def _init_slave_logging(self, machine_id):
- info = self._get_machineinfo(machine_id)
- address = socket.gethostbyname(info["hostname"])
+ if "options" in ifconfig:
+ for name, value in ifconfig["options"]:
+ iface.set_option(name, value)
- info['logger'] = self._log_ctl.add_slave(address)
+ iface.configure()
- def _deconfigure_slaves(self):
- if 'machines' not in self._recipe:
+ def _cleanup_slaves(self, deconfigure=True):
+ if self._machines == None:
return
- for machine_id in self._recipe["machines"]:
- info = self._get_machineinfo(machine_id)
- if self._msg_dispatcher.get_connection(machine_id):
- self._rpc_call(machine_id, "kill_cmds")
- else:
- continue
+ for machine_id, machine in self._machines.iteritems():
+ if machine.is_configured():
+ machine.cleanup()
- if "configured_interfaces" not in info:
- continue
-
- for if_id in reversed(info["configured_interfaces"]):
- self._rpc_call(machine_id, 'deconfigure_interface', if_id)
-
- # detach dynamically created devices
- if "created_devices" not in info:
- continue
- for dev_id, dev in reversed(info["created_devices"]):
- logging.info("Removing interface %s (%s) from machine %s",
- dev_id, dev["hwaddr"], machine_id)
- domain_ctl = info["virt_domain_ctl"]
- domain_ctl.detach_interface(dev["hwaddr"])
-
- #clean-up slave logger
- self._log_ctl.remove_slave(machine_id)
+ #clean-up slave logger
+ self._log_ctl.remove_slave(machine_id)
# remove dynamically created bridges
- networks = self._recipe["networks"]
- for net in networks.itervalues():
- if "default_bridge" in net:
- net["default_bridge"].cleanup()
-
- def _disconnect_slaves(self):
- if 'machines' not in self._recipe:
- return
-
- for machine_id in self._recipe["machines"]:
- if self._msg_dispatcher.get_connection(machine_id):
- self._rpc_call(machine_id, "bye")
- self._msg_dispatcher.disconnect_slave(machine_id)
+ for bridge in self._network_bridges:
+ bridge.cleanup()
def _prepare(self):
# All the perparations are made within the recipe parsing
# This is achieved by handling parser events
try:
- self._ntparse.parse_recipe()
+ self._parser.parse_recipe()
except Exception as exc:
logging.debug("Exception raised during recipe parsing. "\
"Deconfiguring machines.")
log_exc_traceback()
- self._deconfigure_slaves()
- self._disconnect_slaves()
+ self._cleanup_slaves()
raise NetTestError(exc)
def _run_command(self, command):
- machine_id = command["machine_id"]
- try:
- desc = command["desc"]
+ if "desc" in command:
logging.info("Cmd description: %s", desc)
- except KeyError:
- pass
if command["type"] == "ctl_wait":
sleep(command["value"])
cmd_res = {"passed" : True}
return cmd_res
- if "timeout" in command:
- timeout = command["timeout"]
- logging.debug("Setting socket timeout to \"%d\"", timeout)
- socket.setdefaulttimeout(timeout)
- try:
- cmd_res = self._rpc_call(machine_id, 'run_command', command)
- except socket.timeout:
- msg = "RPC connection to machine %s timed out" % machine_id
- raise NetTestError(msg)
- if "timeout" in command:
- logging.debug("Setting socket timeout to default value")
- socket.setdefaulttimeout(None)
-
- if command["type"] == "system_config":
- if cmd_res["passed"]:
- self._update_system_config(machine_id, cmd_res["res_data"],
- command["persistent"])
- else:
- err = "Error occured while setting system configuration (%s)" \
- % cmd_res["err_msg"]
- logging.error(err)
+ machine_id = command["machine_id"]
+ machine = self._machines[machine_id]
+ cmd_res = machine.run_command(command)
return cmd_res
def _run_command_sequence(self, sequence):
@@ -432,13 +220,12 @@ class NetTestController:
def dump_recipe(self):
self._prepare()
pprint(self._recipe)
- self._deconfigure_slaves()
- self._disconnect_slaves()
+ self._cleanup_slaves()
return True
def config_only_recipe(self):
self._prepare()
- self._disconnect_slaves()
+ self._cleanup_slaves(deconfigure=False)
return True
def run_recipe(self, packet_capture=False):
@@ -461,8 +248,7 @@ class NetTestController:
self._stop_packet_capture()
self._gather_capture_files()
- self._deconfigure_slaves()
- self._disconnect_slaves()
+ self._cleanup_slaves()
if not err:
return res
@@ -480,8 +266,8 @@ class NetTestController:
overall_res = False
break
- for machine_id in self._recipe["machines"]:
- self._restore_system_config(machine_id)
+ for machine in self._machines.itervalues():
+ machine.restore_system_config()
# sequence failed, check if we should quit_on_fail
if not res:
@@ -493,22 +279,22 @@ class NetTestController:
def _start_packet_capture(self):
logging.info("Starting packet capture")
- for machine_id in self._recipe["machines"]:
- capture_files = self._rpc_call(machine_id,
- 'start_packet_capture', "")
+ for machine_id, machine in self._machines.iteritems():
+ capture_files = machine.start_packet_capture()
self._remote_capture_files[machine_id] = capture_files
def _stop_packet_capture(self):
logging.info("Stopping packet capture")
- for machine_id in self._recipe["machines"]:
- self._rpc_call(machine_id, 'stop_packet_capture')
+ for machine_id, machine in self._machines.iteritems():
+ machine.stop_packet_capture()
+ # TODO: Move this function to logging
def _gather_capture_files(self):
- logging_root = self._log_root_path
+ logging_root = self._log_ctl.get_recipe_log_path()
logging_root = os.path.abspath(logging_root)
logging.info("Retrieving capture files from slaves")
- for machine_id in self._recipe["machines"]:
- hostname = self._recipe["machines"][machine_id]['info']['hostname']
+ for machine_id, machine in self._machines.iteritems():
+ hostname = machine.get_hostname()
slave_logging_dir = os.path.join(logging_root, hostname + "/")
try:
@@ -520,90 +306,14 @@ class NetTestController:
raise NetTestError(msg)
capture_files = self._remote_capture_files[machine_id]
- for dev_id, remote_path in capture_files.iteritems():
- filename = "%s.pcap" % dev_id
+ for if_id, remote_path in capture_files.iteritems():
+ filename = "%s.pcap" % if_id
local_path = os.path.join(slave_logging_dir, filename)
- self._copy_from_slave(machine_id, remote_path, local_path)
+ machine.copy_file_from_machine(remote_path, local_path)
logging.info("pcap files from machine %s stored at %s",
machine_id, slave_logging_dir)
- def _update_system_config(self, machine_id, res_data, persistent=False):
- info = self._get_machineinfo(machine_id)
- system_config = info["system_config"]
- for option, values in res_data.iteritems():
- if persistent:
- if option in system_config:
- del system_config[option]
- else:
- if not option in system_config:
- initial_val = {"initial_val": values["previous_val"]}
- system_config[option] = initial_val
- system_config[option]["current_val"] = values["current_val"]
-
-
- def _restore_system_config(self, machine_id):
- info = self._get_machineinfo(machine_id)
- system_config = info["system_config"]
-
- if len(system_config) > 0:
- command = {}
- command["machine_id"] = machine_id
- command["type"] = "system_config"
- command["value"] = ""
- command["options"] = {}
- command["persistent"] = True
- for option, values in system_config.iteritems():
- command["options"][option] = [{"value": values["initial_val"]}]
-
- seq = {"commands": [command], "quit_on_fail": "no"}
- self._run_command_sequence(seq)
- info["system_config"] = {}
-
- def _rpc_call(self, machine_id, method_name, *args):
- data = {}
- data["type"] = "command"
- data["method_name"] = method_name
- data["args"] = args
-
- self._msg_dispatcher.send_message(machine_id, data)
-
- result = self._msg_dispatcher.wait_for_result(machine_id)
-
- return result
-
- def _copy_to_slave(self, local_path, machine_id, remote_path=None):
- remote_path = self._rpc_call(machine_id, "start_copy_to", remote_path)
- f = open(local_path, "rb")
-
- while True:
- data = f.read(1024*1024) # 1MB buffer
- if len(data) == 0:
- break
-
- self._rpc_call(machine_id, "copy_part_to",
- remote_path, Binary(data))
-
- self._rpc_call(machine_id, "finish_copy_to", remote_path)
- return remote_path
-
- def _copy_from_slave(self, machine_id, remote_path, local_path):
- status = self._rpc_call(machine_id, "start_copy_from", remote_path)
- if not status:
- raise NetTestError("The requested file cannot be transfered." \
- "It does not exist on machine %s" % machine_id)
-
- local_file = open(local_path, "wb")
-
- binary = "next"
- while binary != "":
- binary = self._rpc_call(machine_id, "copy_part_from",
- remote_path, 1024*1024) # 1MB buffer
- local_file.write(binary.data)
-
- local_file.close()
- self._rpc_call(machine_id, "finish_copy_from", remote_path)
-
def _load_test_modules(self, dirs):
modules = {}
for dir_name in dirs:
diff --git a/lnst/Controller/SlavePool.py b/lnst/Controller/SlavePool.py
index 9a9b3cd..9674fb7 100644
--- a/lnst/Controller/SlavePool.py
+++ b/lnst/Controller/SlavePool.py
@@ -21,6 +21,7 @@ from xml.dom import minidom
from lnst.Common.XmlProcessing import XmlDomTreeInit
from lnst.Common.NetUtils import test_tcp_connection
from lnst.Controller.SlaveMachineParse import SlaveMachineParse
+from lnst.Controller.Machine import Machine
class SlavePool:
"""
@@ -85,7 +86,7 @@ class SlavePool:
logging.warning("libvirtd not found- Machine Pool skipping "\
"machine %s" % machine_id)
- def provision_setup(self, setup_requirements):
+ def provision_machines(self, mreqs):
"""
This method will try to map a dictionary of machines'
requirements to a pool of machines that is available to
@@ -99,16 +100,16 @@ class SlavePool:
"""
mapper = SetupMapper()
- self._map = mapper.map_setup(setup_requirements, self._pool)
+ self._map = mapper.map_setup(mreqs, self._pool)
if self._map == None:
return None
- configs = {}
+ machines = {}
for m_id in self._map["machines"]:
- configs[m_id] = self._get_mapped_slave(m_id)
+ machines[m_id] = self._get_mapped_slave(m_id)
- return configs
+ return machines
def get_provisioner_id(self, m_id):
try:
@@ -134,20 +135,33 @@ class SlavePool:
def _get_mapped_slave(self, tm_id):
pm_id = self._get_machine_mapping(tm_id)
+ pm = self._pool[pm_id]
- machine = copy.deepcopy(self._pool[pm_id])
+ hostname = pm["params"]["hostname"]
+ libvirt_domain = pm["params"]["libvirt_domain"]
- new_interfaces = {}
+ machine = Machine(tm_id, hostname, libvirt_domain)
+
+ used = []
if_map = self._map["machines"][tm_id]["interfaces"]
for t_if, p_if in if_map.iteritems():
- new_interfaces[t_if] = machine["interfaces"][p_if]
+ used.append(p_if)
+ if_data = pm["interfaces"][p_if]
for t_net, p_net in self._map["networks"].iteritems():
- if new_interfaces[t_if]["network"] == p_net:
- new_interfaces[t_if]["network"] = t_net
+ if pm["interfaces"][p_if]["network"] == p_net:
break
- machine["interfaces"] = new_interfaces
+ iface = machine.new_static_interface(t_if, "eth")
+ iface.set_hwaddr(if_data["hwaddr"])
+ iface.set_network(t_net)
+
+ for if_id, if_data in pm["interfaces"].iteritems():
+ if if_id not in used:
+ iface = machine.new_unused_interface("eth")
+ iface.set_hwaddr(if_data["hwaddr"])
+ iface.set_network(t_net)
+
return machine
class SetupMapper:
@@ -190,13 +204,13 @@ class SetupMapper:
_pool_machines = None
@staticmethod
- def _get_topology(machine_configs):
+ def _get_topology(machine_desc):
"""
This function will generate an adjacenty list from machine
configuration dictionary. It can handle both machines and
templates.
- :param machine_configs: dictionary of machines in the topology
+ :param machine_desc: dictionary of machines in the topology
:type machines_configs: dict
:return: Topology - neighbour connection list (adjacency list-like
@@ -205,7 +219,7 @@ class SetupMapper:
"""
networks = {}
- for m_id, m_config in machine_configs.iteritems():
+ for m_id, m_config in machine_desc.iteritems():
for dev_id, dev_info in m_config["interfaces"].iteritems():
net = dev_info["network"]
if not net in networks:
@@ -213,7 +227,7 @@ class SetupMapper:
networks[net].append((m_id, dev_id))
topology = {}
- for m_id, m_config in machine_configs.iteritems():
+ for m_id, m_config in machine_desc.iteritems():
topology[m_id] = []
for net_name, net in networks.iteritems():
devs_in_net = []
@@ -316,7 +330,12 @@ class SetupMapper:
t_if = self._template_machines[tm_id]["interfaces"][t_if_id]
p_if = self._pool_machines[pm_id]["interfaces"][pm_if_id]
- properties = ["type", "hwaddr"]
+
+ for prop_name, prop_value in t_if["params"].iteritems():
+ if p_if["params"][prop_name] != prop_value:
+ return False
+
+ properties = ["type"]
for prop_name, prop_value in t_if.iteritems():
if prop_name in properties:
if p_if[prop_name] != prop_value: