X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/6906a9d8e61fedad8779b0fe5e56756a403a2361..9205a895ac0b5d60b881458a44bb157b88eac21d:/lib/rpc.py diff --git a/lib/rpc.py b/lib/rpc.py index fc71a93..6dc8ab4 100644 --- a/lib/rpc.py +++ b/lib/rpc.py @@ -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. @@ -213,7 +217,7 @@ 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" @@ -234,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) @@ -255,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 @@ -263,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 overriden hypervisor parameters + @type bep: dict or None + @param bep: a dictionary with overriden backend parameters @rtype: dict @return: the instance dict, with the hvparams filled with the cluster defaults @@ -271,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} @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() @@ -290,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: @@ -301,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} @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 @@ -325,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 @@ -346,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] @@ -414,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. @@ -504,24 +523,23 @@ 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): + def call_instance_os_add(self, node, inst, reinstall): """Installs an OS on the given instance. This is a single-node call. """ return self._SingleNodeCall(node, "instance_os_add", - [self._InstDict(inst)]) + [self._InstDict(inst), reinstall]) def call_instance_run_rename(self, node, inst, old_name): """Run the OS rename script for an instance. @@ -1012,6 +1030,16 @@ class RpcRunner(object): """ return self._SingleNodeCall(node, "node_demote_from_mc", []) + + def call_node_powercycle(self, node, hypervisor): + """Tries to powercycle a node. + + This is a single-node call. + + """ + return self._SingleNodeCall(node, "node_powercycle", [hypervisor]) + + def call_test_delay(self, node_list, duration): """Sleep for a fixed time on given node(s).