X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/dfa96ded38527e009bb5d068e9fdf54a2ddc3ab7..29df1f02bbc588ace14f4a40e346aa502b686462:/lib/cmdlib.py diff --git a/lib/cmdlib.py b/lib/cmdlib.py index cb40338..65c52a2 100644 --- a/lib/cmdlib.py +++ b/lib/cmdlib.py @@ -26,7 +26,6 @@ import os import os.path import sha -import socket import time import tempfile import re @@ -70,7 +69,7 @@ class LogicalUnit(object): validity. """ - self.processor = processor + self.proc = processor self.op = op self.cfg = cfg self.sstore = sstore @@ -164,6 +163,23 @@ class NoHooksLU(LogicalUnit): return {}, [], [] +def _AddHostToEtcHosts(hostname): + """Wrapper around utils.SetEtcHostsEntry. + + """ + hi = utils.HostInfo(name=hostname) + utils.SetEtcHostsEntry(constants.ETC_HOSTS, hi.ip, hi.name, [hi.ShortName()]) + + +def _RemoveHostFromEtcHosts(hostname): + """Wrapper around utils.RemoveEtcHostsEntry. + + """ + hi = utils.HostInfo(name=hostname) + utils.RemoveEtcHostsEntry(constants.ETC_HOSTS, hi.name) + utils.RemoveEtcHostsEntry(constants.ETC_HOSTS, hi.ShortName()) + + def _GetWantedNodes(lu, nodes): """Returns list of checked and expanded node names. @@ -251,11 +267,12 @@ def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status, if nics: nic_count = len(nics) - for idx, (ip, bridge) in enumerate(nics): + for idx, (ip, bridge, mac) in enumerate(nics): if ip is None: ip = "" env["INSTANCE_NIC%d_IP" % idx] = ip env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge + env["INSTANCE_NIC%d_HWADDR" % idx] = mac else: nic_count = 0 @@ -279,93 +296,13 @@ def _BuildInstanceHookEnvByObject(instance, override=None): 'status': instance.os, 'memory': instance.memory, 'vcpus': instance.vcpus, - 'nics': [(nic.ip, nic.bridge) for nic in instance.nics], + 'nics': [(nic.ip, nic.bridge, nic.mac) for nic in instance.nics], } if override: args.update(override) return _BuildInstanceHookEnv(**args) -def _UpdateEtcHosts(fullnode, ip): - """Ensure a node has a correct entry in /etc/hosts. - - Args: - fullnode - Fully qualified domain name of host. (str) - ip - IPv4 address of host (str) - - """ - node = fullnode.split(".", 1)[0] - - f = open('/etc/hosts', 'r+') - - inthere = False - - save_lines = [] - add_lines = [] - removed = False - - while True: - rawline = f.readline() - - if not rawline: - # End of file - break - - line = rawline.split('\n')[0] - - # Strip off comments - line = line.split('#')[0] - - if not line: - # Entire line was comment, skip - save_lines.append(rawline) - continue - - fields = line.split() - - haveall = True - havesome = False - for spec in [ ip, fullnode, node ]: - if spec not in fields: - haveall = False - if spec in fields: - havesome = True - - if haveall: - inthere = True - save_lines.append(rawline) - continue - - if havesome and not haveall: - # Line (old, or manual?) which is missing some. Remove. - removed = True - continue - - save_lines.append(rawline) - - if not inthere: - add_lines.append('%s\t%s %s\n' % (ip, fullnode, node)) - - if removed: - if add_lines: - save_lines = save_lines + add_lines - - # We removed a line, write a new file and replace old. - fd, tmpname = tempfile.mkstemp('tmp', 'hosts_', '/etc') - newfile = os.fdopen(fd, 'w') - newfile.write(''.join(save_lines)) - newfile.close() - os.rename(tmpname, '/etc/hosts') - - elif add_lines: - # Simply appending a new line will do the trick. - f.seek(0, 2) - for add in add_lines: - f.write(add) - - f.close() - - def _UpdateKnownHosts(fullnode, ip, pubkey): """Ensure a node has a correct known_hosts entry. @@ -563,6 +500,12 @@ class LUInitCluster(LogicalUnit): if config.ConfigWriter.IsCluster(): raise errors.OpPrereqError("Cluster is already initialised") + if self.op.hypervisor_type == constants.HT_XEN_HVM31: + if not os.path.exists(constants.VNC_PASSWORD_FILE): + raise errors.OpPrereqError("Please prepare the cluster VNC" + "password file %s" % + constants.VNC_PASSWORD_FILE) + self.hostname = hostname = utils.HostInfo() if hostname.ip.startswith("127."): @@ -586,8 +529,8 @@ class LUInitCluster(LogicalUnit): secondary_ip != hostname.ip and (not utils.TcpPing(constants.LOCALHOST_IP_ADDRESS, secondary_ip, constants.DEFAULT_NODED_PORT))): - raise errors.OpPrereqError("You gave %s as secondary IP,\n" - "but it does not belong to this host." % + raise errors.OpPrereqError("You gave %s as secondary IP," + " but it does not belong to this host." % secondary_ip) self.secondary_ip = secondary_ip @@ -602,7 +545,7 @@ class LUInitCluster(LogicalUnit): raise errors.OpPrereqError("Invalid mac prefix given '%s'" % self.op.mac_prefix) - if self.op.hypervisor_type not in hypervisor.VALID_HTYPES: + if self.op.hypervisor_type not in constants.HYPER_TYPES: raise errors.OpPrereqError("Invalid hypervisor type given '%s'" % self.op.hypervisor_type) @@ -612,6 +555,11 @@ class LUInitCluster(LogicalUnit): (self.op.master_netdev, result.output.strip())) + if not (os.path.isfile(constants.NODE_INITD_SCRIPT) and + os.access(constants.NODE_INITD_SCRIPT, os.X_OK)): + raise errors.OpPrereqError("Init.d script '%s' missing or not" + " executable." % constants.NODE_INITD_SCRIPT) + def Exec(self, feedback_fn): """Initialize the cluster. @@ -641,7 +589,7 @@ class LUInitCluster(LogicalUnit): f.close() sshkey = sshline.split(" ")[1] - _UpdateEtcHosts(hostname.name, hostname.ip) + _AddHostToEtcHosts(hostname.name) _UpdateKnownHosts(hostname.name, hostname.ip, sshkey) @@ -683,10 +631,11 @@ class LUDestroyCluster(NoHooksLU): """Destroys the cluster. """ + master = self.sstore.GetMasterNode() priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS) utils.CreateBackup(priv_key) utils.CreateBackup(pub_key) - rpc.call_node_leave_cluster(self.sstore.GetMasterNode()) + rpc.call_node_leave_cluster(master) class LUVerifyCluster(NoHooksLU): @@ -854,9 +803,9 @@ class LUVerifyCluster(NoHooksLU): """ bad = False feedback_fn("* Verifying global settings") - self.cfg.VerifyConfig() + for msg in self.cfg.VerifyConfig(): + feedback_fn(" - ERROR: %s" % msg) - master = self.sstore.GetMasterNode() vg_name = self.cfg.GetVGName() nodelist = utils.NiceSort(self.cfg.GetNodeList()) instancelist = utils.NiceSort(self.cfg.GetInstanceList()) @@ -892,12 +841,17 @@ class LUVerifyCluster(NoHooksLU): # node_volume volumeinfo = all_volumeinfo[node] - if type(volumeinfo) != dict: + if isinstance(volumeinfo, basestring): + feedback_fn(" - ERROR: LVM problem on node %s: %s" % + (node, volumeinfo[-400:].encode('string_escape'))) + bad = True + node_volume[node] = {} + elif not isinstance(volumeinfo, dict): feedback_fn(" - ERROR: connection to %s failed" % (node,)) bad = True continue - - node_volume[node] = volumeinfo + else: + node_volume[node] = volumeinfo # node_instance nodeinstance = all_instanceinfo[node] @@ -933,6 +887,78 @@ class LUVerifyCluster(NoHooksLU): return int(bad) +class LUVerifyDisks(NoHooksLU): + """Verifies the cluster disks status. + + """ + _OP_REQP = [] + + def CheckPrereq(self): + """Check prerequisites. + + This has no prerequisites. + + """ + pass + + def Exec(self, feedback_fn): + """Verify integrity of cluster disks. + + """ + result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {} + + vg_name = self.cfg.GetVGName() + nodes = utils.NiceSort(self.cfg.GetNodeList()) + instances = [self.cfg.GetInstanceInfo(name) + for name in self.cfg.GetInstanceList()] + + nv_dict = {} + for inst in instances: + inst_lvs = {} + if (inst.status != "up" or + inst.disk_template not in constants.DTS_NET_MIRROR): + continue + inst.MapLVsByNode(inst_lvs) + # transform { iname: {node: [vol,],},} to {(node, vol): iname} + for node, vol_list in inst_lvs.iteritems(): + for vol in vol_list: + nv_dict[(node, vol)] = inst + + if not nv_dict: + return result + + node_lvs = rpc.call_volume_list(nodes, vg_name) + + to_act = set() + for node in nodes: + # node_volume + lvs = node_lvs[node] + + if isinstance(lvs, basestring): + logger.Info("error enumerating LVs on node %s: %s" % (node, lvs)) + res_nlvm[node] = lvs + elif not isinstance(lvs, dict): + logger.Info("connection to node %s failed or invalid data returned" % + (node,)) + res_nodes.append(node) + continue + + for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems(): + inst = nv_dict.pop((node, lv_name), None) + if (not lv_online and inst is not None + and inst.name not in res_instances): + res_instances.append(inst.name) + + # any leftover items in nv_dict are missing LVs, let's arrange the + # data better + for key, inst in nv_dict.iteritems(): + if inst.name not in res_missing: + res_missing[inst.name] = [] + res_missing[inst.name].append(key) + + return result + + class LURenameCluster(LogicalUnit): """Rename the cluster. @@ -1008,11 +1034,11 @@ class LURenameCluster(LogicalUnit): (fname, to_node)) finally: if not rpc.call_node_start_master(master): - logger.Error("Could not re-enable the master role on the master,\n" - "please restart manually.") + logger.Error("Could not re-enable the master role on the master," + " please restart manually.") -def _WaitForSync(cfgw, instance, oneshot=False, unlock=False): +def _WaitForSync(cfgw, instance, proc, oneshot=False, unlock=False): """Sleep and poll for an instance's disk to sync. """ @@ -1020,7 +1046,7 @@ def _WaitForSync(cfgw, instance, oneshot=False, unlock=False): return True if not oneshot: - logger.ToStdout("Waiting for instance %s to sync disks." % instance.name) + proc.LogInfo("Waiting for instance %s to sync disks." % instance.name) node = instance.primary_node @@ -1034,7 +1060,7 @@ def _WaitForSync(cfgw, instance, oneshot=False, unlock=False): cumul_degraded = False rstats = rpc.call_blockdev_getmirrorstatus(node, instance.disks) if not rstats: - logger.ToStderr("Can't get any data from node %s" % node) + proc.LogWarning("Can't get any data from node %s" % node) retries += 1 if retries >= 10: raise errors.RemoteError("Can't contact node %s for mirror data," @@ -1045,10 +1071,11 @@ def _WaitForSync(cfgw, instance, oneshot=False, unlock=False): for i in range(len(rstats)): mstat = rstats[i] if mstat is None: - logger.ToStderr("Can't compute data for node %s/%s" % + proc.LogWarning("Can't compute data for node %s/%s" % (node, instance.disks[i].iv_name)) continue - perc_done, est_time, is_degraded = mstat + # we ignore the ldisk parameter + perc_done, est_time, is_degraded, _ = mstat cumul_degraded = cumul_degraded or (is_degraded and perc_done is None) if perc_done is not None: done = False @@ -1057,8 +1084,8 @@ def _WaitForSync(cfgw, instance, oneshot=False, unlock=False): max_time = est_time else: rem_time = "no time estimate" - logger.ToStdout("- device %s: %5.2f%% done, %s" % - (instance.disks[i].iv_name, perc_done, rem_time)) + proc.LogInfo("- device %s: %5.2f%% done, %s" % + (instance.disks[i].iv_name, perc_done, rem_time)) if done or oneshot: break @@ -1071,15 +1098,23 @@ def _WaitForSync(cfgw, instance, oneshot=False, unlock=False): utils.Lock('cmd') if done: - logger.ToStdout("Instance %s's disks are in sync." % instance.name) + proc.LogInfo("Instance %s's disks are in sync." % instance.name) return not cumul_degraded -def _CheckDiskConsistency(cfgw, dev, node, on_primary): +def _CheckDiskConsistency(cfgw, dev, node, on_primary, ldisk=False): """Check that mirrors are not degraded. + The ldisk parameter, if True, will change the test from the + is_degraded attribute (which represents overall non-ok status for + the device(s)) to the ldisk (representing the local storage status). + """ cfgw.SetDiskID(dev, node) + if ldisk: + idx = 6 + else: + idx = 5 result = True if on_primary or dev.AssembleOnSecondary(): @@ -1088,7 +1123,7 @@ def _CheckDiskConsistency(cfgw, dev, node, on_primary): logger.ToStderr("Can't get any data from node %s" % node) result = False else: - result = result and (not rstats[5]) + result = result and (not rstats[idx]) if dev.children: for child in dev.children: result = result and _CheckDiskConsistency(cfgw, child, node, on_primary) @@ -1193,6 +1228,8 @@ class LURemoveNode(LogicalUnit): self.cfg.RemoveNode(node.name) + _RemoveHostFromEtcHosts(node.name) + class LUQueryNodes(NoHooksLU): """Logical unit for querying nodes. @@ -1448,13 +1485,18 @@ class LUAddNode(LogicalUnit): if not utils.TcpPing(myself.secondary_ip, secondary_ip, constants.DEFAULT_NODED_PORT): - raise errors.OpPrereqError( - "Node secondary ip not reachable by TCP based ping to noded port") + raise errors.OpPrereqError("Node secondary ip not reachable by TCP" + " based ping to noded port") self.new_node = objects.Node(name=node, primary_ip=primary_ip, secondary_ip=secondary_ip) + if self.sstore.GetHypervisorType() == constants.HT_XEN_HVM31: + if not os.path.exists(constants.VNC_PASSWORD_FILE): + raise errors.OpPrereqError("Cluster VNC password file %s missing" % + constants.VNC_PASSWORD_FILE) + def Exec(self, feedback_fn): """Adds the new node to the cluster. @@ -1536,7 +1578,8 @@ class LUAddNode(LogicalUnit): raise errors.OpExecError("Cannot transfer ssh keys to the new node") # Add node to our /etc/hosts, and add key to known_hosts - _UpdateEtcHosts(new_node.name, new_node.primary_ip) + _AddHostToEtcHosts(new_node.name) + _UpdateKnownHosts(new_node.name, new_node.primary_ip, self.cfg.GetHostKey()) @@ -1546,16 +1589,15 @@ class LUAddNode(LogicalUnit): new_node.secondary_ip, constants.DEFAULT_NODED_PORT, 10, False): - raise errors.OpExecError("Node claims it doesn't have the" - " secondary ip you gave (%s).\n" - "Please fix and re-run this command." % - new_node.secondary_ip) + raise errors.OpExecError("Node claims it doesn't have the secondary ip" + " you gave (%s). Please fix and re-run this" + " command." % new_node.secondary_ip) success, msg = ssh.VerifyNodeHostname(node) if not success: raise errors.OpExecError("Node '%s' claims it has a different hostname" - " than the one the resolver gives: %s.\n" - "Please fix and re-run this command." % + " than the one the resolver gives: %s." + " Please fix and re-run this command." % (node, msg)) # Distribute updated /etc/hosts and known_hosts to all nodes, @@ -1574,6 +1616,8 @@ class LUAddNode(LogicalUnit): (fname, to_node)) to_copy = ss.GetFileList() + if self.sstore.GetHypervisorType() == constants.HT_XEN_HVM31: + to_copy.append(constants.VNC_PASSWORD_FILE) for fname in to_copy: if not ssh.CopyFileToNode(node, fname): logger.Error("could not copy file %s to node %s" % (fname, node)) @@ -1618,8 +1662,8 @@ class LUMasterFailover(LogicalUnit): if self.old_master == self.new_master: raise errors.OpPrereqError("This commands must be run on the node" - " where you want the new master to be.\n" - "%s is already the master" % + " where you want the new master to be." + " %s is already the master" % self.old_master) def Exec(self, feedback_fn): @@ -1648,8 +1692,8 @@ class LUMasterFailover(LogicalUnit): if not rpc.call_node_start_master(self.new_master): logger.Error("could not start the master role on the new master" " %s, please check" % self.new_master) - feedback_fn("Error in activating the master IP on the new master,\n" - "please fix manually.") + feedback_fn("Error in activating the master IP on the new master," + " please fix manually.") @@ -1816,22 +1860,41 @@ def _AssembleInstanceDisks(instance, cfg, ignore_secondaries=False): """ device_info = [] disks_ok = True + iname = instance.name + # With the two passes mechanism we try to reduce the window of + # opportunity for the race condition of switching DRBD to primary + # before handshaking occured, but we do not eliminate it + + # The proper fix would be to wait (with some limits) until the + # connection has been made and drbd transitions from WFConnection + # into any other network-connected state (Connected, SyncTarget, + # SyncSource, etc.) + + # 1st pass, assemble on all nodes in secondary mode for inst_disk in instance.disks: - master_result = None for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node): cfg.SetDiskID(node_disk, node) - is_primary = node == instance.primary_node - result = rpc.call_blockdev_assemble(node, node_disk, - instance.name, is_primary) + result = rpc.call_blockdev_assemble(node, node_disk, iname, False) if not result: - logger.Error("could not prepare block device %s on node %s (is_pri" - "mary=%s)" % (inst_disk.iv_name, node, is_primary)) - if is_primary or not ignore_secondaries: + logger.Error("could not prepare block device %s on node %s" + " (is_primary=False, pass=1)" % (inst_disk.iv_name, node)) + if not ignore_secondaries: disks_ok = False - if is_primary: - master_result = result - device_info.append((instance.primary_node, inst_disk.iv_name, - master_result)) + + # FIXME: race condition on drbd migration to primary + + # 2nd pass, do only the primary node + for inst_disk in instance.disks: + for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node): + if node != instance.primary_node: + continue + cfg.SetDiskID(node_disk, node) + result = rpc.call_blockdev_assemble(node, node_disk, iname, True) + if not result: + logger.Error("could not prepare block device %s on node %s" + " (is_primary=True, pass=2)" % (inst_disk.iv_name, node)) + disks_ok = False + device_info.append((instance.primary_node, inst_disk.iv_name, result)) # leave the disks configured for the primary node # this is a workaround that would be fixed better by @@ -1914,6 +1977,36 @@ def _ShutdownInstanceDisks(instance, cfg, ignore_primary=False): return result +def _CheckNodeFreeMemory(cfg, node, reason, requested): + """Checks if a node has enough free memory. + + This function check if a given node has the needed amount of free + memory. In case the node has less memory or we cannot get the + information from the node, this function raise an OpPrereqError + exception. + + Args: + - cfg: a ConfigWriter instance + - node: the node name + - reason: string to use in the error message + - requested: the amount of memory in MiB + + """ + nodeinfo = rpc.call_node_info([node], cfg.GetVGName()) + if not nodeinfo or not isinstance(nodeinfo, dict): + raise errors.OpPrereqError("Could not contact node %s for resource" + " information" % (node,)) + + free_mem = nodeinfo[node].get('memory_free') + if not isinstance(free_mem, int): + raise errors.OpPrereqError("Can't compute free memory on node %s, result" + " was '%s'" % (node, free_mem)) + if requested > free_mem: + raise errors.OpPrereqError("Not enough memory on node %s for %s:" + " needed %s MiB, available %s MiB" % + (node, reason, requested, free_mem)) + + class LUStartupInstance(LogicalUnit): """Starts an instance. @@ -1951,6 +2044,10 @@ class LUStartupInstance(LogicalUnit): # check bridges existance _CheckInstanceBridgesExist(instance) + _CheckNodeFreeMemory(self.cfg, instance.primary_node, + "starting instance %s" % instance.name, + instance.memory) + self.instance = instance self.op.instance_name = instance.name @@ -1964,20 +2061,6 @@ class LUStartupInstance(LogicalUnit): node_current = instance.primary_node - nodeinfo = rpc.call_node_info([node_current], self.cfg.GetVGName()) - if not nodeinfo: - raise errors.OpExecError("Could not contact node %s for infos" % - (node_current)) - - freememory = nodeinfo[node_current]['memory_free'] - memory = instance.memory - if memory > freememory: - raise errors.OpExecError("Not enough memory to start instance" - " %s on node %s" - " needed %s MiB, available %s MiB" % - (instance.name, node_current, memory, - freememory)) - _StartInstanceDisks(self.cfg, instance, force) if not rpc.call_instance_start(node_current, instance, extra_args): @@ -2180,8 +2263,8 @@ class LUReinstallInstance(LogicalUnit): try: feedback_fn("Running the instance OS create scripts...") if not rpc.call_instance_os_add(inst.primary_node, inst, "sda", "sdb"): - raise errors.OpExecError("Could not install OS for instance %s " - "on node %s" % + raise errors.OpExecError("Could not install OS for instance %s" + " on node %s" % (inst.name, inst.primary_node)) finally: _ShutdownInstanceDisks(inst, self.cfg) @@ -2256,9 +2339,8 @@ class LURenameInstance(LogicalUnit): try: if not rpc.call_instance_run_rename(inst.primary_node, inst, old_name, "sda", "sdb"): - msg = ("Could run OS rename script for instance %s\n" - "on node %s\n" - "(but the instance has been renamed in Ganeti)" % + msg = ("Could run OS rename script for instance %s on node %s (but the" + " instance has been renamed in Ganeti)" % (inst.name, inst.primary_node)) logger.Error(msg) finally: @@ -2336,11 +2418,11 @@ class LUQueryInstances(NoHooksLU): This checks that the fields required are valid output fields. """ - self.dynamic_fields = frozenset(["oper_state", "oper_ram"]) + self.dynamic_fields = frozenset(["oper_state", "oper_ram", "status"]) _CheckOutputFields(static=["name", "os", "pnode", "snodes", "admin_state", "admin_ram", "disk_template", "ip", "mac", "bridge", - "sda_size", "sdb_size"], + "sda_size", "sdb_size", "vcpus"], dynamic=self.dynamic_fields, selected=self.op.output_fields) @@ -2393,6 +2475,21 @@ class LUQueryInstances(NoHooksLU): val = None else: val = bool(live_data.get(instance.name)) + elif field == "status": + if instance.primary_node in bad_nodes: + val = "ERROR_nodedown" + else: + running = bool(live_data.get(instance.name)) + if running: + if instance.status != "down": + val = "running" + else: + val = "ERROR_up" + else: + if instance.status != "down": + val = "ERROR_down" + else: + val = "ADMIN_down" elif field == "admin_ram": val = instance.memory elif field == "oper_ram": @@ -2416,6 +2513,8 @@ class LUQueryInstances(NoHooksLU): val = None else: val = disk.size + elif field == "vcpus": + val = instance.vcpus else: raise errors.ParameterError(field) iout.append(val) @@ -2466,18 +2565,10 @@ class LUFailoverInstance(LogicalUnit): raise errors.ProgrammerError("no secondary node but using " "DT_REMOTE_RAID1 template") - # check memory requirements on the secondary node target_node = secondary_nodes[0] - nodeinfo = rpc.call_node_info([target_node], self.cfg.GetVGName()) - info = nodeinfo.get(target_node, None) - if not info: - raise errors.OpPrereqError("Cannot get current information" - " from node '%s'" % nodeinfo) - if instance.memory > info['memory_free']: - raise errors.OpPrereqError("Not enough memory on target node %s." - " %d MB available, %d MB required" % - (target_node, info['memory_free'], - instance.memory)) + # check memory requirements on the secondary node + _CheckNodeFreeMemory(self.cfg, target_node, "failing over instance %s" % + instance.name, instance.memory) # check bridge existance brlist = [nic.bridge for nic in instance.nics] @@ -2508,21 +2599,6 @@ class LUFailoverInstance(LogicalUnit): raise errors.OpExecError("Disk %s is degraded on target node," " aborting failover." % dev.iv_name) - feedback_fn("* checking target node resource availability") - nodeinfo = rpc.call_node_info([target_node], self.cfg.GetVGName()) - - if not nodeinfo: - raise errors.OpExecError("Could not contact target node %s." % - target_node) - - free_memory = int(nodeinfo[target_node]['memory_free']) - memory = instance.memory - if memory > free_memory: - raise errors.OpExecError("Not enough memory to create instance %s on" - " node %s. needed %s MiB, available %s MiB" % - (instance.name, target_node, memory, - free_memory)) - feedback_fn("* shutting down instance on source node") logger.Info("Shutting down instance %s on node %s" % (instance.name, source_node)) @@ -2665,9 +2741,9 @@ def _GenerateDiskTemplate(cfg, template_name, #TODO: compute space requirements vgname = cfg.GetVGName() - if template_name == "diskless": + if template_name == constants.DT_DISKLESS: disks = [] - elif template_name == "plain": + elif template_name == constants.DT_PLAIN: if len(secondary_nodes) != 0: raise errors.ProgrammerError("Wrong template configuration") @@ -2679,7 +2755,7 @@ def _GenerateDiskTemplate(cfg, template_name, logical_id=(vgname, names[1]), iv_name = "sdb") disks = [sda_dev, sdb_dev] - elif template_name == "local_raid1": + elif template_name == constants.DT_LOCAL_RAID1: if len(secondary_nodes) != 0: raise errors.ProgrammerError("Wrong template configuration") @@ -2809,7 +2885,7 @@ class LUCreateInstance(LogicalUnit): HTYPE = constants.HTYPE_INSTANCE _OP_REQP = ["instance_name", "mem_size", "disk_size", "pnode", "disk_template", "swap_size", "mode", "start", "vcpus", - "wait_for_sync", "ip_check"] + "wait_for_sync", "ip_check", "mac"] def BuildHooksEnv(self): """Build hooks env. @@ -2835,7 +2911,7 @@ class LUCreateInstance(LogicalUnit): os_type=self.op.os_type, memory=self.op.mem_size, vcpus=self.op.vcpus, - nics=[(self.inst_ip, self.op.bridge)], + nics=[(self.inst_ip, self.op.bridge, self.op.mac)], )) nl = ([self.sstore.GetMasterNode(), self.op.pnode] + @@ -2847,6 +2923,10 @@ class LUCreateInstance(LogicalUnit): """Check prerequisites. """ + for attr in ["kernel_path", "initrd_path", "hvm_boot_order"]: + if not hasattr(self.op, attr): + setattr(self.op, attr, None) + if self.op.mode not in (constants.INSTANCE_CREATE, constants.INSTANCE_IMPORT): raise errors.OpPrereqError("Invalid instance creation mode '%s'" % @@ -2918,13 +2998,9 @@ class LUCreateInstance(LogicalUnit): " the primary node.") self.secondaries.append(snode_name) - # Check lv size requirements - nodenames = [pnode.name] + self.secondaries - nodeinfo = rpc.call_node_info(nodenames, self.cfg.GetVGName()) - # Required free disk space as a function of disk and swap space req_size_dict = { - constants.DT_DISKLESS: 0, + constants.DT_DISKLESS: None, constants.DT_PLAIN: self.op.disk_size + self.op.swap_size, constants.DT_LOCAL_RAID1: (self.op.disk_size + self.op.swap_size) * 2, # 256 MB are added for drbd metadata, 128MB for each drbd device @@ -2938,15 +3014,23 @@ class LUCreateInstance(LogicalUnit): req_size = req_size_dict[self.op.disk_template] - for node in nodenames: - info = nodeinfo.get(node, None) - if not info: - raise errors.OpPrereqError("Cannot get current information" - " from node '%s'" % nodeinfo) - if req_size > info['vg_free']: - raise errors.OpPrereqError("Not enough disk space on target node %s." - " %d MB available, %d MB required" % - (node, info['vg_free'], req_size)) + # Check lv size requirements + if req_size is not None: + nodenames = [pnode.name] + self.secondaries + nodeinfo = rpc.call_node_info(nodenames, self.cfg.GetVGName()) + for node in nodenames: + info = nodeinfo.get(node, None) + if not info: + raise errors.OpPrereqError("Cannot get current information" + " from node '%s'" % nodeinfo) + vg_free = info.get('vg_free', None) + if not isinstance(vg_free, int): + raise errors.OpPrereqError("Can't compute free disk space on" + " node %s" % node) + if req_size > info['vg_free']: + raise errors.OpPrereqError("Not enough disk space on target node %s." + " %d MB available, %d MB required" % + (node, info['vg_free'], req_size)) # os verification os_obj = rpc.call_os_get(pnode.name, self.op.os_type) @@ -2954,6 +3038,9 @@ class LUCreateInstance(LogicalUnit): raise errors.OpPrereqError("OS '%s' not in supported os list for" " primary node" % self.op.os_type) + if self.op.kernel_path == constants.VALUE_NONE: + raise errors.OpPrereqError("Can't set instance kernel to none") + # instance verification hostname1 = utils.HostInfo(self.op.instance_name) @@ -2985,6 +3072,12 @@ class LUCreateInstance(LogicalUnit): raise errors.OpPrereqError("IP %s of instance %s already in use" % (hostname1.ip, instance_name)) + # MAC address verification + if self.op.mac != "auto": + if not utils.IsValidMac(self.op.mac.lower()): + raise errors.OpPrereqError("invalid MAC address specified: %s" % + self.op.mac) + # bridge verification bridge = getattr(self.op, "bridge", None) if bridge is None: @@ -2997,6 +3090,12 @@ class LUCreateInstance(LogicalUnit): " destination node '%s'" % (self.op.bridge, pnode.name)) + # boot order verification + if self.op.hvm_boot_order is not None: + if len(self.op.hvm_boot_order.strip("acdn")) != 0: + raise errors.OpPrereqError("invalid boot order specified," + " must be one or more of [acdn]") + if self.op.start: self.instance_status = 'up' else: @@ -3009,10 +3108,21 @@ class LUCreateInstance(LogicalUnit): instance = self.op.instance_name pnode_name = self.pnode.name - nic = objects.NIC(bridge=self.op.bridge, mac=self.cfg.GenerateMAC()) + if self.op.mac == "auto": + mac_address = self.cfg.GenerateMAC() + else: + mac_address = self.op.mac + + nic = objects.NIC(bridge=self.op.bridge, mac=mac_address) if self.inst_ip is not None: nic.ip = self.inst_ip + ht_kind = self.sstore.GetHypervisorType() + if ht_kind in constants.HTS_REQ_PORT: + network_port = self.cfg.AllocatePort() + else: + network_port = None + disks = _GenerateDiskTemplate(self.cfg, self.op.disk_template, instance, pnode_name, @@ -3026,6 +3136,10 @@ class LUCreateInstance(LogicalUnit): nics=[nic], disks=disks, disk_template=self.op.disk_template, status=self.instance_status, + network_port=network_port, + kernel_path=self.op.kernel_path, + initrd_path=self.op.initrd_path, + hvm_boot_order=self.op.hvm_boot_order, ) feedback_fn("* creating instance disks...") @@ -3038,12 +3152,12 @@ class LUCreateInstance(LogicalUnit): self.cfg.AddInstance(iobj) if self.op.wait_for_sync: - disk_abort = not _WaitForSync(self.cfg, iobj) + disk_abort = not _WaitForSync(self.cfg, iobj, self.proc) elif iobj.disk_template in constants.DTS_NET_MIRROR: # make sure the disks are not degraded (still sync-ing is ok) time.sleep(15) feedback_fn("* checking mirrors status") - disk_abort = not _WaitForSync(self.cfg, iobj, oneshot=True) + disk_abort = not _WaitForSync(self.cfg, iobj, self.proc, oneshot=True) else: disk_abort = False @@ -3125,7 +3239,7 @@ class LUConnectConsole(NoHooksLU): logger.Debug("connecting to console of %s on %s" % (instance.name, node)) hyper = hypervisor.GetHypervisor() - console_cmd = hyper.GetShellCommandForConsole(instance.name) + console_cmd = hyper.GetShellCommandForConsole(instance) # build ssh cmdline argv = ["ssh", "-q", "-t"] argv.extend(ssh.KNOWN_HOSTS_OPTS) @@ -3190,10 +3304,9 @@ class LUAddMDDRBDComponent(LogicalUnit): raise errors.OpPrereqError("Can't find this device ('%s') in the" " instance." % self.op.disk_name) if len(disk.children) > 1: - raise errors.OpPrereqError("The device already has two slave" - " devices.\n" - "This would create a 3-disk raid1" - " which we don't allow.") + raise errors.OpPrereqError("The device already has two slave devices." + " This would create a 3-disk raid1 which we" + " don't allow.") self.disk = disk def Exec(self, feedback_fn): @@ -3245,7 +3358,7 @@ class LUAddMDDRBDComponent(LogicalUnit): self.cfg.AddInstance(instance) - _WaitForSync(self.cfg, instance) + _WaitForSync(self.cfg, instance, self.proc) return 0 @@ -3357,8 +3470,12 @@ class LUReplaceDisks(LogicalUnit): "OLD_SECONDARY": self.instance.secondary_nodes[0], } env.update(_BuildInstanceHookEnvByObject(self.instance)) - nl = [self.sstore.GetMasterNode(), - self.instance.primary_node] + list(self.instance.secondary_nodes) + nl = [ + self.sstore.GetMasterNode(), + self.instance.primary_node, + ] + if self.op.remote_node is not None: + nl.append(self.op.remote_node) return env, nl, nl def CheckPrereq(self): @@ -3373,6 +3490,7 @@ class LUReplaceDisks(LogicalUnit): raise errors.OpPrereqError("Instance '%s' not known" % self.op.instance_name) self.instance = instance + self.op.instance_name = instance.name if instance.disk_template not in constants.DTS_NET_MIRROR: raise errors.OpPrereqError("Instance's disk layout is not" @@ -3398,28 +3516,40 @@ class LUReplaceDisks(LogicalUnit): raise errors.OpPrereqError("The specified node is the primary node of" " the instance.") elif remote_node == self.sec_node: + if self.op.mode == constants.REPLACE_DISK_SEC: + # this is for DRBD8, where we can't execute the same mode of + # replacement as for drbd7 (no different port allocated) + raise errors.OpPrereqError("Same secondary given, cannot execute" + " replacement") # the user gave the current secondary, switch to - # 'no-replace-secondary' mode + # 'no-replace-secondary' mode for drbd7 remote_node = None if (instance.disk_template == constants.DT_REMOTE_RAID1 and self.op.mode != constants.REPLACE_DISK_ALL): raise errors.OpPrereqError("Template 'remote_raid1' only allows all" " disks replacement, not individual ones") if instance.disk_template == constants.DT_DRBD8: + if (self.op.mode == constants.REPLACE_DISK_ALL and + remote_node is not None): + # switch to replace secondary mode + self.op.mode = constants.REPLACE_DISK_SEC + if self.op.mode == constants.REPLACE_DISK_ALL: - raise errors.OpPrereqError("Template 'drbd8' only allows primary or" + raise errors.OpPrereqError("Template 'drbd' only allows primary or" " secondary disk replacement, not" " both at once") elif self.op.mode == constants.REPLACE_DISK_PRI: if remote_node is not None: - raise errors.OpPrereqError("Template 'drbd8' does not allow changing" + raise errors.OpPrereqError("Template 'drbd' does not allow changing" " the secondary while doing a primary" " node disk replacement") self.tgt_node = instance.primary_node + self.oth_node = instance.secondary_nodes[0] elif self.op.mode == constants.REPLACE_DISK_SEC: self.new_node = remote_node # this can be None, in which case # we don't change the secondary self.tgt_node = instance.secondary_nodes[0] + self.oth_node = instance.primary_node else: raise errors.ProgrammerError("Unhandled disk replace mode") @@ -3454,9 +3584,8 @@ class LUReplaceDisks(LogicalUnit): if not _CreateBlockDevOnSecondary(cfg, remote_node, instance, new_drbd, False, _GetInstanceInfoText(instance)): - raise errors.OpExecError("Failed to create new component on" - " secondary node %s\n" - "Full abort, cleanup manually!" % + raise errors.OpExecError("Failed to create new component on secondary" + " node %s. Full abort, cleanup manually!" % remote_node) logger.Info("adding new mirror component on primary") @@ -3467,8 +3596,8 @@ class LUReplaceDisks(LogicalUnit): # remove secondary dev cfg.SetDiskID(new_drbd, remote_node) rpc.call_blockdev_remove(remote_node, new_drbd) - raise errors.OpExecError("Failed to create volume on primary!\n" - "Full abort, cleanup manually!!") + raise errors.OpExecError("Failed to create volume on primary!" + " Full abort, cleanup manually!!") # the device exists now # call the primary node to add the mirror to md @@ -3490,7 +3619,7 @@ class LUReplaceDisks(LogicalUnit): # this can fail as the old devices are degraded and _WaitForSync # does a combined result over all disks, so we don't check its # return value - _WaitForSync(cfg, instance, unlock=True) + _WaitForSync(cfg, instance, self.proc, unlock=True) # so check manually all the devices for name in iv_names: @@ -3540,13 +3669,54 @@ class LUReplaceDisks(LogicalUnit): - remove old LVs (which have the name name_replaces.) Failures are not very well handled. + """ + steps_total = 6 + warning, info = (self.proc.LogWarning, self.proc.LogInfo) instance = self.instance iv_names = {} vgname = self.cfg.GetVGName() # start of work cfg = self.cfg tgt_node = self.tgt_node + oth_node = self.oth_node + + # Step: check device activation + self.proc.LogStep(1, steps_total, "check device existence") + info("checking volume groups") + my_vg = cfg.GetVGName() + results = rpc.call_vg_list([oth_node, tgt_node]) + if not results: + raise errors.OpExecError("Can't list volume groups on the nodes") + for node in oth_node, tgt_node: + res = results.get(node, False) + if not res or my_vg not in res: + raise errors.OpExecError("Volume group '%s' not found on %s" % + (my_vg, node)) + for dev in instance.disks: + if not dev.iv_name in self.op.disks: + continue + for node in tgt_node, oth_node: + info("checking %s on %s" % (dev.iv_name, node)) + cfg.SetDiskID(dev, node) + if not rpc.call_blockdev_find(node, dev): + raise errors.OpExecError("Can't find device %s on node %s" % + (dev.iv_name, node)) + + # Step: check other node consistency + self.proc.LogStep(2, steps_total, "check peer consistency") + for dev in instance.disks: + if not dev.iv_name in self.op.disks: + continue + info("checking %s consistency on %s" % (dev.iv_name, oth_node)) + if not _CheckDiskConsistency(self.cfg, dev, oth_node, + oth_node==instance.primary_node): + raise errors.OpExecError("Peer node (%s) has degraded storage, unsafe" + " to replace disks on this node (%s)" % + (oth_node, tgt_node)) + + # Step: create new storage + self.proc.LogStep(3, steps_total, "allocate new storage") for dev in instance.disks: if not dev.iv_name in self.op.disks: continue @@ -3561,8 +3731,8 @@ class LUReplaceDisks(LogicalUnit): new_lvs = [lv_data, lv_meta] old_lvs = dev.children iv_names[dev.iv_name] = (dev, old_lvs, new_lvs) - logger.Info("adding new local storage on %s for %s" % - (tgt_node, dev.iv_name)) + info("creating new local storage on %s for %s" % + (tgt_node, dev.iv_name)) # since we *always* want to create this LV, we use the # _Create...OnPrimary (which forces the creation), even if we # are talking about the secondary node @@ -3573,60 +3743,69 @@ class LUReplaceDisks(LogicalUnit): " node '%s'" % (new_lv.logical_id[1], tgt_node)) + # Step: for each lv, detach+rename*2+attach + self.proc.LogStep(4, steps_total, "change drbd configuration") + for dev, old_lvs, new_lvs in iv_names.itervalues(): + info("detaching %s drbd from local storage" % dev.iv_name) if not rpc.call_blockdev_removechildren(tgt_node, dev, old_lvs): raise errors.OpExecError("Can't detach drbd from local storage on node" " %s for device %s" % (tgt_node, dev.iv_name)) - dev.children = [] - cfg.Update(instance) + #dev.children = [] + #cfg.Update(instance) # ok, we created the new LVs, so now we know we have the needed # storage; as such, we proceed on the target node to rename # old_lv to _old, and new_lv to old_lv; note that we rename LVs - # using the assumption than logical_id == physical_id (which in + # using the assumption that logical_id == physical_id (which in # turn is the unique_id on that node) + + # FIXME(iustin): use a better name for the replaced LVs temp_suffix = int(time.time()) - logger.Info("renaming the old LVs on the target node") ren_fn = lambda d, suff: (d.physical_id[0], d.physical_id[1] + "_replaced-%s" % suff) - rlist = [(disk, ren_fn(disk, temp_suffix)) for disk in old_lvs] + # build the rename list based on what LVs exist on the node + rlist = [] + for to_ren in old_lvs: + find_res = rpc.call_blockdev_find(tgt_node, to_ren) + if find_res is not None: # device exists + rlist.append((to_ren, ren_fn(to_ren, temp_suffix))) + + info("renaming the old LVs on the target node") if not rpc.call_blockdev_rename(tgt_node, rlist): - logger.Error("Can't rename old LVs on node %s" % tgt_node) - do_change_old = False - else: - do_change_old = True + raise errors.OpExecError("Can't rename old LVs on node %s" % tgt_node) # now we rename the new LVs to the old LVs - logger.Info("renaming the new LVs on the target node") + info("renaming the new LVs on the target node") rlist = [(new, old.physical_id) for old, new in zip(old_lvs, new_lvs)] if not rpc.call_blockdev_rename(tgt_node, rlist): - logger.Error("Can't rename new LVs on node %s" % tgt_node) - else: - for old, new in zip(old_lvs, new_lvs): - new.logical_id = old.logical_id - cfg.SetDiskID(new, tgt_node) + raise errors.OpExecError("Can't rename new LVs on node %s" % tgt_node) - if do_change_old: - for disk in old_lvs: - disk.logical_id = ren_fn(disk, temp_suffix) - cfg.SetDiskID(disk, tgt_node) + for old, new in zip(old_lvs, new_lvs): + new.logical_id = old.logical_id + cfg.SetDiskID(new, tgt_node) + + for disk in old_lvs: + disk.logical_id = ren_fn(disk, temp_suffix) + cfg.SetDiskID(disk, tgt_node) # now that the new lvs have the old name, we can add them to the device - logger.Info("adding new mirror component on %s" % tgt_node) + info("adding new mirror component on %s" % tgt_node) if not rpc.call_blockdev_addchildren(tgt_node, dev, new_lvs): - logger.Error("Can't add local storage to drbd!") for new_lv in new_lvs: if not rpc.call_blockdev_remove(tgt_node, new_lv): - logger.Error("Can't rollback device %s") - return + warning("Can't rollback device %s", hint="manually cleanup unused" + " logical volumes") + raise errors.OpExecError("Can't add local storage to drbd") dev.children = new_lvs cfg.Update(instance) + # Step: wait for sync # this can fail as the old devices are degraded and _WaitForSync # does a combined result over all disks, so we don't check its # return value - logger.Info("Done changing drbd configs, waiting for sync") - _WaitForSync(cfg, instance, unlock=True) + self.proc.LogStep(5, steps_total, "sync devices") + _WaitForSync(cfg, instance, self.proc, unlock=True) # so check manually all the devices for name, (dev, old_lvs, new_lvs) in iv_names.iteritems(): @@ -3635,13 +3814,14 @@ class LUReplaceDisks(LogicalUnit): if is_degr: raise errors.OpExecError("DRBD device %s is degraded!" % name) + # Step: remove old storage + self.proc.LogStep(6, steps_total, "removing old storage") for name, (dev, old_lvs, new_lvs) in iv_names.iteritems(): - logger.Info("remove logical volumes for %s" % name) + info("remove logical volumes for %s" % name) for lv in old_lvs: cfg.SetDiskID(lv, tgt_node) if not rpc.call_blockdev_remove(tgt_node, lv): - logger.Error("Can't cleanup child device, skipping. You need to" - " fix manually!") + warning("Can't remove old LV", hint="manually remove unused LVs") continue def _ExecD8Secondary(self, feedback_fn): @@ -3661,7 +3841,10 @@ class LUReplaceDisks(LogicalUnit): - remove all disks from the old secondary Failures are not very well handled. + """ + steps_total = 6 + warning, info = (self.proc.LogWarning, self.proc.LogInfo) instance = self.instance iv_names = {} vgname = self.cfg.GetVGName() @@ -3670,10 +3853,44 @@ class LUReplaceDisks(LogicalUnit): old_node = self.tgt_node new_node = self.new_node pri_node = instance.primary_node + + # Step: check device activation + self.proc.LogStep(1, steps_total, "check device existence") + info("checking volume groups") + my_vg = cfg.GetVGName() + results = rpc.call_vg_list([pri_node, new_node]) + if not results: + raise errors.OpExecError("Can't list volume groups on the nodes") + for node in pri_node, new_node: + res = results.get(node, False) + if not res or my_vg not in res: + raise errors.OpExecError("Volume group '%s' not found on %s" % + (my_vg, node)) + for dev in instance.disks: + if not dev.iv_name in self.op.disks: + continue + info("checking %s on %s" % (dev.iv_name, pri_node)) + cfg.SetDiskID(dev, pri_node) + if not rpc.call_blockdev_find(pri_node, dev): + raise errors.OpExecError("Can't find device %s on node %s" % + (dev.iv_name, pri_node)) + + # Step: check other node consistency + self.proc.LogStep(2, steps_total, "check peer consistency") + for dev in instance.disks: + if not dev.iv_name in self.op.disks: + continue + info("checking %s consistency on %s" % (dev.iv_name, pri_node)) + if not _CheckDiskConsistency(self.cfg, dev, pri_node, True, ldisk=True): + raise errors.OpExecError("Primary node (%s) has degraded storage," + " unsafe to replace the secondary" % + pri_node) + + # Step: create new storage + self.proc.LogStep(3, steps_total, "allocate new storage") for dev in instance.disks: size = dev.size - logger.Info("adding new local storage on %s for %s" % - (new_node, dev.iv_name)) + info("adding new local storage on %s for %s" % (new_node, dev.iv_name)) # since we *always* want to create this LV, we use the # _Create...OnPrimary (which forces the creation), even if we # are talking about the secondary node @@ -3684,6 +3901,12 @@ class LUReplaceDisks(LogicalUnit): " node '%s'" % (new_lv.logical_id[1], new_node)) + iv_names[dev.iv_name] = (dev, dev.children) + + self.proc.LogStep(4, steps_total, "changing drbd configuration") + for dev in instance.disks: + size = dev.size + info("activating a new drbd on %s for %s" % (new_node, dev.iv_name)) # create new devices on new_node new_drbd = objects.Disk(dev_type=constants.LD_DRBD8, logical_id=(pri_node, new_node, @@ -3695,32 +3918,59 @@ class LUReplaceDisks(LogicalUnit): raise errors.OpExecError("Failed to create new DRBD on" " node '%s'" % new_node) + for dev in instance.disks: # we have new devices, shutdown the drbd on the old secondary + info("shutting down drbd for %s on old node" % dev.iv_name) cfg.SetDiskID(dev, old_node) if not rpc.call_blockdev_shutdown(old_node, dev): - raise errors.OpExecError("Failed to shutdown DRBD on old node") + warning("Failed to shutdown drbd for %s on old node" % dev.iv_name, + hint="Please cleanup this device manually as soon as possible") - # we have new storage, we 'rename' the network on the primary + info("detaching primary drbds from the network (=> standalone)") + done = 0 + for dev in instance.disks: cfg.SetDiskID(dev, pri_node) - # rename to the ip of the new node - new_uid = list(dev.physical_id) - new_uid[2] = self.remote_node_info.secondary_ip - rlist = [(dev, tuple(new_uid))] - if not rpc.call_blockdev_rename(pri_node, rlist): - raise errors.OpExecError("Can't detach re-attach drbd %s on node" - " %s from %s to %s" % - (dev.iv_name, pri_node, old_node, new_node)) - dev.logical_id = (pri_node, new_node, dev.logical_id[2]) + # set the physical (unique in bdev terms) id to None, meaning + # detach from network + dev.physical_id = (None,) * len(dev.physical_id) + # and 'find' the device, which will 'fix' it to match the + # standalone state + if rpc.call_blockdev_find(pri_node, dev): + done += 1 + else: + warning("Failed to detach drbd %s from network, unusual case" % + dev.iv_name) + + if not done: + # no detaches succeeded (very unlikely) + raise errors.OpExecError("Can't detach at least one DRBD from old node") + + # if we managed to detach at least one, we update all the disks of + # the instance to point to the new secondary + info("updating instance configuration") + for dev in instance.disks: + dev.logical_id = (pri_node, new_node) + dev.logical_id[2:] cfg.SetDiskID(dev, pri_node) - cfg.Update(instance) + cfg.Update(instance) - iv_names[dev.iv_name] = (dev, dev.children) + # and now perform the drbd attach + info("attaching primary drbds to new secondary (standalone => connected)") + failures = [] + for dev in instance.disks: + info("attaching primary drbd for %s to new secondary node" % dev.iv_name) + # since the attach is smart, it's enough to 'find' the device, + # it will automatically activate the network, if the physical_id + # is correct + cfg.SetDiskID(dev, pri_node) + if not rpc.call_blockdev_find(pri_node, dev): + warning("can't attach drbd %s to new secondary!" % dev.iv_name, + "please do a gnt-instance info to see the status of disks") # this can fail as the old devices are degraded and _WaitForSync # does a combined result over all disks, so we don't check its # return value - logger.Info("Done changing drbd configs, waiting for sync") - _WaitForSync(cfg, instance, unlock=True) + self.proc.LogStep(5, steps_total, "sync devices") + _WaitForSync(cfg, instance, self.proc, unlock=True) # so check manually all the devices for name, (dev, old_lvs) in iv_names.iteritems(): @@ -3729,14 +3979,14 @@ class LUReplaceDisks(LogicalUnit): if is_degr: raise errors.OpExecError("DRBD device %s is degraded!" % name) + self.proc.LogStep(6, steps_total, "removing old storage") for name, (dev, old_lvs) in iv_names.iteritems(): - logger.Info("remove logical volumes for %s" % name) + info("remove logical volumes for %s" % name) for lv in old_lvs: cfg.SetDiskID(lv, old_node) if not rpc.call_blockdev_remove(old_node, lv): - logger.Error("Can't cleanup child device, skipping. You need to" - " fix manually!") - continue + warning("Can't remove LV on old secondary", + hint="Cleanup stale volumes by hand") def Exec(self, feedback_fn): """Execute disk replacement. @@ -3778,7 +4028,7 @@ class LUQueryInstanceData(NoHooksLU): instance = self.cfg.GetInstanceInfo(self.cfg.ExpandInstanceName(name)) if instance is None: raise errors.OpPrereqError("No such instance name '%s'" % name) - self.wanted_instances.append(instance) + self.wanted_instances.append(instance) else: self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name in self.cfg.GetInstanceList()] @@ -3850,7 +4100,11 @@ class LUQueryInstanceData(NoHooksLU): "memory": instance.memory, "nics": [(nic.mac, nic.ip, nic.bridge) for nic in instance.nics], "disks": disks, + "network_port": instance.network_port, "vcpus": instance.vcpus, + "kernel_path": instance.kernel_path, + "initrd_path": instance.initrd_path, + "hvm_boot_order": instance.hvm_boot_order, } result[instance.name] = idict @@ -3877,7 +4131,7 @@ class LUSetInstanceParms(LogicalUnit): args['memory'] = self.mem if self.vcpus: args['vcpus'] = self.vcpus - if self.do_ip or self.do_bridge: + if self.do_ip or self.do_bridge or self.mac: if self.do_ip: ip = self.ip else: @@ -3886,7 +4140,11 @@ class LUSetInstanceParms(LogicalUnit): bridge = self.bridge else: bridge = self.instance.nics[0].bridge - args['nics'] = [(ip, bridge)] + if self.mac: + mac = self.mac + else: + mac = self.instance.nics[0].mac + args['nics'] = [(ip, bridge, mac)] env = _BuildInstanceHookEnvByObject(self.instance, override=args) nl = [self.sstore.GetMasterNode(), self.instance.primary_node] + list(self.instance.secondary_nodes) @@ -3901,8 +4159,14 @@ class LUSetInstanceParms(LogicalUnit): self.mem = getattr(self.op, "mem", None) self.vcpus = getattr(self.op, "vcpus", None) self.ip = getattr(self.op, "ip", None) + self.mac = getattr(self.op, "mac", None) self.bridge = getattr(self.op, "bridge", None) - if [self.mem, self.vcpus, self.ip, self.bridge].count(None) == 4: + self.kernel_path = getattr(self.op, "kernel_path", None) + self.initrd_path = getattr(self.op, "initrd_path", None) + self.hvm_boot_order = getattr(self.op, "hvm_boot_order", None) + all_parms = [self.mem, self.vcpus, self.ip, self.bridge, self.mac, + self.kernel_path, self.initrd_path, self.hvm_boot_order] + if all_parms.count(None) == len(all_parms): raise errors.OpPrereqError("No changes submitted") if self.mem is not None: try: @@ -3924,6 +4188,42 @@ class LUSetInstanceParms(LogicalUnit): else: self.do_ip = False self.do_bridge = (self.bridge is not None) + if self.mac is not None: + if self.cfg.IsMacInUse(self.mac): + raise errors.OpPrereqError('MAC address %s already in use in cluster' % + self.mac) + if not utils.IsValidMac(self.mac): + raise errors.OpPrereqError('Invalid MAC address %s' % self.mac) + + if self.kernel_path is not None: + self.do_kernel_path = True + if self.kernel_path == constants.VALUE_NONE: + raise errors.OpPrereqError("Can't set instance to no kernel") + + if self.kernel_path != constants.VALUE_DEFAULT: + if not os.path.isabs(self.kernel_path): + raise errors.OpPrereqError("The kernel path must be an absolute" + " filename") + else: + self.do_kernel_path = False + + if self.initrd_path is not None: + self.do_initrd_path = True + if self.initrd_path not in (constants.VALUE_NONE, + constants.VALUE_DEFAULT): + if not os.path.isabs(self.initrd_path): + raise errors.OpPrereqError("The initrd path must be an absolute" + " filename") + else: + self.do_initrd_path = False + + # boot order verification + if self.hvm_boot_order is not None: + if self.hvm_boot_order != constants.VALUE_DEFAULT: + if len(self.hvm_boot_order.strip("acdn")) != 0: + raise errors.OpPrereqError("invalid boot order specified," + " must be one or more of [acdn]" + " or 'default'") instance = self.cfg.GetInstanceInfo( self.cfg.ExpandInstanceName(self.op.instance_name)) @@ -3953,6 +4253,21 @@ class LUSetInstanceParms(LogicalUnit): if self.bridge: instance.nics[0].bridge = self.bridge result.append(("bridge", self.bridge)) + if self.mac: + instance.nics[0].mac = self.mac + result.append(("mac", self.mac)) + if self.do_kernel_path: + instance.kernel_path = self.kernel_path + result.append(("kernel_path", self.kernel_path)) + if self.do_initrd_path: + instance.initrd_path = self.initrd_path + result.append(("initrd_path", self.initrd_path)) + if self.hvm_boot_order: + if self.hvm_boot_order == constants.VALUE_DEFAULT: + instance.hvm_boot_order = None + else: + instance.hvm_boot_order = self.hvm_boot_order + result.append(("hvm_boot_order", self.hvm_boot_order)) self.cfg.AddInstance(instance) @@ -4037,7 +4352,7 @@ class LUExportInstance(LogicalUnit): # shutdown the instance, unless requested not to do so if self.op.shutdown: op = opcodes.OpShutdownInstance(instance_name=instance.name) - self.processor.ChainOpCode(op) + self.proc.ChainOpCode(op) vgname = self.cfg.GetVGName() @@ -4063,7 +4378,7 @@ class LUExportInstance(LogicalUnit): if self.op.shutdown: op = opcodes.OpStartupInstance(instance_name=instance.name, force=False) - self.processor.ChainOpCode(op) + self.proc.ChainOpCode(op) # TODO: check for size @@ -4089,7 +4404,7 @@ class LUExportInstance(LogicalUnit): # substitutes an empty list with the full cluster node list. if nodelist: op = opcodes.OpQueryExports(nodes=nodelist) - exportlist = self.processor.ChainOpCode(op) + exportlist = self.proc.ChainOpCode(op) for node in exportlist: if instance.name in exportlist[node]: if not rpc.call_export_remove(node, instance.name): @@ -4246,3 +4561,39 @@ class LUDelTags(TagsLU): raise errors.OpRetryError("There has been a modification to the" " config file and the operation has been" " aborted. Please retry.") + +class LUTestDelay(NoHooksLU): + """Sleep for a specified amount of time. + + This LU sleeps on the master and/or nodes for a specified amoutn of + time. + + """ + _OP_REQP = ["duration", "on_master", "on_nodes"] + + def CheckPrereq(self): + """Check prerequisites. + + This checks that we have a good list of nodes and/or the duration + is valid. + + """ + + if self.op.on_nodes: + self.op.on_nodes = _GetWantedNodes(self, self.op.on_nodes) + + def Exec(self, feedback_fn): + """Do the actual sleep. + + """ + if self.op.on_master: + if not utils.TestDelay(self.op.duration): + raise errors.OpExecError("Error during master delay test") + if self.op.on_nodes: + result = rpc.call_test_delay(self.op.on_nodes, self.op.duration) + if not result: + raise errors.OpExecError("Complete failure from rpc call") + for node, node_result in result.items(): + if not node_result: + raise errors.OpExecError("Failure during rpc call to node %s," + " result: %s" % (node, node_result))