X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/dd875d32f454ef0898d9cf3e102734144aa161a4..fc8a6b8f3334c1352caafcede1e31d801fc146ba:/lib/rpc.py diff --git a/lib/rpc.py b/lib/rpc.py index a9619f9..b1054ca 100644 --- a/lib/rpc.py +++ b/lib/rpc.py @@ -31,7 +31,6 @@ # R0904: Too many public methods import os -import socket import logging import zlib import base64 @@ -43,7 +42,8 @@ from ganeti import serializer from ganeti import constants from ganeti import errors -import ganeti.http.client +# pylint has a bug here, doesn't see this import +import ganeti.http.client # pylint: disable-msg=W0611 # Module level variable @@ -56,7 +56,7 @@ def Init(): Must be called before using any RPC function. """ - global _http_manager + global _http_manager # pylint: disable-msg=W0603 assert not _http_manager, "RPC module initialized more than once" @@ -69,7 +69,7 @@ def Shutdown(): Must be called before quitting the program. """ - global _http_manager + global _http_manager # pylint: disable-msg=W0603 if _http_manager: _http_manager.Shutdown() @@ -83,7 +83,7 @@ class RpcResult(object): calls we can't raise an exception just because one one out of many failed, and therefore we use this class to encapsulate the result. - @ivar data: the data payload, for successfull results, or None + @ivar data: the data payload, for successful results, or None @type failed: boolean @ivar failed: whether the operation failed at RPC level (not application level on the remote node) @@ -104,13 +104,17 @@ class RpcResult(object): if offline: self.failed = True self.error = "Node is marked offline" - self.data = None + self.data = self.payload = None elif failed: self.error = data - self.data = None + self.data = self.payload = None else: self.data = data self.error = None + if isinstance(data, (tuple, list)) and len(data) == 2: + self.payload = data[1] + else: + self.payload = None def Raise(self): """If the result has failed, raise an OpExecError. @@ -123,6 +127,32 @@ class RpcResult(object): raise errors.OpExecError("Call '%s' to node '%s' has failed: %s" % (self.call, self.node, self.error)) + def RemoteFailMsg(self): + """Check if the remote procedure failed. + + This is valid only for RPC calls which return result of the form + (status, data | error_msg). + + @return: empty string for succcess, otherwise an error message + + """ + def _EnsureErr(val): + """Helper to ensure we return a 'True' value for error.""" + if val: + return val + else: + return "No error information" + + if self.failed: + return _EnsureErr(self.error) + if not isinstance(self.data, (tuple, list)): + return "Invalid result type (%s)" % type(self.data) + if len(self.data) != 2: + return "Invalid result length (%d), expected 2" % len(self.data) + if not self.data[0]: + return _EnsureErr(self.data[1]) + return "" + class Client: """RPC Client class. @@ -131,7 +161,7 @@ class Client: list of nodes, will contact (in parallel) all nodes, and return a dict of results (key: node name, value: result). - One current bug is that generic failure is still signalled by + One current bug is that generic failure is still signaled by 'False' result, which is not good. This overloading of values can cause bugs. @@ -187,10 +217,10 @@ class Client: """Call nodes and return results. @rtype: list - @returns: List of RPC results + @return: List of RPC results """ - assert _http_manager, "RPC module not intialized" + assert _http_manager, "RPC module not initialized" _http_manager.ExecRequests(self.nc.values()) @@ -208,7 +238,8 @@ class Client: else: msg = req.resp_body - logging.error("RPC error from node %s: %s", name, msg) + logging.error("RPC error in %s from node %s: %s", + self.procedure, name, msg) results[name] = RpcResult(data=msg, failed=True, node=name, call=self.procedure) @@ -229,7 +260,7 @@ class RpcRunner(object): self._cfg = cfg self.port = utils.GetNodeDaemonPort() - def _InstDict(self, instance): + def _InstDict(self, instance, hvp=None, bep=None): """Convert the given instance to a dict. This is done via the instance's ToDict() method and additionally @@ -237,6 +268,10 @@ class RpcRunner(object): @type instance: L{objects.Instance} @param instance: an Instance object + @type hvp: dict or None + @param hvp: a dictionary with overridden hypervisor parameters + @type bep: dict or None + @param bep: a dictionary with overridden backend parameters @rtype: dict @return: the instance dict, with the hvparams filled with the cluster defaults @@ -245,16 +280,23 @@ class RpcRunner(object): idict = instance.ToDict() cluster = self._cfg.GetClusterInfo() idict["hvparams"] = cluster.FillHV(instance) + if hvp is not None: + idict["hvparams"].update(hvp) idict["beparams"] = cluster.FillBE(instance) + if bep is not None: + idict["beparams"].update(bep) return idict - def _ConnectList(self, client, node_list): + def _ConnectList(self, client, node_list, call): """Helper for computing node addresses. - @type client: L{Client} + @type client: L{ganeti.rpc.Client} @param client: a C{Client} instance @type node_list: list @param node_list: the node list we should connect + @type call: string + @param call: the name of the remote procedure call, for filling in + correctly any eventual offline nodes' results """ all_nodes = self._cfg.GetAllNodesInfo() @@ -264,7 +306,7 @@ class RpcRunner(object): for node in node_list: if node in all_nodes: if all_nodes[node].offline: - skip_dict[node] = RpcResult(node=node, offline=True) + skip_dict[node] = RpcResult(node=node, offline=True, call=call) continue val = all_nodes[node].primary_ip else: @@ -275,19 +317,22 @@ class RpcRunner(object): client.ConnectList(name_list, address_list=addr_list) return skip_dict - def _ConnectNode(self, client, node): + def _ConnectNode(self, client, node, call): """Helper for computing one node's address. - @type client: L{Client} + @type client: L{ganeti.rpc.Client} @param client: a C{Client} instance @type node: str @param node: the node we should connect + @type call: string + @param call: the name of the remote procedure call, for filling in + correctly any eventual offline nodes' results """ node_info = self._cfg.GetNodeInfo(node) if node_info is not None: if node_info.offline: - return RpcResult(node=node, offline=True) + return RpcResult(node=node, offline=True, call=call) addr = node_info.primary_ip else: addr = None @@ -299,7 +344,7 @@ class RpcRunner(object): """ body = serializer.DumpJson(args, indent=False) c = Client(procedure, body, self.port) - skip_dict = self._ConnectList(c, node_list) + skip_dict = self._ConnectList(c, node_list, procedure) skip_dict.update(c.GetResults()) return skip_dict @@ -320,7 +365,7 @@ class RpcRunner(object): """ body = serializer.DumpJson(args, indent=False) c = Client(procedure, body, self.port) - result = self._ConnectNode(c, node) + result = self._ConnectNode(c, node, procedure) if result is None: # we did connect, node is not offline result = c.GetResults()[node] @@ -388,14 +433,14 @@ class RpcRunner(object): """ return self._SingleNodeCall(node, "bridges_exist", [bridges_list]) - def call_instance_start(self, node, instance, extra_args): + def call_instance_start(self, node, instance, hvp, bep): """Starts an instance. This is a single-node call. """ - return self._SingleNodeCall(node, "instance_start", - [self._InstDict(instance), extra_args]) + idict = self._InstDict(instance, hvp=hvp, bep=bep) + return self._SingleNodeCall(node, "instance_start", [idict]) def call_instance_shutdown(self, node, instance): """Stops an instance. @@ -406,6 +451,59 @@ class RpcRunner(object): return self._SingleNodeCall(node, "instance_shutdown", [self._InstDict(instance)]) + def call_migration_info(self, node, instance): + """Gather the information necessary to prepare an instance migration. + + This is a single-node call. + + @type node: string + @param node: the node on which the instance is currently running + @type instance: C{objects.Instance} + @param instance: the instance definition + + """ + return self._SingleNodeCall(node, "migration_info", + [self._InstDict(instance)]) + + def call_accept_instance(self, node, instance, info, target): + """Prepare a node to accept an instance. + + This is a single-node call. + + @type node: string + @param node: the target node for the migration + @type instance: C{objects.Instance} + @param instance: the instance definition + @type info: opaque/hypervisor specific (string/data) + @param info: result for the call_migration_info call + @type target: string + @param target: target hostname (usually ip address) (on the node itself) + + """ + return self._SingleNodeCall(node, "accept_instance", + [self._InstDict(instance), info, target]) + + def call_finalize_migration(self, node, instance, info, success): + """Finalize any target-node migration specific operation. + + This is called both in case of a successful migration and in case of error + (in which case it should abort the migration). + + This is a single-node call. + + @type node: string + @param node: the target node for the migration + @type instance: C{objects.Instance} + @param instance: the instance definition + @type info: opaque/hypervisor specific (string/data) + @param info: result for the call_migration_info call + @type success: boolean + @param success: whether the migration was a success or a failure + + """ + return self._SingleNodeCall(node, "finalize_migration", + [self._InstDict(instance), info, success]) + def call_instance_migrate(self, node, instance, target, live): """Migrate an instance. @@ -425,15 +523,14 @@ class RpcRunner(object): return self._SingleNodeCall(node, "instance_migrate", [self._InstDict(instance), target, live]) - def call_instance_reboot(self, node, instance, reboot_type, extra_args): + def call_instance_reboot(self, node, instance, reboot_type): """Reboots an instance. This is a single-node call. """ return self._SingleNodeCall(node, "instance_reboot", - [self._InstDict(instance), reboot_type, - extra_args]) + [self._InstDict(instance), reboot_type]) def call_instance_os_add(self, node, inst): """Installs an OS on the given instance. @@ -468,6 +565,20 @@ class RpcRunner(object): """ return self._SingleNodeCall(node, "instance_info", [instance, hname]) + def call_instance_migratable(self, node, instance): + """Checks whether the given instance can be migrated. + + This is a single-node call. + + @param node: the node to query + @type instance: L{objects.Instance} + @param instance: the instance to check + + + """ + return self._SingleNodeCall(node, "instance_migratable", + [self._InstDict(instance)]) + def call_all_instances_info(self, node_list, hypervisor_list): """Returns information about all instances on the given nodes. @@ -538,6 +649,10 @@ class RpcRunner(object): for result in retux.itervalues(): if result.failed or not isinstance(result.data, dict): result.data = {} + if result.offline: + log_name = None + else: + log_name = "call_node_info" utils.CheckDict(result.data, { 'memory_total' : '-', @@ -545,7 +660,7 @@ class RpcRunner(object): 'memory_free' : '-', 'vg_size' : 'node_unreachable', 'vg_free' : '-', - }, "call_node_info") + }, log_name) return retux def call_node_add(self, node, dsa, dsapub, rsa, rsapub, ssh, sshpub): @@ -567,14 +682,14 @@ class RpcRunner(object): [checkdict, cluster_name]) @classmethod - def call_node_start_master(cls, node, start_daemons): + def call_node_start_master(cls, node, start_daemons, no_voting): """Tells a node to activate itself as a master. This is a single-node call. """ return cls._StaticSingleNodeCall(node, "node_start_master", - [start_daemons]) + [start_daemons, no_voting]) @classmethod def call_node_stop_master(cls, node, stop_daemons): @@ -683,14 +798,52 @@ class RpcRunner(object): """ return self._SingleNodeCall(node, "blockdev_find", [disk.ToDict()]) - def call_blockdev_close(self, node, disks): + def call_blockdev_close(self, node, instance_name, disks): """Closes the given block devices. This is a single-node call. """ - return self._SingleNodeCall(node, "blockdev_close", - [cf.ToDict() for cf in disks]) + params = [instance_name, [cf.ToDict() for cf in disks]] + return self._SingleNodeCall(node, "blockdev_close", params) + + def call_blockdev_getsizes(self, node, disks): + """Returns the size of the given disks. + + This is a single-node call. + + """ + params = [[cf.ToDict() for cf in disks]] + return self._SingleNodeCall(node, "blockdev_getsize", params) + + def call_drbd_disconnect_net(self, node_list, nodes_ip, disks): + """Disconnects the network of the given drbd devices. + + This is a multi-node call. + + """ + return self._MultiNodeCall(node_list, "drbd_disconnect_net", + [nodes_ip, [cf.ToDict() for cf in disks]]) + + def call_drbd_attach_net(self, node_list, nodes_ip, + disks, instance_name, multimaster): + """Disconnects the given drbd devices. + + This is a multi-node call. + + """ + return self._MultiNodeCall(node_list, "drbd_attach_net", + [nodes_ip, [cf.ToDict() for cf in disks], + instance_name, multimaster]) + + def call_drbd_wait_sync(self, node_list, nodes_ip, disks): + """Waits for the synchronization of drbd devices is complete. + + This is a multi-node call. + + """ + return self._MultiNodeCall(node_list, "drbd_wait_sync", + [nodes_ip, [cf.ToDict() for cf in disks]]) @classmethod def call_upload_file(cls, node_list, file_name, address_list=None): @@ -815,7 +968,10 @@ class RpcRunner(object): """ flat_disks = [] for disk in snap_disks: - flat_disks.append(disk.ToDict()) + if isinstance(disk, bool): + flat_disks.append(disk) + else: + flat_disks.append(disk.ToDict()) return self._SingleNodeCall(node, "finalize_export", [self._InstDict(instance), flat_disks])