commit 8858385938cb9f4fe4916a00a0ab10c6d1087dd5 Author: Ondrej Lichtner olichtne@redhat.com Date: Fri Mar 8 16:24:46 2013 +0100
NetTestController: reimplement communication
This commit reimplements the communication from the controllers perspective.
I added a new class MessageDispatcher that inherits the class ConnectionHandler. This class adds new methods that can be easily used to send messages to the slaves. It encapsulates communication with every slave used by the controller. For this reason this patch also removes the rpc objects from machines infos and related methods. Instead we create a normal TCP socket that is stored in the MessageDispatcher object.
For compatibility reasons I simple rewrote the _rpc_call method so that the communication just works with through the MessageDispatcher but I expect that this will change in the future when we refactor the rest of the controller.
Signed-off-by: Ondrej Lichtner olichtne@redhat.com
lnst/Controller/NetTestController.py | 104 ++++++++++++++++++++++------------ 1 files changed, 68 insertions(+), 36 deletions(-) --- diff --git a/lnst/Controller/NetTestController.py b/lnst/Controller/NetTestController.py index c034f33..e84c658 100644 --- a/lnst/Controller/NetTestController.py +++ b/lnst/Controller/NetTestController.py @@ -30,6 +30,8 @@ from lnst.Common.NetTestCommand import NetTestCommandContext, NetTestCommand from lnst.Common.NetTestCommand import str_command from lnst.Controller.NetTestParse import NetTestParse from lnst.Controller.SlavePool import SlavePool +from lnst.Common.ConnectionHandler import send_data, recv_data +from lnst.Common.ConnectionHandler import ConnectionHandler
class NetTestError(Exception): pass @@ -46,6 +48,7 @@ class NetTestController: self._config = config self._log_ctl = log_ctl self._recipe_path = recipe_path + self._msg_dispatcher = MessageDispatcher()
sp = SlavePool(config.get_option('environment', 'pool_dirs'), check_process_running("libvirtd")) @@ -89,15 +92,6 @@ class NetTestController:
return info
- def _get_machinerpc(self, machine_id): - try: - rpc = self._get_machineinfo(machine_id)["rpc"] - except KeyError: - msg = "XMLRPC connection required, but not yet available" - raise NetTestError(msg) - - return rpc - @staticmethod def _session_die(session, status): logging.debug("%s terminated with status %s", session.command, status) @@ -244,14 +238,13 @@ class NetTestController: domain_ctl = VirtDomainCtl(info["libvirt_domain"]) info["virt_domain_ctl"] = domain_ctl
- if not "rpc" in info: - self._init_slave_logging(machine_id) - self._init_slave_rpc(machine_id) + self._init_slave_logging(machine_id) + self._init_slave_rpc(machine_id)
- info["configured_interfaces"] = [] + info["configured_interfaces"] = []
- self._rpc_call(machine_id, "clear_resource_table") - required = self._resource_table + self._rpc_call(machine_id, "clear_resource_table") + required = self._resource_table
if self._docleanup and not info["skip_cleanup"]: self._rpc_call(machine_id, 'machine_cleanup') @@ -310,9 +303,9 @@ class NetTestController: port = self._config.get_option('environment', 'rpcport') logging.info("Connecting to RPC on machine %s", hostname)
- url = "http://%s:%d" % (hostname, port) - rpc = ServerProxy(url, allow_none = True) - info["rpc"] = rpc + rpc = socket.create_connection((hostname, port)) + self._msg_dispatcher.add_slave(machine_id, rpc, info) + if self._rpc_call(machine_id, 'hello', self._recipe_path) != "hello": msg = "Unable to establish RPC connection to machine %s. " \ % hostname @@ -330,7 +323,7 @@ class NetTestController: return for machine_id in self._recipe["machines"]: info = self._get_machineinfo(machine_id) - if "rpc" not in info or "configured_interfaces" not in info: + if "configured_interfaces" not in info: continue
self._rpc_call(machine_id, "bye") @@ -557,26 +550,16 @@ class NetTestController: info["system_config"] = {}
def _rpc_call(self, machine_id, method_name, *args): - rpc = self._get_machinerpc(machine_id) - rpc_method = getattr(rpc, method_name) - - result = rpc_method(*args) + data = {} + data["type"] = "command" + data["method_name"] = method_name + data["args"] = args
- logs = rpc.get_new_logs() - self._add_client_logs(machine_id, logs) - return result + self._msg_dispatcher.send_message(machine_id, data)
- def _add_client_logs(self, machine_id, logs): - info = self._get_machineinfo(machine_id) - address = socket.gethostbyname(info['hostname']) - logger = info['logger'] + result = self._msg_dispatcher.wait_for_result(machine_id)
- for log in logs: - data = log.data - data = pickle.loads(data) - data['address'] = '(' + address + ')' - record = logging.makeLogRecord(data) - logger.handle(record) + 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) @@ -653,3 +636,52 @@ class NetTestController: packages[pkg_name] = {"path": pkg_path, "hash": pkg_hash} return packages + +class MessageDispatcher(ConnectionHandler): + def __init__(self): + super(MessageDispatcher, self).__init__() + self._slaves = {} + + def add_slave(self, machine_id, connection, machine_info): + self._slaves[machine_id] = machine_info + self.add_connection(machine_id, connection) + + def send_message(self, machine_id, data): + soc = self.get_connection(machine_id) + + if send_data(soc, data) == False: + msg = "Connection error from slave %s" % machine_id + raise NetTestError(msg) + + def wait_for_result(self, machine_id): + wait = True + while wait: + messages = self.check_connections() + for msg in messages: + if msg[1]["type"] == "result" and msg[0] == machine_id: + wait = False + result = msg[1]["result"] + else: + self._process_message(msg) + + return result + + def _process_message(self, message): + if message[1]["type"] == "log": + record = message[1]["record"] + self._add_client_log(message[0], record) + elif message[1]["type"] == "result": + msg = "Recieved result message from different slave %s" % message[0] + logging.debug(msg) + else: + msg = "Unknown message type: %s" % message[1]["type"] + raise NetTestError(msg) + + def _add_client_log(self, machine_id, log_record): + info = self._slaves[machine_id] + address = socket.gethostbyname(info['hostname']) + logger = info['logger'] + + log_record['address'] = '(' + address + ')' + record = logging.makeLogRecord(log_record) + logger.handle(record)
lnst-developers@lists.fedorahosted.org