X-Git-Url: https://code.grnet.gr/git/ganeti-local/blobdiff_plain/43017d2635df9c82161ec6c469987eab967a1438..f12eadb345d80b51c0aa667b0e7cf62d97925448:/lib/cmdlib.py diff --git a/lib/cmdlib.py b/lib/cmdlib.py index ac4300a..9eb80cb 100644 --- a/lib/cmdlib.py +++ b/lib/cmdlib.py @@ -455,8 +455,8 @@ def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status, @param secondary_nodes: list of secondary nodes as strings @type os_type: string @param os_type: the name of the instance's OS - @type status: string - @param status: the desired status of the instances + @type status: boolean + @param status: the should_run status of the instance @type memory: string @param memory: the memory size of the instance @type vcpus: string @@ -468,13 +468,17 @@ def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status, @return: the hook environment for this instance """ + if status: + str_status = "up" + else: + str_status = "down" env = { "OP_TARGET": name, "INSTANCE_NAME": name, "INSTANCE_PRIMARY": primary_node, "INSTANCE_SECONDARIES": " ".join(secondary_nodes), "INSTANCE_OS_TYPE": os_type, - "INSTANCE_STATUS": status, + "INSTANCE_STATUS": str_status, "INSTANCE_MEMORY": memory, "INSTANCE_VCPUS": vcpus, } @@ -516,7 +520,7 @@ def _BuildInstanceHookEnvByObject(lu, instance, override=None): 'primary_node': instance.primary_node, 'secondary_nodes': instance.secondary_nodes, 'os_type': instance.os, - 'status': instance.os, + 'status': instance.admin_up, 'memory': bep[constants.BE_MEMORY], 'vcpus': bep[constants.BE_VCPUS], 'nics': [(nic.ip, nic.bridge, nic.mac) for nic in instance.nics], @@ -613,7 +617,8 @@ class LUVerifyCluster(LogicalUnit): self.share_locks = dict(((i, 1) for i in locking.LEVELS)) def _VerifyNode(self, nodeinfo, file_list, local_cksum, - node_result, feedback_fn, master_files): + node_result, feedback_fn, master_files, + drbd_map): """Run multiple tests against a node. Test list: @@ -630,6 +635,9 @@ class LUVerifyCluster(LogicalUnit): @param node_result: the results from the node @param feedback_fn: function used to accumulate results @param master_files: list of files that only masters should have + @param drbd_map: the useddrbd minors for this node, in + form of minor: (instance, must_exist) which correspond to instances + and their running status """ node = nodeinfo.name @@ -724,6 +732,19 @@ class LUVerifyCluster(LogicalUnit): if hv_result is not None: feedback_fn(" - ERROR: hypervisor %s verify failure: '%s'" % (hv_name, hv_result)) + + # check used drbd list + used_minors = node_result.get(constants.NV_DRBDLIST, []) + for minor, (iname, must_exist) in drbd_map.items(): + if minor not in used_minors and must_exist: + feedback_fn(" - ERROR: drbd minor %d of instance %s is not active" % + (minor, iname)) + bad = True + for minor in used_minors: + if minor not in drbd_map: + feedback_fn(" - ERROR: unallocated drbd minor %d is in use" % minor) + bad = True + return bad def _VerifyInstance(self, instance, instanceconfig, node_vol_is, @@ -751,7 +772,7 @@ class LUVerifyCluster(LogicalUnit): (volume, node)) bad = True - if not instanceconfig.status == 'down': + if instanceconfig.admin_up: if ((node_current not in node_instance or not instance in node_instance[node_current]) and node_current not in n_offline): @@ -867,6 +888,8 @@ class LUVerifyCluster(LogicalUnit): nodelist = utils.NiceSort(self.cfg.GetNodeList()) nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist] instancelist = utils.NiceSort(self.cfg.GetInstanceList()) + instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname)) + for iname in instancelist) i_non_redundant = [] # Non redundant instances i_non_a_balanced = [] # Non auto-balanced instances n_offline = [] # List of offline nodes @@ -881,6 +904,7 @@ class LUVerifyCluster(LogicalUnit): file_names = ssconf.SimpleStore().GetFileList() file_names.append(constants.SSL_CERT_FILE) + file_names.append(constants.RAPI_CERT_FILE) file_names.extend(master_files) local_checksums = utils.FingerprintFiles(file_names) @@ -888,21 +912,26 @@ class LUVerifyCluster(LogicalUnit): feedback_fn("* Gathering data (%d nodes)" % len(nodelist)) node_verify_param = { constants.NV_FILELIST: file_names, - constants.NV_NODELIST: nodelist, + constants.NV_NODELIST: [node.name for node in nodeinfo + if not node.offline], constants.NV_HYPERVISOR: hypervisors, constants.NV_NODENETTEST: [(node.name, node.primary_ip, - node.secondary_ip) for node in nodeinfo], + node.secondary_ip) for node in nodeinfo + if not node.offline], constants.NV_LVLIST: vg_name, constants.NV_INSTANCELIST: hypervisors, constants.NV_VGLIST: None, constants.NV_VERSION: None, constants.NV_HVINFO: self.cfg.GetHypervisorType(), + constants.NV_DRBDLIST: None, } all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param, self.cfg.GetClusterName()) cluster = self.cfg.GetClusterInfo() master_node = self.cfg.GetMasterNode() + all_drbd_map = self.cfg.ComputeDRBDMap() + for node_i in nodeinfo: node = node_i.name nresult = all_nvinfo[node].data @@ -925,8 +954,13 @@ class LUVerifyCluster(LogicalUnit): bad = True continue + node_drbd = {} + for minor, instance in all_drbd_map[node].items(): + instance = instanceinfo[instance] + node_drbd[minor] = (instance.name, instance.admin_up) result = self._VerifyNode(node_i, file_names, local_checksums, - nresult, feedback_fn, master_files) + nresult, feedback_fn, master_files, + node_drbd) bad = bad or result lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data") @@ -982,7 +1016,7 @@ class LUVerifyCluster(LogicalUnit): for instance in instancelist: feedback_fn("* Verifying instance %s" % instance) - inst_config = self.cfg.GetInstanceInfo(instance) + inst_config = instanceinfo[instance] result = self._VerifyInstance(instance, inst_config, node_volume, node_instance, feedback_fn, n_offline) bad = bad or result @@ -1151,7 +1185,7 @@ class LUVerifyDisks(NoHooksLU): nv_dict = {} for inst in instances: inst_lvs = {} - if (inst.status != "up" or + if (not inst.admin_up or inst.disk_template not in constants.DTS_NET_MIRROR): continue inst.MapLVsByNode(inst_lvs) @@ -1434,6 +1468,33 @@ class LUSetClusterParams(LogicalUnit): _AdjustCandidatePool(self) +class LURedistributeConfig(NoHooksLU): + """Force the redistribution of cluster configuration. + + This is a very simple LU. + + """ + _OP_REQP = [] + REQ_BGL = False + + def ExpandNames(self): + self.needed_locks = { + locking.LEVEL_NODE: locking.ALL_SET, + } + self.share_locks[locking.LEVEL_NODE] = 1 + + def CheckPrereq(self): + """Check prerequisites. + + """ + + def Exec(self, feedback_fn): + """Redistribute the configuration. + + """ + self.cfg.Update(self.cfg.GetClusterInfo()) + + def _WaitForSync(lu, instance, oneshot=False, unlock=False): """Sleep and poll for an instance's disk to sync. @@ -1465,8 +1526,7 @@ def _WaitForSync(lu, instance, oneshot=False, unlock=False): continue rstats = rstats.data retries = 0 - for i in range(len(rstats)): - mstat = rstats[i] + for i, mstat in enumerate(rstats): if mstat is None: lu.LogWarning("Can't compute data for node %s/%s", node, instance.disks[i].iv_name) @@ -1584,10 +1644,12 @@ class LUDiagnoseOS(NoHooksLU): """ node_list = self.acquired_locks[locking.LEVEL_NODE] - node_data = self.rpc.call_os_diagnose(node_list) + valid_nodes = [node for node in self.cfg.GetOnlineNodeList() + if node in node_list] + node_data = self.rpc.call_os_diagnose(valid_nodes) if node_data == False: raise errors.OpExecError("Can't gather the list of OSes") - pol = self._DiagnoseByOS(node_list, node_data) + pol = self._DiagnoseByOS(valid_nodes, node_data) output = [] for os_name, os_data in pol.iteritems(): row = [] @@ -1655,11 +1717,8 @@ class LURemoveNode(LogicalUnit): for instance_name in instance_list: instance = self.cfg.GetInstanceInfo(instance_name) - if node.name == instance.primary_node: - raise errors.OpPrereqError("Instance %s still running on the node," - " please remove first." % instance_name) - if node.name in instance.secondary_nodes: - raise errors.OpPrereqError("Instance %s has node as a secondary," + if node.name in instance.all_nodes: + raise errors.OpPrereqError("Instance %s is still running on the node," " please remove first." % instance_name) self.op.node_name = node.name self.node = node @@ -2401,7 +2460,7 @@ def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False): " (is_primary=True, pass=2)", inst_disk.iv_name, node) disks_ok = False - device_info.append((instance.primary_node, inst_disk.iv_name, result)) + device_info.append((instance.primary_node, inst_disk.iv_name, result.data)) # leave the disks configured for the primary node # this is a workaround that would be fixed better by @@ -2504,7 +2563,7 @@ def _ShutdownInstanceDisks(lu, instance, ignore_primary=False): return result -def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor): +def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name): """Checks if a node has enough free memory. This function check if a given node has the needed amount of free @@ -2520,13 +2579,13 @@ def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor): @param reason: string to use in the error message @type requested: C{int} @param requested: the amount of memory in MiB to check for - @type hypervisor: C{str} - @param hypervisor: the hypervisor to ask for memory stats + @type hypervisor_name: C{str} + @param hypervisor_name: the hypervisor to ask for memory stats @raise errors.OpPrereqError: if the node doesn't have enough memory, or we cannot check the node """ - nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor) + nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name) nodeinfo[node].Raise() free_mem = nodeinfo[node].data.get('memory_free') if not isinstance(free_mem, int): @@ -2560,8 +2619,7 @@ class LUStartupInstance(LogicalUnit): "FORCE": self.op.force, } env.update(_BuildInstanceHookEnvByObject(self, self.instance)) - nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] + - list(self.instance.secondary_nodes)) + nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes) return env, nl, nl def CheckPrereq(self): @@ -2599,9 +2657,10 @@ class LUStartupInstance(LogicalUnit): _StartInstanceDisks(self, instance, force) result = self.rpc.call_instance_start(node_current, instance, extra_args) - if result.failed or not result.data: + msg = result.RemoteFailMsg() + if msg: _ShutdownInstanceDisks(self, instance) - raise errors.OpExecError("Could not start instance") + raise errors.OpExecError("Could not start instance: %s" % msg) class LURebootInstance(LogicalUnit): @@ -2633,8 +2692,7 @@ class LURebootInstance(LogicalUnit): "IGNORE_SECONDARIES": self.op.ignore_secondaries, } env.update(_BuildInstanceHookEnvByObject(self, self.instance)) - nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] + - list(self.instance.secondary_nodes)) + nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes) return env, nl, nl def CheckPrereq(self): @@ -2675,9 +2733,11 @@ class LURebootInstance(LogicalUnit): _ShutdownInstanceDisks(self, instance) _StartInstanceDisks(self, instance, ignore_secondaries) result = self.rpc.call_instance_start(node_current, instance, extra_args) - if result.failed or not result.data: + msg = result.RemoteFailMsg() + if msg: _ShutdownInstanceDisks(self, instance) - raise errors.OpExecError("Could not start instance for full reboot") + raise errors.OpExecError("Could not start instance for" + " full reboot: %s" % msg) self.cfg.MarkInstanceUp(instance.name) @@ -2701,8 +2761,7 @@ class LUShutdownInstance(LogicalUnit): """ env = _BuildInstanceHookEnvByObject(self, self.instance) - nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] + - list(self.instance.secondary_nodes)) + nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes) return env, nl, nl def CheckPrereq(self): @@ -2749,8 +2808,7 @@ class LUReinstallInstance(LogicalUnit): """ env = _BuildInstanceHookEnvByObject(self, self.instance) - nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] + - list(self.instance.secondary_nodes)) + nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes) return env, nl, nl def CheckPrereq(self): @@ -2767,7 +2825,7 @@ class LUReinstallInstance(LogicalUnit): if instance.disk_template == constants.DT_DISKLESS: raise errors.OpPrereqError("Instance '%s' has no disks" % self.op.instance_name) - if instance.status != "down": + if instance.admin_up: raise errors.OpPrereqError("Instance '%s' is marked to be up" % self.op.instance_name) remote_info = self.rpc.call_instance_info(instance.primary_node, @@ -2809,11 +2867,11 @@ class LUReinstallInstance(LogicalUnit): try: feedback_fn("Running the instance OS create scripts...") result = self.rpc.call_instance_os_add(inst.primary_node, inst) - result.Raise() - if not result.data: + msg = result.RemoteFailMsg() + if msg: raise errors.OpExecError("Could not install OS for instance %s" - " on node %s" % - (inst.name, inst.primary_node)) + " on node %s: %s" % + (inst.name, inst.primary_node, msg)) finally: _ShutdownInstanceDisks(self, inst) @@ -2834,8 +2892,7 @@ class LURenameInstance(LogicalUnit): """ env = _BuildInstanceHookEnvByObject(self, self.instance) env["INSTANCE_NEW_NAME"] = self.op.new_name - nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] + - list(self.instance.secondary_nodes)) + nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes) return env, nl, nl def CheckPrereq(self): @@ -2851,7 +2908,7 @@ class LURenameInstance(LogicalUnit): self.op.instance_name) _CheckNodeOnline(self, instance.primary_node) - if instance.status != "down": + if instance.admin_up: raise errors.OpPrereqError("Instance '%s' is marked to be up" % self.op.instance_name) remote_info = self.rpc.call_instance_info(instance.primary_node, @@ -2920,10 +2977,11 @@ class LURenameInstance(LogicalUnit): try: result = self.rpc.call_instance_run_rename(inst.primary_node, inst, old_name) - if result.failed or not result.data: + msg = result.RemoteFailMsg() + if msg: msg = ("Could not run OS rename script for instance %s on node %s" - " (but the instance has been renamed in Ganeti)" % - (inst.name, inst.primary_node)) + " (but the instance has been renamed in Ganeti): %s" % + (inst.name, inst.primary_node, msg)) self.proc.LogWarning(msg) finally: _ShutdownInstanceDisks(self, inst) @@ -3115,7 +3173,7 @@ class LUQueryInstances(NoHooksLU): elif field == "snodes": val = list(instance.secondary_nodes) elif field == "admin_state": - val = (instance.status != "down") + val = instance.admin_up elif field == "oper_state": if instance.primary_node in bad_nodes: val = None @@ -3129,12 +3187,12 @@ class LUQueryInstances(NoHooksLU): else: running = bool(live_data.get(instance.name)) if running: - if instance.status != "down": + if instance.admin_up: val = "running" else: val = "ERROR_up" else: - if instance.status != "down": + if instance.admin_up: val = "ERROR_down" else: val = "ADMIN_down" @@ -3308,7 +3366,7 @@ class LUFailoverInstance(LogicalUnit): for dev in instance.disks: # for drbd, these are drbd over lvm if not _CheckDiskConsistency(self, dev, target_node, False): - if instance.status == "up" and not self.op.ignore_consistency: + if instance.admin_up and not self.op.ignore_consistency: raise errors.OpExecError("Disk %s is degraded on target node," " aborting failover." % dev.iv_name) @@ -3336,7 +3394,7 @@ class LUFailoverInstance(LogicalUnit): self.cfg.Update(instance) # Only start the instance if it's marked as up - if instance.status == "up": + if instance.admin_up: feedback_fn("* activating the instance's disks on target node") logging.info("Starting instance %s on node %s", instance.name, target_node) @@ -3349,60 +3407,454 @@ class LUFailoverInstance(LogicalUnit): feedback_fn("* starting the instance on the target node") result = self.rpc.call_instance_start(target_node, instance, None) - if result.failed or not result.data: + msg = result.RemoteFailMsg() + if msg: _ShutdownInstanceDisks(self, instance) - raise errors.OpExecError("Could not start instance %s on node %s." % - (instance.name, target_node)) + raise errors.OpExecError("Could not start instance %s on node %s: %s" % + (instance.name, target_node, msg)) -def _CreateBlockDevOnPrimary(lu, node, instance, device, info): - """Create a tree of block devices on the primary node. +class LUMigrateInstance(LogicalUnit): + """Migrate an instance. - This always creates all devices. + This is migration without shutting down, compared to the failover, + which is done with shutdown. """ - if device.children: - for child in device.children: - if not _CreateBlockDevOnPrimary(lu, node, instance, child, info): - return False + HPATH = "instance-migrate" + HTYPE = constants.HTYPE_INSTANCE + _OP_REQP = ["instance_name", "live", "cleanup"] - lu.cfg.SetDiskID(device, node) - new_id = lu.rpc.call_blockdev_create(node, device, device.size, - instance.name, True, info) - if new_id.failed or not new_id.data: - return False - if device.physical_id is None: - device.physical_id = new_id - return True + REQ_BGL = False + + def ExpandNames(self): + self._ExpandAndLockInstance() + self.needed_locks[locking.LEVEL_NODE] = [] + self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE + + def DeclareLocks(self, level): + if level == locking.LEVEL_NODE: + self._LockInstancesNodes() + + def BuildHooksEnv(self): + """Build hooks env. + + This runs on master, primary and secondary nodes of the instance. + + """ + env = _BuildInstanceHookEnvByObject(self, self.instance) + nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes) + return env, nl, nl + + def CheckPrereq(self): + """Check prerequisites. + + This checks that the instance is in the cluster. + + """ + instance = self.cfg.GetInstanceInfo( + self.cfg.ExpandInstanceName(self.op.instance_name)) + if instance is None: + raise errors.OpPrereqError("Instance '%s' not known" % + self.op.instance_name) + + if instance.disk_template != constants.DT_DRBD8: + raise errors.OpPrereqError("Instance's disk layout is not" + " drbd8, cannot migrate.") + + secondary_nodes = instance.secondary_nodes + if not secondary_nodes: + raise errors.ProgrammerError("no secondary node but using " + "drbd8 disk template") + + i_be = self.cfg.GetClusterInfo().FillBE(instance) + + target_node = secondary_nodes[0] + # check memory requirements on the secondary node + _CheckNodeFreeMemory(self, target_node, "migrating instance %s" % + instance.name, i_be[constants.BE_MEMORY], + instance.hypervisor) + + # check bridge existance + brlist = [nic.bridge for nic in instance.nics] + result = self.rpc.call_bridges_exist(target_node, brlist) + if result.failed or not result.data: + raise errors.OpPrereqError("One or more target bridges %s does not" + " exist on destination node '%s'" % + (brlist, target_node)) + + if not self.op.cleanup: + result = self.rpc.call_instance_migratable(instance.primary_node, + instance) + msg = result.RemoteFailMsg() + if msg: + raise errors.OpPrereqError("Can't migrate: %s - please use failover" % + msg) + + self.instance = instance + + def _WaitUntilSync(self): + """Poll with custom rpc for disk sync. + + This uses our own step-based rpc call. + + """ + self.feedback_fn("* wait until resync is done") + all_done = False + while not all_done: + all_done = True + result = self.rpc.call_drbd_wait_sync(self.all_nodes, + self.nodes_ip, + self.instance.disks) + min_percent = 100 + for node, nres in result.items(): + msg = nres.RemoteFailMsg() + if msg: + raise errors.OpExecError("Cannot resync disks on node %s: %s" % + (node, msg)) + node_done, node_percent = nres.data[1] + all_done = all_done and node_done + if node_percent is not None: + min_percent = min(min_percent, node_percent) + if not all_done: + if min_percent < 100: + self.feedback_fn(" - progress: %.1f%%" % min_percent) + time.sleep(2) + + def _EnsureSecondary(self, node): + """Demote a node to secondary. + """ + self.feedback_fn("* switching node %s to secondary mode" % node) + + for dev in self.instance.disks: + self.cfg.SetDiskID(dev, node) + + result = self.rpc.call_blockdev_close(node, self.instance.name, + self.instance.disks) + msg = result.RemoteFailMsg() + if msg: + raise errors.OpExecError("Cannot change disk to secondary on node %s," + " error %s" % (node, msg)) + + def _GoStandalone(self): + """Disconnect from the network. + + """ + self.feedback_fn("* changing into standalone mode") + result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip, + self.instance.disks) + for node, nres in result.items(): + msg = nres.RemoteFailMsg() + if msg: + raise errors.OpExecError("Cannot disconnect disks node %s," + " error %s" % (node, msg)) + + def _GoReconnect(self, multimaster): + """Reconnect to the network. + + """ + if multimaster: + msg = "dual-master" + else: + msg = "single-master" + self.feedback_fn("* changing disks into %s mode" % msg) + result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip, + self.instance.disks, + self.instance.name, multimaster) + for node, nres in result.items(): + msg = nres.RemoteFailMsg() + if msg: + raise errors.OpExecError("Cannot change disks config on node %s," + " error: %s" % (node, msg)) + + def _ExecCleanup(self): + """Try to cleanup after a failed migration. + + The cleanup is done by: + - check that the instance is running only on one node + (and update the config if needed) + - change disks on its secondary node to secondary + - wait until disks are fully synchronized + - disconnect from the network + - change disks into single-master mode + - wait again until disks are fully synchronized -def _CreateBlockDevOnSecondary(lu, node, instance, device, force, info): - """Create a tree of block devices on a secondary node. + """ + instance = self.instance + target_node = self.target_node + source_node = self.source_node + + # check running on only one node + self.feedback_fn("* checking where the instance actually runs" + " (if this hangs, the hypervisor might be in" + " a bad state)") + ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor]) + for node, result in ins_l.items(): + result.Raise() + if not isinstance(result.data, list): + raise errors.OpExecError("Can't contact node '%s'" % node) + + runningon_source = instance.name in ins_l[source_node].data + runningon_target = instance.name in ins_l[target_node].data + + if runningon_source and runningon_target: + raise errors.OpExecError("Instance seems to be running on two nodes," + " or the hypervisor is confused. You will have" + " to ensure manually that it runs only on one" + " and restart this operation.") + + if not (runningon_source or runningon_target): + raise errors.OpExecError("Instance does not seem to be running at all." + " In this case, it's safer to repair by" + " running 'gnt-instance stop' to ensure disk" + " shutdown, and then restarting it.") + + if runningon_target: + # the migration has actually succeeded, we need to update the config + self.feedback_fn("* instance running on secondary node (%s)," + " updating config" % target_node) + instance.primary_node = target_node + self.cfg.Update(instance) + demoted_node = source_node + else: + self.feedback_fn("* instance confirmed to be running on its" + " primary node (%s)" % source_node) + demoted_node = target_node + + self._EnsureSecondary(demoted_node) + try: + self._WaitUntilSync() + except errors.OpExecError: + # we ignore here errors, since if the device is standalone, it + # won't be able to sync + pass + self._GoStandalone() + self._GoReconnect(False) + self._WaitUntilSync() + + self.feedback_fn("* done") + + def _RevertDiskStatus(self): + """Try to revert the disk status after a failed migration. + + """ + target_node = self.target_node + try: + self._EnsureSecondary(target_node) + self._GoStandalone() + self._GoReconnect(False) + self._WaitUntilSync() + except errors.OpExecError, err: + self.LogWarning("Migration failed and I can't reconnect the" + " drives: error '%s'\n" + "Please look and recover the instance status" % + str(err)) + + def _AbortMigration(self): + """Call the hypervisor code to abort a started migration. + + """ + instance = self.instance + target_node = self.target_node + migration_info = self.migration_info + + abort_result = self.rpc.call_finalize_migration(target_node, + instance, + migration_info, + False) + abort_msg = abort_result.RemoteFailMsg() + if abort_msg: + logging.error("Aborting migration failed on target node %s: %s" % + (target_node, abort_msg)) + # Don't raise an exception here, as we stil have to try to revert the + # disk status, even if this step failed. + + def _ExecMigration(self): + """Migrate an instance. + + The migrate is done by: + - change the disks into dual-master mode + - wait until disks are fully synchronized again + - migrate the instance + - change disks on the new secondary node (the old primary) to secondary + - wait until disks are fully synchronized + - change disks into single-master mode + + """ + instance = self.instance + target_node = self.target_node + source_node = self.source_node + + self.feedback_fn("* checking disk consistency between source and target") + for dev in instance.disks: + if not _CheckDiskConsistency(self, dev, target_node, False): + raise errors.OpExecError("Disk %s is degraded or not fully" + " synchronized on target node," + " aborting migrate." % dev.iv_name) + + # First get the migration information from the remote node + result = self.rpc.call_migration_info(source_node, instance) + msg = result.RemoteFailMsg() + if msg: + log_err = ("Failed fetching source migration information from %s: %s" % + (source_node, msg)) + logging.error(log_err) + raise errors.OpExecError(log_err) + + self.migration_info = migration_info = result.data[1] + + # Then switch the disks to master/master mode + self._EnsureSecondary(target_node) + self._GoStandalone() + self._GoReconnect(True) + self._WaitUntilSync() + + self.feedback_fn("* preparing %s to accept the instance" % target_node) + result = self.rpc.call_accept_instance(target_node, + instance, + migration_info, + self.nodes_ip[target_node]) + + msg = result.RemoteFailMsg() + if msg: + logging.error("Instance pre-migration failed, trying to revert" + " disk status: %s", msg) + self._AbortMigration() + self._RevertDiskStatus() + raise errors.OpExecError("Could not pre-migrate instance %s: %s" % + (instance.name, msg)) + + self.feedback_fn("* migrating instance to %s" % target_node) + time.sleep(10) + result = self.rpc.call_instance_migrate(source_node, instance, + self.nodes_ip[target_node], + self.op.live) + msg = result.RemoteFailMsg() + if msg: + logging.error("Instance migration failed, trying to revert" + " disk status: %s", msg) + self._AbortMigration() + self._RevertDiskStatus() + raise errors.OpExecError("Could not migrate instance %s: %s" % + (instance.name, msg)) + time.sleep(10) + + instance.primary_node = target_node + # distribute new instance config to the other nodes + self.cfg.Update(instance) + + result = self.rpc.call_finalize_migration(target_node, + instance, + migration_info, + True) + msg = result.RemoteFailMsg() + if msg: + logging.error("Instance migration succeeded, but finalization failed:" + " %s" % msg) + raise errors.OpExecError("Could not finalize instance migration: %s" % + msg) + + self._EnsureSecondary(source_node) + self._WaitUntilSync() + self._GoStandalone() + self._GoReconnect(False) + self._WaitUntilSync() + + self.feedback_fn("* done") + + def Exec(self, feedback_fn): + """Perform the migration. + + """ + self.feedback_fn = feedback_fn + + self.source_node = self.instance.primary_node + self.target_node = self.instance.secondary_nodes[0] + self.all_nodes = [self.source_node, self.target_node] + self.nodes_ip = { + self.source_node: self.cfg.GetNodeInfo(self.source_node).secondary_ip, + self.target_node: self.cfg.GetNodeInfo(self.target_node).secondary_ip, + } + if self.op.cleanup: + return self._ExecCleanup() + else: + return self._ExecMigration() + + +def _CreateBlockDev(lu, node, instance, device, force_create, + info, force_open): + """Create a tree of block devices on a given node. If this device type has to be created on secondaries, create it and all its children. If not, just recurse to children keeping the same 'force' value. + @param lu: the lu on whose behalf we execute + @param node: the node on which to create the device + @type instance: L{objects.Instance} + @param instance: the instance which owns the device + @type device: L{objects.Disk} + @param device: the device to create + @type force_create: boolean + @param force_create: whether to force creation of this device; this + will be change to True whenever we find a device which has + CreateOnSecondary() attribute + @param info: the extra 'metadata' we should attach to the device + (this will be represented as a LVM tag) + @type force_open: boolean + @param force_open: this parameter will be passes to the + L{backend.CreateBlockDevice} function where it specifies + whether we run on primary or not, and it affects both + the child assembly and the device own Open() execution + """ if device.CreateOnSecondary(): - force = True + force_create = True + if device.children: for child in device.children: - if not _CreateBlockDevOnSecondary(lu, node, instance, - child, force, info): - return False + _CreateBlockDev(lu, node, instance, child, force_create, + info, force_open) - if not force: - return True + if not force_create: + return + + _CreateSingleBlockDev(lu, node, instance, device, info, force_open) + + +def _CreateSingleBlockDev(lu, node, instance, device, info, force_open): + """Create a single block device on a given node. + + This will not recurse over children of the device, so they must be + created in advance. + + @param lu: the lu on whose behalf we execute + @param node: the node on which to create the device + @type instance: L{objects.Instance} + @param instance: the instance which owns the device + @type device: L{objects.Disk} + @param device: the device to create + @param info: the extra 'metadata' we should attach to the device + (this will be represented as a LVM tag) + @type force_open: boolean + @param force_open: this parameter will be passes to the + L{backend.CreateBlockDevice} function where it specifies + whether we run on primary or not, and it affects both + the child assembly and the device own Open() execution + + """ lu.cfg.SetDiskID(device, node) - new_id = lu.rpc.call_blockdev_create(node, device, device.size, - instance.name, False, info) - if new_id.failed or not new_id.data: - return False + result = lu.rpc.call_blockdev_create(node, device, device.size, + instance.name, force_open, info) + msg = result.RemoteFailMsg() + if msg: + raise errors.OpExecError("Can't create block device %s on" + " node %s for instance %s: %s" % + (device, node, instance.name, msg)) if device.physical_id is None: - device.physical_id = new_id - return True + device.physical_id = result.data[1] def _GenerateUniqueNames(lu, exts): @@ -3464,7 +3916,8 @@ def _GenerateDiskTemplate(lu, template_name, disk_index = idx + base_index disk_dev = objects.Disk(dev_type=constants.LD_LV, size=disk["size"], logical_id=(vgname, names[idx]), - iv_name="disk/%d" % disk_index) + iv_name="disk/%d" % disk_index, + mode=disk["mode"]) disks.append(disk_dev) elif template_name == constants.DT_DRBD8: if len(secondary_nodes) != 1: @@ -3473,17 +3926,18 @@ def _GenerateDiskTemplate(lu, template_name, minors = lu.cfg.AllocateDRBDMinor( [primary_node, remote_node] * len(disk_info), instance_name) - names = _GenerateUniqueNames(lu, - [".disk%d_%s" % (i, s) - for i in range(disk_count) - for s in ("data", "meta") - ]) + names = [] + for lv_prefix in _GenerateUniqueNames(lu, [".disk%d" % i + for i in range(disk_count)]): + names.append(lv_prefix + "_data") + names.append(lv_prefix + "_meta") for idx, disk in enumerate(disk_info): disk_index = idx + base_index disk_dev = _GenerateDRBD8Branch(lu, primary_node, remote_node, disk["size"], names[idx*2:idx*2+2], "disk/%d" % disk_index, minors[idx*2], minors[idx*2+1]) + disk_dev.mode = disk["mode"] disks.append(disk_dev) elif template_name == constants.DT_FILE: if len(secondary_nodes) != 0: @@ -3495,7 +3949,8 @@ def _GenerateDiskTemplate(lu, template_name, iv_name="disk/%d" % disk_index, logical_id=(file_driver, "%s/disk%d" % (file_storage_dir, - idx))) + idx)), + mode=disk["mode"]) disks.append(disk_dev) else: raise errors.ProgrammerError("Invalid disk template '%s'" % template_name) @@ -3523,19 +3978,18 @@ def _CreateDisks(lu, instance): """ info = _GetInstanceInfoText(instance) + pnode = instance.primary_node if instance.disk_template == constants.DT_FILE: file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1]) - result = lu.rpc.call_file_storage_dir_create(instance.primary_node, - file_storage_dir) + result = lu.rpc.call_file_storage_dir_create(pnode, file_storage_dir) if result.failed or not result.data: - logging.error("Could not connect to node '%s'", instance.primary_node) - return False + raise errors.OpExecError("Could not connect to node '%s'" % pnode) if not result.data[0]: - logging.error("Failed to create directory '%s'", file_storage_dir) - return False + raise errors.OpExecError("Failed to create directory '%s'" % + file_storage_dir) # Note: this needs to be kept in sync with adding of disks in # LUSetInstanceParams @@ -3543,19 +3997,9 @@ def _CreateDisks(lu, instance): logging.info("Creating volume %s for instance %s", device.iv_name, instance.name) #HARDCODE - for secondary_node in instance.secondary_nodes: - if not _CreateBlockDevOnSecondary(lu, secondary_node, instance, - device, False, info): - logging.error("Failed to create volume %s (%s) on secondary node %s!", - device.iv_name, device, secondary_node) - return False - #HARDCODE - if not _CreateBlockDevOnPrimary(lu, instance.primary_node, - instance, device, info): - logging.error("Failed to create volume %s on primary!", device.iv_name) - return False - - return True + for node in instance.all_nodes: + f_create = node == pnode + _CreateBlockDev(lu, node, instance, device, f_create, info, f_create) def _RemoveDisks(lu, instance): @@ -3639,6 +4083,8 @@ def _CheckHVParams(lu, nodenames, hvname, hvparams): hvparams) for node in nodenames: info = hvinfo[node] + if info.offline: + continue info.Raise() if not info.data or not isinstance(info.data, (tuple, list)): raise errors.OpPrereqError("Cannot get current information" @@ -3752,7 +4198,9 @@ class LUCreateInstance(LogicalUnit): raise errors.OpPrereqError("Invalid MAC address specified: %s" % mac) # bridge verification - bridge = nic.get("bridge", self.cfg.GetDefBridge()) + bridge = nic.get("bridge", None) + if bridge is None: + bridge = self.cfg.GetDefBridge() self.nics.append(objects.NIC(mac=mac, ip=nic_ip, bridge=bridge)) # disk checks/pre-build @@ -4054,10 +4502,7 @@ class LUCreateInstance(LogicalUnit): self.be_full[constants.BE_MEMORY], self.op.hypervisor) - if self.op.start: - self.instance_status = 'up' - else: - self.instance_status = 'down' + self.instance_status = self.op.start def Exec(self, feedback_fn): """Create and add the instance to the cluster. @@ -4104,7 +4549,7 @@ class LUCreateInstance(LogicalUnit): primary_node=pnode_name, nics=self.nics, disks=disks, disk_template=self.op.disk_template, - status=self.instance_status, + admin_up=self.instance_status, network_port=network_port, beparams=self.op.beparams, hvparams=self.op.hvparams, @@ -4112,10 +4557,15 @@ class LUCreateInstance(LogicalUnit): ) feedback_fn("* creating instance disks...") - if not _CreateDisks(self, iobj): - _RemoveDisks(self, iobj) - self.cfg.ReleaseDRBDMinors(instance) - raise errors.OpExecError("Device creation failed, reverting...") + try: + _CreateDisks(self, iobj) + except errors.OpExecError: + self.LogWarning("Device creation failed, reverting...") + try: + _RemoveDisks(self, iobj) + finally: + self.cfg.ReleaseDRBDMinors(instance) + raise feedback_fn("adding instance %s to cluster config" % instance) @@ -4123,8 +4573,6 @@ class LUCreateInstance(LogicalUnit): # Declare that we don't want to remove the instance lock anymore, as we've # added the instance to the config del self.remove_locks[locking.LEVEL_INSTANCE] - # Remove the temp. assignements for the instance's drbds - self.cfg.ReleaseDRBDMinors(instance) # Unlock all the nodes if self.op.mode == constants.INSTANCE_IMPORT: nodes_keep = [self.op.src_node] @@ -4161,11 +4609,11 @@ class LUCreateInstance(LogicalUnit): if self.op.mode == constants.INSTANCE_CREATE: feedback_fn("* running the instance OS create scripts...") result = self.rpc.call_instance_os_add(pnode_name, iobj) - result.Raise() - if not result.data: + msg = result.RemoteFailMsg() + if msg: raise errors.OpExecError("Could not add os for instance %s" - " on node %s" % - (instance, pnode_name)) + " on node %s: %s" % + (instance, pnode_name, msg)) elif self.op.mode == constants.INSTANCE_IMPORT: feedback_fn("* running the instance OS import scripts...") @@ -4190,9 +4638,9 @@ class LUCreateInstance(LogicalUnit): logging.info("Starting instance %s on node %s", instance, pnode_name) feedback_fn("* starting instance...") result = self.rpc.call_instance_start(pnode_name, iobj, None) - result.Raise() - if not result.data: - raise errors.OpExecError("Could not start instance") + msg = result.RemoteFailMsg() + if msg: + raise errors.OpExecError("Could not start instance: %s" % msg) class LUConnectConsole(NoHooksLU): @@ -4218,7 +4666,7 @@ class LUConnectConsole(NoHooksLU): self.instance = self.cfg.GetInstanceInfo(self.op.instance_name) assert self.instance is not None, \ "Cannot retrieve locked instance %s" % self.op.instance_name - _CheckNodeOnline(self, self.op.primary_node) + _CheckNodeOnline(self, self.instance.primary_node) def Exec(self, feedback_fn): """Connect to the console of an instance @@ -4252,17 +4700,32 @@ class LUReplaceDisks(LogicalUnit): _OP_REQP = ["instance_name", "mode", "disks"] REQ_BGL = False - def ExpandNames(self): - self._ExpandAndLockInstance() - + def CheckArguments(self): if not hasattr(self.op, "remote_node"): self.op.remote_node = None - - ia_name = getattr(self.op, "iallocator", None) - if ia_name is not None: - if self.op.remote_node is not None: + if not hasattr(self.op, "iallocator"): + self.op.iallocator = None + + # check for valid parameter combination + cnt = [self.op.remote_node, self.op.iallocator].count(None) + if self.op.mode == constants.REPLACE_DISK_CHG: + if cnt == 2: + raise errors.OpPrereqError("When changing the secondary either an" + " iallocator script must be used or the" + " new node given") + elif cnt == 0: raise errors.OpPrereqError("Give either the iallocator or the new" " secondary, not both") + else: # not replacing the secondary + if cnt != 2: + raise errors.OpPrereqError("The iallocator and new node options can" + " be used only when changing the" + " secondary node") + + def ExpandNames(self): + self._ExpandAndLockInstance() + + if self.op.iallocator is not None: self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET elif self.op.remote_node is not None: remote_node = self.cfg.ExpandNodeName(self.op.remote_node) @@ -4270,6 +4733,10 @@ class LUReplaceDisks(LogicalUnit): raise errors.OpPrereqError("Node '%s' not known" % self.op.remote_node) self.op.remote_node = remote_node + # Warning: do not remove the locking of the new secondary here + # unless DRBD8.AddChildren is changed to work in parallel; + # currently it doesn't since parallel invocations of + # FindUnusedMinor will conflict self.needed_locks[locking.LEVEL_NODE] = [remote_node] self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND else: @@ -4337,9 +4804,9 @@ class LUReplaceDisks(LogicalUnit): "Cannot retrieve locked instance %s" % self.op.instance_name self.instance = instance - if instance.disk_template not in constants.DTS_NET_MIRROR: - raise errors.OpPrereqError("Instance's disk layout is not" - " network mirrored.") + if instance.disk_template != constants.DT_DRBD8: + raise errors.OpPrereqError("Can only run replace disks for DRBD8-based" + " instances") if len(instance.secondary_nodes) != 1: raise errors.OpPrereqError("The instance has a strange layout," @@ -4348,8 +4815,7 @@ class LUReplaceDisks(LogicalUnit): self.sec_node = instance.secondary_nodes[0] - ia_name = getattr(self.op, "iallocator", None) - if ia_name is not None: + if self.op.iallocator is not None: self._RunAllocator() remote_node = self.op.remote_node @@ -4363,42 +4829,24 @@ 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") - 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 '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 '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] - _CheckNodeOnline(self, self.tgt_node) - _CheckNodeOnline(self, self.oth_node) - 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 - _CheckNodeOnline(self, self.oth_node) - if self.new_node is not None: - _CheckNodeOnline(self, self.new_node) - else: - _CheckNodeOnline(self, self.tgt_node) - else: - raise errors.ProgrammerError("Unhandled disk replace mode") + raise errors.OpPrereqError("The specified node is already the" + " secondary node of the instance.") + + if self.op.mode == constants.REPLACE_DISK_PRI: + n1 = self.tgt_node = instance.primary_node + n2 = self.oth_node = self.sec_node + elif self.op.mode == constants.REPLACE_DISK_SEC: + n1 = self.tgt_node = self.sec_node + n2 = self.oth_node = instance.primary_node + elif self.op.mode == constants.REPLACE_DISK_CHG: + n1 = self.new_node = remote_node + n2 = self.oth_node = instance.primary_node + self.tgt_node = self.sec_node + else: + raise errors.ProgrammerError("Unhandled disk replace mode") + + _CheckNodeOnline(self, n1) + _CheckNodeOnline(self, n2) if not self.op.disks: self.op.disks = range(len(instance.disks)) @@ -4491,15 +4939,10 @@ class LUReplaceDisks(LogicalUnit): iv_names[dev.iv_name] = (dev, old_lvs, new_lvs) 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 + # we pass force_create=True to force the LVM creation for new_lv in new_lvs: - if not _CreateBlockDevOnPrimary(self, tgt_node, instance, new_lv, - _GetInstanceInfoText(instance)): - raise errors.OpExecError("Failed to create new LV named '%s' on" - " node '%s'" % - (new_lv.logical_id[1], tgt_node)) + _CreateBlockDev(self, tgt_node, instance, new_lv, True, + _GetInstanceInfoText(instance), False) # Step: for each lv, detach+rename*2+attach self.proc.LogStep(4, steps_total, "change drbd configuration") @@ -4553,7 +4996,7 @@ class LUReplaceDisks(LogicalUnit): # now that the new lvs have the old name, we can add them to the device info("adding new mirror component on %s" % tgt_node) - result =self.rpc.call_blockdev_addchildren(tgt_node, dev, new_lvs) + result = self.rpc.call_blockdev_addchildren(tgt_node, dev, new_lvs) if result.failed or not result.data: for new_lv in new_lvs: result = self.rpc.call_blockdev_remove(tgt_node, new_lv) @@ -4614,12 +5057,16 @@ class LUReplaceDisks(LogicalUnit): warning, info = (self.proc.LogWarning, self.proc.LogInfo) instance = self.instance iv_names = {} - vgname = self.cfg.GetVGName() # start of work cfg = self.cfg old_node = self.tgt_node new_node = self.new_node pri_node = instance.primary_node + nodes_ip = { + old_node: self.cfg.GetNodeInfo(old_node).secondary_ip, + new_node: self.cfg.GetNodeInfo(new_node).secondary_ip, + pri_node: self.cfg.GetNodeInfo(pri_node).secondary_ip, + } # Step: check device activation self.proc.LogStep(1, steps_total, "check device existence") @@ -4656,18 +5103,12 @@ class LUReplaceDisks(LogicalUnit): # Step: create new storage self.proc.LogStep(3, steps_total, "allocate new storage") for idx, dev in enumerate(instance.disks): - size = dev.size info("adding new local storage on %s for disk/%d" % (new_node, idx)) - # 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 + # we pass force_create=True to force LVM creation for new_lv in dev.children: - if not _CreateBlockDevOnPrimary(self, new_node, instance, new_lv, - _GetInstanceInfoText(instance)): - raise errors.OpExecError("Failed to create new LV named '%s' on" - " node '%s'" % - (new_lv.logical_id[1], new_node)) + _CreateBlockDev(self, new_node, instance, new_lv, True, + _GetInstanceInfoText(instance), False) # Step 4: dbrd minors and drbd setups changes # after this, we must manually remove the drbd minors on both the @@ -4679,27 +5120,31 @@ class LUReplaceDisks(LogicalUnit): for idx, (dev, new_minor) in enumerate(zip(instance.disks, minors)): size = dev.size info("activating a new drbd on %s for disk/%d" % (new_node, idx)) - # create new devices on new_node - if pri_node == dev.logical_id[0]: - new_logical_id = (pri_node, new_node, - dev.logical_id[2], dev.logical_id[3], new_minor, - dev.logical_id[5]) + # create new devices on new_node; note that we create two IDs: + # one without port, so the drbd will be activated without + # networking information on the new node at this stage, and one + # with network, for the latter activation in step 4 + (o_node1, o_node2, o_port, o_minor1, o_minor2, o_secret) = dev.logical_id + if pri_node == o_node1: + p_minor = o_minor1 else: - new_logical_id = (new_node, pri_node, - dev.logical_id[2], new_minor, dev.logical_id[4], - dev.logical_id[5]) - iv_names[idx] = (dev, dev.children, new_logical_id) + p_minor = o_minor2 + + new_alone_id = (pri_node, new_node, None, p_minor, new_minor, o_secret) + new_net_id = (pri_node, new_node, o_port, p_minor, new_minor, o_secret) + + iv_names[idx] = (dev, dev.children, new_net_id) logging.debug("Allocated new_minor: %s, new_logical_id: %s", new_minor, - new_logical_id) + new_net_id) new_drbd = objects.Disk(dev_type=constants.LD_DRBD8, - logical_id=new_logical_id, + logical_id=new_alone_id, children=dev.children) - if not _CreateBlockDevOnSecondary(self, new_node, instance, - new_drbd, False, - _GetInstanceInfoText(instance)): + try: + _CreateSingleBlockDev(self, new_node, instance, new_drbd, + _GetInstanceInfoText(instance), False) + except errors.BlockDeviceError: self.cfg.ReleaseDRBDMinors(instance.name) - raise errors.OpExecError("Failed to create new DRBD on" - " node '%s'" % new_node) + raise for idx, dev in enumerate(instance.disks): # we have new devices, shutdown the drbd on the old secondary @@ -4711,25 +5156,15 @@ class LUReplaceDisks(LogicalUnit): hint="Please cleanup this device manually as soon as possible") info("detaching primary drbds from the network (=> standalone)") - done = 0 - for idx, dev in enumerate(instance.disks): - cfg.SetDiskID(dev, pri_node) - # set the network part of the physical (unique in bdev terms) id - # to None, meaning detach from network - dev.physical_id = (None, None, None, None) + dev.physical_id[4:] - # and 'find' the device, which will 'fix' it to match the - # standalone state - result = self.rpc.call_blockdev_find(pri_node, dev) - if not result.failed and result.data: - done += 1 - else: - warning("Failed to detach drbd disk/%d from network, unusual case" % - idx) + result = self.rpc.call_drbd_disconnect_net([pri_node], nodes_ip, + instance.disks)[pri_node] - if not done: - # no detaches succeeded (very unlikely) + msg = result.RemoteFailMsg() + if msg: + # detaches didn't succeed (unlikely) self.cfg.ReleaseDRBDMinors(instance.name) - raise errors.OpExecError("Can't detach at least one DRBD from old node") + raise errors.OpExecError("Can't detach the disks from the network on" + " old node: %s" % (msg,)) # if we managed to detach at least one, we update all the disks of # the instance to point to the new secondary @@ -4738,24 +5173,18 @@ class LUReplaceDisks(LogicalUnit): dev.logical_id = new_logical_id cfg.SetDiskID(dev, pri_node) cfg.Update(instance) - # we can remove now the temp minors as now the new values are - # written to the config file (and therefore stable) - self.cfg.ReleaseDRBDMinors(instance.name) # and now perform the drbd attach info("attaching primary drbds to new secondary (standalone => connected)") - failures = [] - for idx, dev in enumerate(instance.disks): - info("attaching primary drbd for disk/%d to new secondary node" % idx) - # 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) - logging.debug("Disk to attach: %s", dev) - result = self.rpc.call_blockdev_find(pri_node, dev) - if result.failed or not result.data: - warning("can't attach drbd disk/%d to new secondary!" % idx, - "please do a gnt-instance info to see the status of disks") + result = self.rpc.call_drbd_attach_net([pri_node, new_node], nodes_ip, + instance.disks, instance.name, + False) + for to_node, to_result in result.items(): + msg = to_result.RemoteFailMsg() + if msg: + warning("can't attach drbd disks on node %s: %s", to_node, msg, + hint="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 @@ -4790,21 +5219,18 @@ class LUReplaceDisks(LogicalUnit): instance = self.instance # Activate the instance disks if we're replacing them on a down instance - if instance.status == "down": + if not instance.admin_up: _StartInstanceDisks(self, instance, True) - if instance.disk_template == constants.DT_DRBD8: - if self.op.remote_node is None: - fn = self._ExecD8DiskOnly - else: - fn = self._ExecD8Secondary + if self.op.mode == constants.REPLACE_DISK_CHG: + fn = self._ExecD8Secondary else: - raise errors.ProgrammerError("Unhandled disk replacement case") + fn = self._ExecD8DiskOnly ret = fn(feedback_fn) # Deactivate the instance disks if we're replacing them on a down instance - if instance.status == "down": + if not instance.admin_up: _SafeShutdownInstanceDisks(self, instance) return ret @@ -4854,8 +5280,8 @@ class LUGrowDisk(LogicalUnit): instance = self.cfg.GetInstanceInfo(self.op.instance_name) assert instance is not None, \ "Cannot retrieve locked instance %s" % self.op.instance_name - _CheckNodeOnline(self, instance.primary_node) - for node in instance.secondary_nodes: + nodenames = list(instance.all_nodes) + for node in nodenames: _CheckNodeOnline(self, node) @@ -4867,7 +5293,6 @@ class LUGrowDisk(LogicalUnit): self.disk = instance.FindDisk(self.op.disk) - nodenames = [instance.primary_node] + list(instance.secondary_nodes) nodeinfo = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(), instance.hypervisor) for node in nodenames: @@ -4890,7 +5315,7 @@ class LUGrowDisk(LogicalUnit): """ instance = self.instance disk = self.disk - for node in (instance.secondary_nodes + (instance.primary_node,)): + for node in instance.all_nodes: self.cfg.SetDiskID(disk, node) result = self.rpc.call_blockdev_grow(node, disk, self.op.amount) result.Raise() @@ -4928,8 +5353,7 @@ class LUQueryInstanceData(NoHooksLU): for name in self.op.instances: full_name = self.cfg.ExpandInstanceName(name) if full_name is None: - raise errors.OpPrereqError("Instance '%s' not known" % - self.op.instance_name) + raise errors.OpPrereqError("Instance '%s' not known" % name) self.wanted_names.append(full_name) self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted_names else: @@ -5022,10 +5446,10 @@ class LUQueryInstanceData(NoHooksLU): remote_state = "down" else: remote_state = None - if instance.status == "down": - config_state = "down" - else: + if instance.admin_up: config_state = "up" + else: + config_state = "down" disks = [self._ComputeDiskStatus(instance, None, device) for device in instance.disks] @@ -5090,7 +5514,7 @@ class LUSetInstanceParams(LogicalUnit): raise errors.OpPrereqError("Invalid disk index") if disk_op == constants.DDM_ADD: mode = disk_dict.setdefault('mode', constants.DISK_RDWR) - if mode not in (constants.DISK_RDONLY, constants.DISK_RDWR): + if mode not in constants.DISK_ACCESS_SET: raise errors.OpPrereqError("Invalid disk access mode '%s'" % mode) size = disk_dict.get('size', None) if size is None: @@ -5170,8 +5594,7 @@ class LUSetInstanceParams(LogicalUnit): args['vcpus'] = self.be_new[constants.BE_VCPUS] # FIXME: readd disk/nic changes env = _BuildInstanceHookEnvByObject(self, self.instance, override=args) - nl = [self.cfg.GetMasterNode(), - self.instance.primary_node] + list(self.instance.secondary_nodes) + nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes) return env, nl, nl def CheckPrereq(self): @@ -5187,9 +5610,8 @@ class LUSetInstanceParams(LogicalUnit): instance = self.instance = self.cfg.GetInstanceInfo(self.op.instance_name) assert self.instance is not None, \ "Cannot retrieve locked instance %s" % self.op.instance_name - pnode = self.instance.primary_node - nodelist = [pnode] - nodelist.extend(instance.secondary_nodes) + pnode = instance.primary_node + nodelist = list(instance.all_nodes) # hvparams processing if self.op.hvparams: @@ -5265,7 +5687,9 @@ class LUSetInstanceParams(LogicalUnit): " missing on its primary node" % miss_mem) if be_new[constants.BE_AUTO_BALANCE]: - for node, nres in instance.secondary_nodes.iteritems(): + for node, nres in nodeinfo.iteritems(): + if node not in instance.secondary_nodes: + continue if nres.failed or not isinstance(nres.data, dict): self.warn.append("Can't get info from secondary node %s" % node) elif be_new[constants.BE_MEMORY] > nres.data['memory_free']: @@ -5305,9 +5729,9 @@ class LUSetInstanceParams(LogicalUnit): " an instance") ins_l = self.rpc.call_instance_list([pnode], [instance.hypervisor]) ins_l = ins_l[pnode] - if not type(ins_l) is list: + if ins_l.failed or not isinstance(ins_l.data, list): raise errors.OpPrereqError("Can't contact node '%s'" % pnode) - if instance.name in ins_l: + if instance.name in ins_l.data: raise errors.OpPrereqError("Instance is running, can't remove" " disks.") @@ -5345,8 +5769,8 @@ class LUSetInstanceParams(LogicalUnit): device_idx = len(instance.disks) for node, disk in device.ComputeNodeTree(instance.primary_node): self.cfg.SetDiskID(disk, node) - result = self.rpc.call_blockdev_remove(node, disk) - if result.failed or not result.data: + rpc_result = self.rpc.call_blockdev_remove(node, disk) + if rpc_result.failed or not rpc_result.data: self.proc.LogWarning("Could not remove disk/%d on node %s," " continuing anyway", device_idx, node) result.append(("disk/%d" % device_idx, "remove")) @@ -5360,13 +5784,12 @@ class LUSetInstanceParams(LogicalUnit): disk_idx_base = len(instance.disks) new_disk = _GenerateDiskTemplate(self, instance.disk_template, - instance, instance.primary_node, + instance.name, instance.primary_node, instance.secondary_nodes, [disk_dict], file_path, file_driver, disk_idx_base)[0] - new_disk.mode = disk_dict['mode'] instance.disks.append(new_disk) info = _GetInstanceInfoText(instance) @@ -5374,17 +5797,15 @@ class LUSetInstanceParams(LogicalUnit): new_disk.iv_name, instance.name) # Note: this needs to be kept in sync with _CreateDisks #HARDCODE - for secondary_node in instance.secondary_nodes: - if not _CreateBlockDevOnSecondary(self, secondary_node, instance, - new_disk, False, info): + for node in instance.all_nodes: + f_create = node == instance.primary_node + try: + _CreateBlockDev(self, node, instance, new_disk, + f_create, info, f_create) + except errors.OpExecError, err: self.LogWarning("Failed to create volume %s (%s) on" - " secondary node %s!", - new_disk.iv_name, new_disk, secondary_node) - #HARDCODE - if not _CreateBlockDevOnPrimary(self, instance.primary_node, - instance, new_disk, info): - self.LogWarning("Failed to create volume %s on primary!", - new_disk.iv_name) + " node %s: %s", + new_disk.iv_name, new_disk, node, err) result.append(("disk/%d" % disk_idx_base, "add:size=%s,mode=%s" % (new_disk.size, new_disk.mode))) else: @@ -5535,7 +5956,7 @@ class LUExportInstance(LogicalUnit): if self.dst_node is None: # This is wrong node name, not a non-locked node raise errors.OpPrereqError("Wrong node name %s" % self.op.target_node) - _CheckNodeOnline(self, self.op.target_node) + _CheckNodeOnline(self, self.dst_node.name) # instance disk type verification for disk in self.instance.disks: @@ -5562,6 +5983,11 @@ class LUExportInstance(LogicalUnit): snap_disks = [] + # set the disks ID correctly since call_instance_start needs the + # correct drbd minor to create the symlinks + for disk in instance.disks: + self.cfg.SetDiskID(disk, src_node) + try: for disk in instance.disks: # new_dev_name will be a snapshot of an lvm leaf of the one we passed @@ -5578,11 +6004,12 @@ class LUExportInstance(LogicalUnit): snap_disks.append(new_dev) finally: - if self.op.shutdown and instance.status == "up": + if self.op.shutdown and instance.admin_up: result = self.rpc.call_instance_start(src_node, instance, None) - if result.failed or not result.data: + msg = result.RemoteFailMsg() + if msg: _ShutdownInstanceDisks(self, instance) - raise errors.OpExecError("Could not start instance") + raise errors.OpExecError("Could not start instance: %s" % msg) # TODO: check for size @@ -5916,6 +6343,7 @@ class IAllocator(object): self.name = name self.mem_size = self.disks = self.disk_template = None self.os = self.tags = self.nics = self.vcpus = None + self.hypervisor = None self.relocate_from = None # computed fields self.required_nodes = None @@ -5952,7 +6380,7 @@ class IAllocator(object): "version": 1, "cluster_name": cfg.GetClusterName(), "cluster_tags": list(cluster_info.GetTags()), - "enable_hypervisors": list(cluster_info.enabled_hypervisors), + "enabled_hypervisors": list(cluster_info.enabled_hypervisors), # we don't have job IDs } iinfo = cfg.GetAllInstancesInfo().values() @@ -5963,60 +6391,68 @@ class IAllocator(object): node_list = cfg.GetNodeList() if self.mode == constants.IALLOCATOR_MODE_ALLOC: - hypervisor = self.hypervisor + hypervisor_name = self.hypervisor elif self.mode == constants.IALLOCATOR_MODE_RELOC: - hypervisor = cfg.GetInstanceInfo(self.name).hypervisor + hypervisor_name = cfg.GetInstanceInfo(self.name).hypervisor node_data = self.lu.rpc.call_node_info(node_list, cfg.GetVGName(), - hypervisor) + hypervisor_name) node_iinfo = self.lu.rpc.call_all_instances_info(node_list, cluster_info.enabled_hypervisors) - for nname in node_list: + for nname, nresult in node_data.items(): + # first fill in static (config-based) values ninfo = cfg.GetNodeInfo(nname) - node_data[nname].Raise() - if not isinstance(node_data[nname].data, dict): - raise errors.OpExecError("Can't get data for node %s" % nname) - remote_info = node_data[nname].data - for attr in ['memory_total', 'memory_free', 'memory_dom0', - 'vg_size', 'vg_free', 'cpu_total']: - if attr not in remote_info: - raise errors.OpExecError("Node '%s' didn't return attribute '%s'" % - (nname, attr)) - try: - remote_info[attr] = int(remote_info[attr]) - except ValueError, err: - raise errors.OpExecError("Node '%s' returned invalid value for '%s':" - " %s" % (nname, attr, str(err))) - # compute memory used by primary instances - i_p_mem = i_p_up_mem = 0 - for iinfo, beinfo in i_list: - if iinfo.primary_node == nname: - i_p_mem += beinfo[constants.BE_MEMORY] - if iinfo.name not in node_iinfo[nname]: - i_used_mem = 0 - else: - i_used_mem = int(node_iinfo[nname][iinfo.name]['memory']) - i_mem_diff = beinfo[constants.BE_MEMORY] - i_used_mem - remote_info['memory_free'] -= max(0, i_mem_diff) - - if iinfo.status == "up": - i_p_up_mem += beinfo[constants.BE_MEMORY] - - # compute memory used by instances pnr = { "tags": list(ninfo.GetTags()), - "total_memory": remote_info['memory_total'], - "reserved_memory": remote_info['memory_dom0'], - "free_memory": remote_info['memory_free'], - "i_pri_memory": i_p_mem, - "i_pri_up_memory": i_p_up_mem, - "total_disk": remote_info['vg_size'], - "free_disk": remote_info['vg_free'], "primary_ip": ninfo.primary_ip, "secondary_ip": ninfo.secondary_ip, - "total_cpus": remote_info['cpu_total'], "offline": ninfo.offline, + "master_candidate": ninfo.master_candidate, } + + if not ninfo.offline: + nresult.Raise() + if not isinstance(nresult.data, dict): + raise errors.OpExecError("Can't get data for node %s" % nname) + remote_info = nresult.data + for attr in ['memory_total', 'memory_free', 'memory_dom0', + 'vg_size', 'vg_free', 'cpu_total']: + if attr not in remote_info: + raise errors.OpExecError("Node '%s' didn't return attribute" + " '%s'" % (nname, attr)) + try: + remote_info[attr] = int(remote_info[attr]) + except ValueError, err: + raise errors.OpExecError("Node '%s' returned invalid value" + " for '%s': %s" % (nname, attr, err)) + # compute memory used by primary instances + i_p_mem = i_p_up_mem = 0 + for iinfo, beinfo in i_list: + if iinfo.primary_node == nname: + i_p_mem += beinfo[constants.BE_MEMORY] + if iinfo.name not in node_iinfo[nname].data: + i_used_mem = 0 + else: + i_used_mem = int(node_iinfo[nname].data[iinfo.name]['memory']) + i_mem_diff = beinfo[constants.BE_MEMORY] - i_used_mem + remote_info['memory_free'] -= max(0, i_mem_diff) + + if iinfo.admin_up: + i_p_up_mem += beinfo[constants.BE_MEMORY] + + # compute memory used by instances + pnr_dyn = { + "total_memory": remote_info['memory_total'], + "reserved_memory": remote_info['memory_dom0'], + "free_memory": remote_info['memory_free'], + "total_disk": remote_info['vg_size'], + "free_disk": remote_info['vg_free'], + "total_cpus": remote_info['cpu_total'], + "i_pri_memory": i_p_mem, + "i_pri_up_memory": i_p_up_mem, + } + pnr.update(pnr_dyn) + node_results[nname] = pnr data["nodes"] = node_results @@ -6027,13 +6463,13 @@ class IAllocator(object): for n in iinfo.nics] pir = { "tags": list(iinfo.GetTags()), - "should_run": iinfo.status == "up", + "admin_up": iinfo.admin_up, "vcpus": beinfo[constants.BE_VCPUS], "memory": beinfo[constants.BE_MEMORY], "os": iinfo.os, "nodes": [iinfo.primary_node] + list(iinfo.secondary_nodes), "nics": nic_data, - "disks": [{"size": dsk.size, "mode": "w"} for dsk in iinfo.disks], + "disks": [{"size": dsk.size, "mode": dsk.mode} for dsk in iinfo.disks], "disk_template": iinfo.disk_template, "hypervisor": iinfo.hypervisor, } @@ -6054,8 +6490,6 @@ class IAllocator(object): """ data = self.in_data - if len(self.disks) != 2: - raise errors.OpExecError("Only two-disk configurations supported") disk_space = _ComputeDiskSize(self.disk_template, self.disks) @@ -6212,8 +6646,6 @@ class LUTestAllocator(NoHooksLU): " 'nics' parameter") if not isinstance(self.op.disks, list): raise errors.OpPrereqError("Invalid parameter 'disks'") - if len(self.op.disks) != 2: - raise errors.OpPrereqError("Only two-disk configurations supported") for row in self.op.disks: if (not isinstance(row, dict) or "size" not in row or @@ -6222,7 +6654,7 @@ class LUTestAllocator(NoHooksLU): row["mode"] not in ['r', 'w']): raise errors.OpPrereqError("Invalid contents of the" " 'disks' parameter") - if self.op.hypervisor is None: + if not hasattr(self.op, "hypervisor") or self.op.hypervisor is None: self.op.hypervisor = self.cfg.GetHypervisorType() elif self.op.mode == constants.IALLOCATOR_MODE_RELOC: if not hasattr(self.op, "name"):