/lib/cmdlib/__init__.py - snf-ganeti - Greek Research and Technology Network's projects

| Branch: | Tag: | Revision:

root / lib / cmdlib / init.py @ 31b836b8

History | View | Annotate | Download (340 kB)

+      #
+      #
       # Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 Google Inc.
+      #
       # This program is free software; you can redistribute it and/or modify
       # it under the terms of the GNU General Public License as published by
       # the Free Software Foundation; either version 2 of the License, or
       # (at your option) any later version.
+      #
       # This program is distributed in the hope that it will be useful, but
       # WITHOUT ANY WARRANTY; without even the implied warranty of
       # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
       # General Public License for more details.
+      #
       # You should have received a copy of the GNU General Public License
       # along with this program; if not, write to the Free Software
       # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
       # 02110-1301, USA.
       """Module implementing the master-side code."""
       # pylint: disable=W0201,C0302
       # W0201 since most LU attributes are defined in CheckPrereq or similar
       # functions
       # C0302: since we have waaaay too many lines in this module
       import os
       import time
       import logging
       import copy
       import OpenSSL
       import itertools
       import operator
       from ganeti import utils
       from ganeti import errors
       from ganeti import hypervisor
       from ganeti import locking
       from ganeti import constants
       from ganeti import objects
       from ganeti import compat
       from ganeti import masterd
       from ganeti import netutils
       from ganeti import query
       from ganeti import qlang
       from ganeti import opcodes
       from ganeti import ht
       from ganeti import rpc
       from ganeti import pathutils
       from ganeti import network
       from ganeti.masterd import iallocator
       from ganeti.cmdlib.base import ResultWithJobs, LogicalUnit, NoHooksLU, \
         Tasklet, _QueryBase
       from ganeti.cmdlib.common import INSTANCE_DOWN, INSTANCE_ONLINE, \
         INSTANCE_NOT_RUNNING, CAN_CHANGE_INSTANCE_OFFLINE, \
         _ExpandInstanceName, _ExpandItemName, \
         _ExpandNodeName, _ShareAll, _CheckNodeGroupInstances, _GetWantedNodes, \
         _GetWantedInstances, _RunPostHook, _RedistributeAncillaryFiles, \
         _MergeAndVerifyHvState, _MergeAndVerifyDiskState, _GetUpdatedIPolicy, \
         _ComputeNewInstanceViolations, _GetUpdatedParams, _CheckOSParams, \
         _CheckHVParams, _AdjustCandidatePool, _CheckNodePVs, \
         _ComputeIPolicyInstanceViolation, _AnnotateDiskParams, _SupportsOob, \
         _ComputeIPolicySpecViolation, _GetDefaultIAllocator, \
         _CheckInstancesNodeGroups, _LoadNodeEvacResult, _MapInstanceDisksToNodes, \
         _CheckInstanceNodeGroups, _CheckParamsNotGlobal, \
         _IsExclusiveStorageEnabledNode, _CheckInstanceState, \
         _CheckIAllocatorOrNode, _FindFaultyInstanceDisks
       from ganeti.cmdlib.cluster import LUClusterActivateMasterIp, \
         LUClusterDeactivateMasterIp, LUClusterConfigQuery, LUClusterDestroy, \
         LUClusterPostInit, _ClusterQuery, LUClusterQuery, LUClusterRedistConf, \
         LUClusterRename, LUClusterRepairDiskSizes, LUClusterSetParams, \
         LUClusterVerify, LUClusterVerifyConfig, LUClusterVerifyGroup, \
         LUClusterVerifyDisks
       from ganeti.cmdlib.group import LUGroupAdd, LUGroupAssignNodes, \
         _GroupQuery, LUGroupQuery, LUGroupSetParams, LUGroupRemove, \
         LUGroupRename, LUGroupEvacuate, LUGroupVerifyDisks
       from ganeti.cmdlib.node import LUNodeAdd, LUNodeSetParams, \
         LUNodePowercycle, LUNodeEvacuate, LUNodeMigrate, LUNodeModifyStorage, \
         _NodeQuery, LUNodeQuery, LUNodeQueryvols, LUNodeQueryStorage, \
         LUNodeRemove, LURepairNodeStorage
       from ganeti.cmdlib.tags import LUTagsGet, LUTagsSearch, LUTagsSet, LUTagsDel
       from ganeti.cmdlib.network import LUNetworkAdd, LUNetworkRemove, \
         LUNetworkSetParams, _NetworkQuery, LUNetworkQuery, LUNetworkConnect, \
         LUNetworkDisconnect
       from ganeti.cmdlib.test import LUTestDelay, LUTestJqueue, LUTestAllocator
       import ganeti.masterd.instance # pylint: disable=W0611
       def _IsExclusiveStorageEnabledNodeName(cfg, nodename):
         """Whether exclusive_storage is in effect for the given node.
         @type cfg: L{config.ConfigWriter}
         @param cfg: The cluster configuration
         @type nodename: string
         @param nodename: The node
         @rtype: bool
         @return: The effective value of exclusive_storage
         @raise errors.OpPrereqError: if no node exists with the given name
         """
         ni = cfg.GetNodeInfo(nodename)
         if ni is None:
           raise errors.OpPrereqError("Invalid node name %s" % nodename,
                                      errors.ECODE_NOENT)
         return _IsExclusiveStorageEnabledNode(cfg, ni)
       def _CopyLockList(names):
         """Makes a copy of a list of lock names.
         Handles L{locking.ALL_SET} correctly.
         """
         if names == locking.ALL_SET:
           return locking.ALL_SET
         else:
           return names[:]
       def _ReleaseLocks(lu, level, names=None, keep=None):
         """Releases locks owned by an LU.
         @type lu: L{LogicalUnit}
         @param level: Lock level
         @type names: list or None
         @param names: Names of locks to release
         @type keep: list or None
         @param keep: Names of locks to retain
         """
         assert not (keep is not None and names is not None), \
                "Only one of the 'names' and the 'keep' parameters can be given"
         if names is not None:
           should_release = names.__contains__
         elif keep:
           should_release = lambda name: name not in keep
         else:
           should_release = None
         owned = lu.owned_locks(level)
         if not owned:
           # Not owning any lock at this level, do nothing
           pass
         elif should_release:
           retain = []
           release = []
           # Determine which locks to release
           for name in owned:
             if should_release(name):
               release.append(name)
             else:
               retain.append(name)
           assert len(lu.owned_locks(level)) == (len(retain) + len(release))
           # Release just some locks
           lu.glm.release(level, names=release)
           assert frozenset(lu.owned_locks(level)) == frozenset(retain)
         else:
           # Release everything
           lu.glm.release(level)
           assert not lu.glm.is_owned(level), "No locks should be owned"
       def _CheckNodeOnline(lu, node, msg=None):
         """Ensure that a given node is online.
         @param lu: the LU on behalf of which we make the check
         @param node: the node to check
         @param msg: if passed, should be a message to replace the default one
         @raise errors.OpPrereqError: if the node is offline
         """
         if msg is None:
           msg = "Can't use offline node"
         if lu.cfg.GetNodeInfo(node).offline:
           raise errors.OpPrereqError("%s: %s" % (msg, node), errors.ECODE_STATE)
       def _CheckNodeNotDrained(lu, node):
         """Ensure that a given node is not drained.
         @param lu: the LU on behalf of which we make the check
         @param node: the node to check
         @raise errors.OpPrereqError: if the node is drained
         """
         if lu.cfg.GetNodeInfo(node).drained:
           raise errors.OpPrereqError("Can't use drained node %s" % node,
                                      errors.ECODE_STATE)
       def _CheckNodeVmCapable(lu, node):
         """Ensure that a given node is vm capable.
         @param lu: the LU on behalf of which we make the check
         @param node: the node to check
         @raise errors.OpPrereqError: if the node is not vm capable
         """
         if not lu.cfg.GetNodeInfo(node).vm_capable:
           raise errors.OpPrereqError("Can't use non-vm_capable node %s" % node,
                                      errors.ECODE_STATE)
       def _CheckNodeHasOS(lu, node, os_name, force_variant):
         """Ensure that a node supports a given OS.
         @param lu: the LU on behalf of which we make the check
         @param node: the node to check
         @param os_name: the OS to query about
         @param force_variant: whether to ignore variant errors
         @raise errors.OpPrereqError: if the node is not supporting the OS
         """
         result = lu.rpc.call_os_get(node, os_name)
         result.Raise("OS '%s' not in supported OS list for node %s" %
                      (os_name, node),
                      prereq=True, ecode=errors.ECODE_INVAL)
         if not force_variant:
           _CheckOSVariant(result.payload, os_name)
       def _GetClusterDomainSecret():
         """Reads the cluster domain secret.
         """
         return utils.ReadOneLineFile(pathutils.CLUSTER_DOMAIN_SECRET_FILE,
                                      strict=True)
       def _ComputeIPolicyInstanceSpecViolation(
         ipolicy, instance_spec, disk_template,
         _compute_fn=_ComputeIPolicySpecViolation):
         """Compute if instance specs meets the specs of ipolicy.
         @type ipolicy: dict
         @param ipolicy: The ipolicy to verify against
         @param instance_spec: dict
         @param instance_spec: The instance spec to verify
         @type disk_template: string
         @param disk_template: the disk template of the instance
         @param _compute_fn: The function to verify ipolicy (unittest only)
         @see: L{_ComputeIPolicySpecViolation}
         """
         mem_size = instance_spec.get(constants.ISPEC_MEM_SIZE, None)
         cpu_count = instance_spec.get(constants.ISPEC_CPU_COUNT, None)
         disk_count = instance_spec.get(constants.ISPEC_DISK_COUNT, 0)
         disk_sizes = instance_spec.get(constants.ISPEC_DISK_SIZE, [])
         nic_count = instance_spec.get(constants.ISPEC_NIC_COUNT, 0)
         spindle_use = instance_spec.get(constants.ISPEC_SPINDLE_USE, None)
         return _compute_fn(ipolicy, mem_size, cpu_count, disk_count, nic_count,
                            disk_sizes, spindle_use, disk_template)
       def _ComputeIPolicyNodeViolation(ipolicy, instance, current_group,
                                        target_group, cfg,
                                        _compute_fn=_ComputeIPolicyInstanceViolation):
         """Compute if instance meets the specs of the new target group.
         @param ipolicy: The ipolicy to verify
         @param instance: The instance object to verify
         @param current_group: The current group of the instance
         @param target_group: The new group of the instance
         @type cfg: L{config.ConfigWriter}
         @param cfg: Cluster configuration
         @param _compute_fn: The function to verify ipolicy (unittest only)
         @see: L{_ComputeIPolicySpecViolation}
         """
         if current_group == target_group:
           return []
         else:
           return _compute_fn(ipolicy, instance, cfg)
       def _CheckTargetNodeIPolicy(lu, ipolicy, instance, node, cfg, ignore=False,
                                   _compute_fn=_ComputeIPolicyNodeViolation):
         """Checks that the target node is correct in terms of instance policy.
         @param ipolicy: The ipolicy to verify
         @param instance: The instance object to verify
         @param node: The new node to relocate
         @type cfg: L{config.ConfigWriter}
         @param cfg: Cluster configuration
         @param ignore: Ignore violations of the ipolicy
         @param _compute_fn: The function to verify ipolicy (unittest only)
         @see: L{_ComputeIPolicySpecViolation}
         """
         primary_node = lu.cfg.GetNodeInfo(instance.primary_node)
         res = _compute_fn(ipolicy, instance, primary_node.group, node.group, cfg)
         if res:
           msg = ("Instance does not meet target node group's (%s) instance"
                  " policy: %s") % (node.group, utils.CommaJoin(res))
           if ignore:
             lu.LogWarning(msg)
           else:
             raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
       def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
                                 minmem, maxmem, vcpus, nics, disk_template, disks,
                                 bep, hvp, hypervisor_name, tags):
         """Builds instance related env variables for hooks
         This builds the hook environment from individual variables.
         @type name: string
         @param name: the name of the instance
         @type primary_node: string
         @param primary_node: the name of the instance's primary node
         @type secondary_nodes: list
         @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 instance
         @type minmem: string
         @param minmem: the minimum memory size of the instance
         @type maxmem: string
         @param maxmem: the maximum memory size of the instance
         @type vcpus: string
         @param vcpus: the count of VCPUs the instance has
         @type nics: list
         @param nics: list of tuples (name, uuid, ip, mac, mode, link, net, netinfo)
             representing the NICs the instance has
         @type disk_template: string
         @param disk_template: the disk template of the instance
         @type disks: list
         @param disks: list of tuples (name, uuid, size, mode)
         @type bep: dict
         @param bep: the backend parameters for the instance
         @type hvp: dict
         @param hvp: the hypervisor parameters for the instance
         @type hypervisor_name: string
         @param hypervisor_name: the hypervisor for the instance
         @type tags: list
         @param tags: list of instance tags as strings
         @rtype: dict
         @return: the hook environment for this instance
         """
         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_MINMEM": minmem,
           "INSTANCE_MAXMEM": maxmem,
           # TODO(2.9) remove deprecated "memory" value
           "INSTANCE_MEMORY": maxmem,
           "INSTANCE_VCPUS": vcpus,
           "INSTANCE_DISK_TEMPLATE": disk_template,
           "INSTANCE_HYPERVISOR": hypervisor_name,
+        }
         if nics:
           nic_count = len(nics)
           for idx, (name, _, ip, mac, mode, link, net, netinfo) in enumerate(nics):
             if ip is None:
               ip = ""
             env["INSTANCE_NIC%d_NAME" % idx] = name
             env["INSTANCE_NIC%d_IP" % idx] = ip
             env["INSTANCE_NIC%d_MAC" % idx] = mac
             env["INSTANCE_NIC%d_MODE" % idx] = mode
             env["INSTANCE_NIC%d_LINK" % idx] = link
             if netinfo:
               nobj = objects.Network.FromDict(netinfo)
               env.update(nobj.HooksDict("INSTANCE_NIC%d_" % idx))
             elif network:
               # FIXME: broken network reference: the instance NIC specifies a
               # network, but the relevant network entry was not in the config. This
               # should be made impossible.
               env["INSTANCE_NIC%d_NETWORK_NAME" % idx] = net
             if mode == constants.NIC_MODE_BRIDGED:
               env["INSTANCE_NIC%d_BRIDGE" % idx] = link
         else:
           nic_count = 0
         env["INSTANCE_NIC_COUNT"] = nic_count
         if disks:
           disk_count = len(disks)
           for idx, (name, size, mode) in enumerate(disks):
             env["INSTANCE_DISK%d_NAME" % idx] = name
             env["INSTANCE_DISK%d_SIZE" % idx] = size
             env["INSTANCE_DISK%d_MODE" % idx] = mode
         else:
           disk_count = 0
         env["INSTANCE_DISK_COUNT"] = disk_count
         if not tags:
           tags = []
         env["INSTANCE_TAGS"] = " ".join(tags)
         for source, kind in [(bep, "BE"), (hvp, "HV")]:
           for key, value in source.items():
             env["INSTANCE_%s_%s" % (kind, key)] = value
         return env
       def _NICToTuple(lu, nic):
         """Build a tupple of nic information.
         @type lu:  L{LogicalUnit}
         @param lu: the logical unit on whose behalf we execute
         @type nic: L{objects.NIC}
         @param nic: nic to convert to hooks tuple
         """
         cluster = lu.cfg.GetClusterInfo()
         filled_params = cluster.SimpleFillNIC(nic.nicparams)
         mode = filled_params[constants.NIC_MODE]
         link = filled_params[constants.NIC_LINK]
         netinfo = None
         if nic.network:
           nobj = lu.cfg.GetNetwork(nic.network)
           netinfo = objects.Network.ToDict(nobj)
         return (nic.name, nic.uuid, nic.ip, nic.mac, mode, link, nic.network, netinfo)
       def _NICListToTuple(lu, nics):
         """Build a list of nic information tuples.
         This list is suitable to be passed to _BuildInstanceHookEnv or as a return
         value in LUInstanceQueryData.
         @type lu:  L{LogicalUnit}
         @param lu: the logical unit on whose behalf we execute
         @type nics: list of L{objects.NIC}
         @param nics: list of nics to convert to hooks tuples
         """
         hooks_nics = []
         for nic in nics:
           hooks_nics.append(_NICToTuple(lu, nic))
         return hooks_nics
       def _BuildInstanceHookEnvByObject(lu, instance, override=None):
         """Builds instance related env variables for hooks from an object.
         @type lu: L{LogicalUnit}
         @param lu: the logical unit on whose behalf we execute
         @type instance: L{objects.Instance}
         @param instance: the instance for which we should build the
             environment
         @type override: dict
         @param override: dictionary with key/values that will override
             our values
         @rtype: dict
         @return: the hook environment dictionary
         """
         cluster = lu.cfg.GetClusterInfo()
         bep = cluster.FillBE(instance)
         hvp = cluster.FillHV(instance)
         args = {
           "name": instance.name,
           "primary_node": instance.primary_node,
           "secondary_nodes": instance.secondary_nodes,
           "os_type": instance.os,
           "status": instance.admin_state,
           "maxmem": bep[constants.BE_MAXMEM],
           "minmem": bep[constants.BE_MINMEM],
           "vcpus": bep[constants.BE_VCPUS],
           "nics": _NICListToTuple(lu, instance.nics),
           "disk_template": instance.disk_template,
           "disks": [(disk.name, disk.size, disk.mode)
                     for disk in instance.disks],
           "bep": bep,
           "hvp": hvp,
           "hypervisor_name": instance.hypervisor,
           "tags": instance.tags,
+        }
         if override:
           args.update(override)
         return _BuildInstanceHookEnv(**args) # pylint: disable=W0142
       def _CheckNicsBridgesExist(lu, target_nics, target_node):
         """Check that the brigdes needed by a list of nics exist.
         """
         cluster = lu.cfg.GetClusterInfo()
         paramslist = [cluster.SimpleFillNIC(nic.nicparams) for nic in target_nics]
         brlist = [params[constants.NIC_LINK] for params in paramslist
                   if params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED]
         if brlist:
           result = lu.rpc.call_bridges_exist(target_node, brlist)
           result.Raise("Error checking bridges on destination node '%s'" %
                        target_node, prereq=True, ecode=errors.ECODE_ENVIRON)
       def _CheckInstanceBridgesExist(lu, instance, node=None):
         """Check that the brigdes needed by an instance exist.
         """
         if node is None:
           node = instance.primary_node
         _CheckNicsBridgesExist(lu, instance.nics, node)
       def _CheckOSVariant(os_obj, name):
         """Check whether an OS name conforms to the os variants specification.
         @type os_obj: L{objects.OS}
         @param os_obj: OS object to check
         @type name: string
         @param name: OS name passed by the user, to check for validity
         """
         variant = objects.OS.GetVariant(name)
         if not os_obj.supported_variants:
           if variant:
             raise errors.OpPrereqError("OS '%s' doesn't support variants ('%s'"
                                        " passed)" % (os_obj.name, variant),
                                        errors.ECODE_INVAL)
           return
         if not variant:
           raise errors.OpPrereqError("OS name must include a variant",
                                      errors.ECODE_INVAL)
         if variant not in os_obj.supported_variants:
           raise errors.OpPrereqError("Unsupported OS variant", errors.ECODE_INVAL)
       def _CheckHostnameSane(lu, name):
         """Ensures that a given hostname resolves to a 'sane' name.
         The given name is required to be a prefix of the resolved hostname,
         to prevent accidental mismatches.
         @param lu: the logical unit on behalf of which we're checking
         @param name: the name we should resolve and check
         @return: the resolved hostname object
         """
         hostname = netutils.GetHostname(name=name)
         if hostname.name != name:
           lu.LogInfo("Resolved given name '%s' to '%s'", name, hostname.name)
         if not utils.MatchNameComponent(name, [hostname.name]):
           raise errors.OpPrereqError(("Resolved hostname '%s' does not look the"
                                       " same as given hostname '%s'") %
                                       (hostname.name, name), errors.ECODE_INVAL)
         return hostname
       def _WaitForSync(lu, instance, disks=None, oneshot=False):
         """Sleep and poll for an instance's disk to sync.
         """
         if not instance.disks or disks is not None and not disks:
           return True
         disks = _ExpandCheckDisks(instance, disks)
         if not oneshot:
           lu.LogInfo("Waiting for instance %s to sync disks", instance.name)
         node = instance.primary_node
         for dev in disks:
           lu.cfg.SetDiskID(dev, node)
         # TODO: Convert to utils.Retry
         retries = 0
         degr_retries = 10 # in seconds, as we sleep 1 second each time
         while True:
           max_time = 0
           done = True
           cumul_degraded = False
           rstats = lu.rpc.call_blockdev_getmirrorstatus(node, (disks, instance))
           msg = rstats.fail_msg
           if msg:
             lu.LogWarning("Can't get any data from node %s: %s", node, msg)
             retries += 1
             if retries >= 10:
               raise errors.RemoteError("Can't contact node %s for mirror data,"
                                        " aborting." % node)
             time.sleep(6)
             continue
           rstats = rstats.payload
           retries = 0
           for i, mstat in enumerate(rstats):
             if mstat is None:
               lu.LogWarning("Can't compute data for node %s/%s",
                                  node, disks[i].iv_name)
               continue
             cumul_degraded = (cumul_degraded or
                               (mstat.is_degraded and mstat.sync_percent is None))
             if mstat.sync_percent is not None:
               done = False
               if mstat.estimated_time is not None:
                 rem_time = ("%s remaining (estimated)" %
                             utils.FormatSeconds(mstat.estimated_time))
                 max_time = mstat.estimated_time
               else:
                 rem_time = "no time estimate"
               lu.LogInfo("- device %s: %5.2f%% done, %s",
                          disks[i].iv_name, mstat.sync_percent, rem_time)
           # if we're done but degraded, let's do a few small retries, to
           # make sure we see a stable and not transient situation; therefore
           # we force restart of the loop
           if (done or oneshot) and cumul_degraded and degr_retries > 0:
             logging.info("Degraded disks found, %d retries left", degr_retries)
             degr_retries -= 1
             time.sleep(1)
             continue
           if done or oneshot:
             break
           time.sleep(min(60, max_time))
         if done:
           lu.LogInfo("Instance %s's disks are in sync", instance.name)
         return not cumul_degraded
       def _BlockdevFind(lu, node, dev, instance):
         """Wrapper around call_blockdev_find to annotate diskparams.
         @param lu: A reference to the lu object
         @param node: The node to call out
         @param dev: The device to find
         @param instance: The instance object the device belongs to
         @returns The result of the rpc call
         """
         (disk,) = _AnnotateDiskParams(instance, [dev], lu.cfg)
         return lu.rpc.call_blockdev_find(node, disk)
       def _CheckDiskConsistency(lu, instance, dev, node, on_primary, ldisk=False):
         """Wrapper around L{_CheckDiskConsistencyInner}.
         """
         (disk,) = _AnnotateDiskParams(instance, [dev], lu.cfg)
         return _CheckDiskConsistencyInner(lu, instance, disk, node, on_primary,
                                           ldisk=ldisk)
       def _CheckDiskConsistencyInner(lu, instance, dev, node, on_primary,
                                      ldisk=False):
         """Check that mirrors are not degraded.
         @attention: The device has to be annotated already.
         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).
         """
         lu.cfg.SetDiskID(dev, node)
         result = True
         if on_primary or dev.AssembleOnSecondary():
           rstats = lu.rpc.call_blockdev_find(node, dev)
           msg = rstats.fail_msg
           if msg:
             lu.LogWarning("Can't find disk on node %s: %s", node, msg)
             result = False
           elif not rstats.payload:
             lu.LogWarning("Can't find disk on node %s", node)
             result = False
           else:
             if ldisk:
               result = result and rstats.payload.ldisk_status == constants.LDS_OKAY
             else:
               result = result and not rstats.payload.is_degraded
         if dev.children:
           for child in dev.children:
             result = result and _CheckDiskConsistencyInner(lu, instance, child, node,
                                                            on_primary)
         return result
       class LUOobCommand(NoHooksLU):
         """Logical unit for OOB handling.
         """
         REQ_BGL = False
         _SKIP_MASTER = (constants.OOB_POWER_OFF, constants.OOB_POWER_CYCLE)
         def ExpandNames(self):
           """Gather locks we need.
           """
           if self.op.node_names:
             self.op.node_names = _GetWantedNodes(self, self.op.node_names)
             lock_names = self.op.node_names
           else:
             lock_names = locking.ALL_SET
           self.needed_locks = {
             locking.LEVEL_NODE: lock_names,
+            }
           self.share_locks[locking.LEVEL_NODE_ALLOC] = 1
           if not self.op.node_names:
             # Acquire node allocation lock only if all nodes are affected
             self.needed_locks[locking.LEVEL_NODE_ALLOC] = locking.ALL_SET
         def CheckPrereq(self):
           """Check prerequisites.
           This checks:
            - the node exists in the configuration
            - OOB is supported
           Any errors are signaled by raising errors.OpPrereqError.
           """
           self.nodes = []
           self.master_node = self.cfg.GetMasterNode()
           assert self.op.power_delay >= 0.0
           if self.op.node_names:
             if (self.op.command in self._SKIP_MASTER and
                 self.master_node in self.op.node_names):
               master_node_obj = self.cfg.GetNodeInfo(self.master_node)
               master_oob_handler = _SupportsOob(self.cfg, master_node_obj)
               if master_oob_handler:
                 additional_text = ("run '%s %s %s' if you want to operate on the"
                                    " master regardless") % (master_oob_handler,
                                                             self.op.command,
                                                             self.master_node)
               else:
                 additional_text = "it does not support out-of-band operations"
               raise errors.OpPrereqError(("Operating on the master node %s is not"
                                           " allowed for %s; %s") %
                                          (self.master_node, self.op.command,
                                           additional_text), errors.ECODE_INVAL)
           else:
             self.op.node_names = self.cfg.GetNodeList()
             if self.op.command in self._SKIP_MASTER:
               self.op.node_names.remove(self.master_node)
           if self.op.command in self._SKIP_MASTER:
             assert self.master_node not in self.op.node_names
           for (node_name, node) in self.cfg.GetMultiNodeInfo(self.op.node_names):
             if node is None:
               raise errors.OpPrereqError("Node %s not found" % node_name,
                                          errors.ECODE_NOENT)
             else:
               self.nodes.append(node)
             if (not self.op.ignore_status and
                 (self.op.command == constants.OOB_POWER_OFF and not node.offline)):
               raise errors.OpPrereqError(("Cannot power off node %s because it is"
                                           " not marked offline") % node_name,
                                          errors.ECODE_STATE)
         def Exec(self, feedback_fn):
           """Execute OOB and return result if we expect any.
           """
           master_node = self.master_node
           ret = []
           for idx, node in enumerate(utils.NiceSort(self.nodes,
                                                     key=lambda node: node.name)):
             node_entry = [(constants.RS_NORMAL, node.name)]
             ret.append(node_entry)
             oob_program = _SupportsOob(self.cfg, node)
             if not oob_program:
               node_entry.append((constants.RS_UNAVAIL, None))
               continue
             logging.info("Executing out-of-band command '%s' using '%s' on %s",
                          self.op.command, oob_program, node.name)
             result = self.rpc.call_run_oob(master_node, oob_program,
                                            self.op.command, node.name,
                                            self.op.timeout)
             if result.fail_msg:
               self.LogWarning("Out-of-band RPC failed on node '%s': %s",
                               node.name, result.fail_msg)
               node_entry.append((constants.RS_NODATA, None))
             else:
               try:
                 self._CheckPayload(result)
               except errors.OpExecError, err:
                 self.LogWarning("Payload returned by node '%s' is not valid: %s",
                                 node.name, err)
                 node_entry.append((constants.RS_NODATA, None))
               else:
                 if self.op.command == constants.OOB_HEALTH:
                   # For health we should log important events
                   for item, status in result.payload:
                     if status in [constants.OOB_STATUS_WARNING,
                                   constants.OOB_STATUS_CRITICAL]:
                       self.LogWarning("Item '%s' on node '%s' has status '%s'",
                                       item, node.name, status)
                 if self.op.command == constants.OOB_POWER_ON:
                   node.powered = True
                 elif self.op.command == constants.OOB_POWER_OFF:
                   node.powered = False
                 elif self.op.command == constants.OOB_POWER_STATUS:
                   powered = result.payload[constants.OOB_POWER_STATUS_POWERED]
                   if powered != node.powered:
                     logging.warning(("Recorded power state (%s) of node '%s' does not"
                                      " match actual power state (%s)"), node.powered,
                                     node.name, powered)
                 # For configuration changing commands we should update the node
                 if self.op.command in (constants.OOB_POWER_ON,
                                        constants.OOB_POWER_OFF):
                   self.cfg.Update(node, feedback_fn)
                 node_entry.append((constants.RS_NORMAL, result.payload))
                 if (self.op.command == constants.OOB_POWER_ON and
                     idx < len(self.nodes) - 1):
                   time.sleep(self.op.power_delay)
           return ret
         def _CheckPayload(self, result):
           """Checks if the payload is valid.
           @param result: RPC result
           @raises errors.OpExecError: If payload is not valid
           """
           errs = []
           if self.op.command == constants.OOB_HEALTH:
             if not isinstance(result.payload, list):
               errs.append("command 'health' is expected to return a list but got %s" %
                           type(result.payload))
             else:
               for item, status in result.payload:
                 if status not in constants.OOB_STATUSES:
                   errs.append("health item '%s' has invalid status '%s'" %
                               (item, status))
           if self.op.command == constants.OOB_POWER_STATUS:
             if not isinstance(result.payload, dict):
               errs.append("power-status is expected to return a dict but got %s" %
                           type(result.payload))
           if self.op.command in [
             constants.OOB_POWER_ON,
             constants.OOB_POWER_OFF,
             constants.OOB_POWER_CYCLE,
             ]:
             if result.payload is not None:
               errs.append("%s is expected to not return payload but got '%s'" %
                           (self.op.command, result.payload))
           if errs:
             raise errors.OpExecError("Check of out-of-band payload failed due to %s" %
                                      utils.CommaJoin(errs))
       class _OsQuery(_QueryBase):
         FIELDS = query.OS_FIELDS
         def ExpandNames(self, lu):
           # Lock all nodes in shared mode
           # Temporary removal of locks, should be reverted later
           # TODO: reintroduce locks when they are lighter-weight
           lu.needed_locks = {}
           #self.share_locks[locking.LEVEL_NODE] = 1
           #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
           # The following variables interact with _QueryBase._GetNames
           if self.names:
             self.wanted = self.names
           else:
             self.wanted = locking.ALL_SET
           self.do_locking = self.use_locking
         def DeclareLocks(self, lu, level):
           pass
         @staticmethod
         def _DiagnoseByOS(rlist):
           """Remaps a per-node return list into an a per-os per-node dictionary
           @param rlist: a map with node names as keys and OS objects as values
           @rtype: dict
           @return: a dictionary with osnames as keys and as value another
               map, with nodes as keys and tuples of (path, status, diagnose,
               variants, parameters, api_versions) as values, eg::
                 {"debian-etch": {"node1": [(/usr/lib/..., True, "", [], []),
                                            (/srv/..., False, "invalid api")],
                                  "node2": [(/srv/..., True, "", [], [])]}
+                }
           """
           all_os = {}
           # we build here the list of nodes that didn't fail the RPC (at RPC
           # level), so that nodes with a non-responding node daemon don't
           # make all OSes invalid
           good_nodes = [node_name for node_name in rlist
                         if not rlist[node_name].fail_msg]
           for node_name, nr in rlist.items():
             if nr.fail_msg or not nr.payload:
               continue
             for (name, path, status, diagnose, variants,
                  params, api_versions) in nr.payload:
               if name not in all_os:
                 # build a list of nodes for this os containing empty lists
                 # for each node in node_list
                 all_os[name] = {}
                 for nname in good_nodes:
                   all_os[name][nname] = []
               # convert params from [name, help] to (name, help)
               params = [tuple(v) for v in params]
               all_os[name][node_name].append((path, status, diagnose,
                                               variants, params, api_versions))
           return all_os
         def _GetQueryData(self, lu):
           """Computes the list of nodes and their attributes.
           """
           # Locking is not used
           assert not (compat.any(lu.glm.is_owned(level)
                                  for level in locking.LEVELS
                                  if level != locking.LEVEL_CLUSTER) or
                       self.do_locking or self.use_locking)
           valid_nodes = [node.name
                          for node in lu.cfg.GetAllNodesInfo().values()
                          if not node.offline and node.vm_capable]
           pol = self._DiagnoseByOS(lu.rpc.call_os_diagnose(valid_nodes))
           cluster = lu.cfg.GetClusterInfo()
           data = {}
           for (os_name, os_data) in pol.items():
             info = query.OsInfo(name=os_name, valid=True, node_status=os_data,
                                 hidden=(os_name in cluster.hidden_os),
                                 blacklisted=(os_name in cluster.blacklisted_os))
             variants = set()
             parameters = set()
             api_versions = set()
             for idx, osl in enumerate(os_data.values()):
               info.valid = bool(info.valid and osl and osl[0][1])
               if not info.valid:
                 break
               (node_variants, node_params, node_api) = osl[0][3:6]
               if idx == 0:
                 # First entry
                 variants.update(node_variants)
                 parameters.update(node_params)
                 api_versions.update(node_api)
               else:
                 # Filter out inconsistent values
                 variants.intersection_update(node_variants)
                 parameters.intersection_update(node_params)
                 api_versions.intersection_update(node_api)
             info.variants = list(variants)
             info.parameters = list(parameters)
             info.api_versions = list(api_versions)
             data[os_name] = info
           # Prepare data in requested order
           return [data[name] for name in self._GetNames(lu, pol.keys(), None)
                   if name in data]
       class LUOsDiagnose(NoHooksLU):
         """Logical unit for OS diagnose/query.
         """
         REQ_BGL = False
         @staticmethod
         def _BuildFilter(fields, names):
           """Builds a filter for querying OSes.
           """
           name_filter = qlang.MakeSimpleFilter("name", names)
           # Legacy behaviour: Hide hidden, blacklisted or invalid OSes if the
           # respective field is not requested
           status_filter = [[qlang.OP_NOT, [qlang.OP_TRUE, fname]]
                            for fname in ["hidden", "blacklisted"]
                            if fname not in fields]
           if "valid" not in fields:
             status_filter.append([qlang.OP_TRUE, "valid"])
           if status_filter:
             status_filter.insert(0, qlang.OP_AND)
           else:
             status_filter = None
           if name_filter and status_filter:
             return [qlang.OP_AND, name_filter, status_filter]
           elif name_filter:
             return name_filter
           else:
             return status_filter
         def CheckArguments(self):
           self.oq = _OsQuery(self._BuildFilter(self.op.output_fields, self.op.names),
                              self.op.output_fields, False)
         def ExpandNames(self):
           self.oq.ExpandNames(self)
         def Exec(self, feedback_fn):
           return self.oq.OldStyleQuery(self)
       class _ExtStorageQuery(_QueryBase):
         FIELDS = query.EXTSTORAGE_FIELDS
         def ExpandNames(self, lu):
           # Lock all nodes in shared mode
           # Temporary removal of locks, should be reverted later
           # TODO: reintroduce locks when they are lighter-weight
           lu.needed_locks = {}
           #self.share_locks[locking.LEVEL_NODE] = 1
           #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
           # The following variables interact with _QueryBase._GetNames
           if self.names:
             self.wanted = self.names
           else:
             self.wanted = locking.ALL_SET
           self.do_locking = self.use_locking
         def DeclareLocks(self, lu, level):
           pass
         @staticmethod
         def _DiagnoseByProvider(rlist):
           """Remaps a per-node return list into an a per-provider per-node dictionary
           @param rlist: a map with node names as keys and ExtStorage objects as values
           @rtype: dict
           @return: a dictionary with extstorage providers as keys and as
               value another map, with nodes as keys and tuples of
               (path, status, diagnose, parameters) as values, eg::
                 {"provider1": {"node1": [(/usr/lib/..., True, "", [])]
                                "node2": [(/srv/..., False, "missing file")]
                                "node3": [(/srv/..., True, "", [])]
+                }
           """
           all_es = {}
           # we build here the list of nodes that didn't fail the RPC (at RPC
           # level), so that nodes with a non-responding node daemon don't
           # make all OSes invalid
           good_nodes = [node_name for node_name in rlist
                         if not rlist[node_name].fail_msg]
           for node_name, nr in rlist.items():
             if nr.fail_msg or not nr.payload:
               continue
             for (name, path, status, diagnose, params) in nr.payload:
               if name not in all_es:
                 # build a list of nodes for this os containing empty lists
                 # for each node in node_list
                 all_es[name] = {}
                 for nname in good_nodes:
                   all_es[name][nname] = []
               # convert params from [name, help] to (name, help)
               params = [tuple(v) for v in params]
               all_es[name][node_name].append((path, status, diagnose, params))
           return all_es
         def _GetQueryData(self, lu):
           """Computes the list of nodes and their attributes.
           """
           # Locking is not used
           assert not (compat.any(lu.glm.is_owned(level)
                                  for level in locking.LEVELS
                                  if level != locking.LEVEL_CLUSTER) or
                       self.do_locking or self.use_locking)
           valid_nodes = [node.name
                          for node in lu.cfg.GetAllNodesInfo().values()
                          if not node.offline and node.vm_capable]
           pol = self._DiagnoseByProvider(lu.rpc.call_extstorage_diagnose(valid_nodes))
           data = {}
           nodegroup_list = lu.cfg.GetNodeGroupList()
           for (es_name, es_data) in pol.items():
             # For every provider compute the nodegroup validity.
             # To do this we need to check the validity of each node in es_data
             # and then construct the corresponding nodegroup dict:
             #      { nodegroup1: status
             #        nodegroup2: status
             #      }
             ndgrp_data = {}
             for nodegroup in nodegroup_list:
               ndgrp = lu.cfg.GetNodeGroup(nodegroup)
               nodegroup_nodes = ndgrp.members
               nodegroup_name = ndgrp.name
               node_statuses = []
               for node in nodegroup_nodes:
                 if node in valid_nodes:
                   if es_data[node] != []:
                     node_status = es_data[node][0][1]
                     node_statuses.append(node_status)
                   else:
                     node_statuses.append(False)
               if False in node_statuses:
                 ndgrp_data[nodegroup_name] = False
               else:
                 ndgrp_data[nodegroup_name] = True
             # Compute the provider's parameters
             parameters = set()
             for idx, esl in enumerate(es_data.values()):
               valid = bool(esl and esl[0][1])
               if not valid:
                 break
               node_params = esl[0][3]
               if idx == 0:
                 # First entry
                 parameters.update(node_params)
               else:
                 # Filter out inconsistent values
                 parameters.intersection_update(node_params)
             params = list(parameters)
             # Now fill all the info for this provider
             info = query.ExtStorageInfo(name=es_name, node_status=es_data,
                                         nodegroup_status=ndgrp_data,
                                         parameters=params)
             data[es_name] = info
           # Prepare data in requested order
           return [data[name] for name in self._GetNames(lu, pol.keys(), None)
                   if name in data]
       class LUExtStorageDiagnose(NoHooksLU):
         """Logical unit for ExtStorage diagnose/query.
         """
         REQ_BGL = False
         def CheckArguments(self):
           self.eq = _ExtStorageQuery(qlang.MakeSimpleFilter("name", self.op.names),
                                      self.op.output_fields, False)
         def ExpandNames(self):
           self.eq.ExpandNames(self)
         def Exec(self, feedback_fn):
           return self.eq.OldStyleQuery(self)
       class _InstanceQuery(_QueryBase):
         FIELDS = query.INSTANCE_FIELDS
         def ExpandNames(self, lu):
           lu.needed_locks = {}
           lu.share_locks = _ShareAll()
           if self.names:
             self.wanted = _GetWantedInstances(lu, self.names)
           else:
             self.wanted = locking.ALL_SET
           self.do_locking = (self.use_locking and
                              query.IQ_LIVE in self.requested_data)
           if self.do_locking:
             lu.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
             lu.needed_locks[locking.LEVEL_NODEGROUP] = []
             lu.needed_locks[locking.LEVEL_NODE] = []
             lu.needed_locks[locking.LEVEL_NETWORK] = []
             lu.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
           self.do_grouplocks = (self.do_locking and
                                 query.IQ_NODES in self.requested_data)
         def DeclareLocks(self, lu, level):
           if self.do_locking:
             if level == locking.LEVEL_NODEGROUP and self.do_grouplocks:
               assert not lu.needed_locks[locking.LEVEL_NODEGROUP]
               # Lock all groups used by instances optimistically; this requires going
               # via the node before it's locked, requiring verification later on
               lu.needed_locks[locking.LEVEL_NODEGROUP] = \
                 set(group_uuid
                     for instance_name in lu.owned_locks(locking.LEVEL_INSTANCE)
                     for group_uuid in lu.cfg.GetInstanceNodeGroups(instance_name))
             elif level == locking.LEVEL_NODE:
               lu._LockInstancesNodes() # pylint: disable=W0212
             elif level == locking.LEVEL_NETWORK:
               lu.needed_locks[locking.LEVEL_NETWORK] = \
                 frozenset(net_uuid
                           for instance_name in lu.owned_locks(locking.LEVEL_INSTANCE)
                           for net_uuid in lu.cfg.GetInstanceNetworks(instance_name))
         @staticmethod
         def _CheckGroupLocks(lu):
           owned_instances = frozenset(lu.owned_locks(locking.LEVEL_INSTANCE))
           owned_groups = frozenset(lu.owned_locks(locking.LEVEL_NODEGROUP))
           # Check if node groups for locked instances are still correct
           for instance_name in owned_instances:
             _CheckInstanceNodeGroups(lu.cfg, instance_name, owned_groups)
         def _GetQueryData(self, lu):
           """Computes the list of instances and their attributes.
           """
           if self.do_grouplocks:
             self._CheckGroupLocks(lu)
           cluster = lu.cfg.GetClusterInfo()
           all_info = lu.cfg.GetAllInstancesInfo()
           instance_names = self._GetNames(lu, all_info.keys(), locking.LEVEL_INSTANCE)
           instance_list = [all_info[name] for name in instance_names]
           nodes = frozenset(itertools.chain(*(inst.all_nodes
                                               for inst in instance_list)))
           hv_list = list(set([inst.hypervisor for inst in instance_list]))
           bad_nodes = []
           offline_nodes = []
           wrongnode_inst = set()
           # Gather data as requested
           if self.requested_data & set([query.IQ_LIVE, query.IQ_CONSOLE]):
             live_data = {}
             node_data = lu.rpc.call_all_instances_info(nodes, hv_list)
             for name in nodes:
               result = node_data[name]
               if result.offline:
                 # offline nodes will be in both lists
                 assert result.fail_msg
                 offline_nodes.append(name)
               if result.fail_msg:
                 bad_nodes.append(name)
               elif result.payload:
                 for inst in result.payload:
                   if inst in all_info:
                     if all_info[inst].primary_node == name:
                       live_data.update(result.payload)
                     else:
                       wrongnode_inst.add(inst)
                   else:
                     # orphan instance; we don't list it here as we don't
                     # handle this case yet in the output of instance listing
                     logging.warning("Orphan instance '%s' found on node %s",
                                     inst, name)
               # else no instance is alive
           else:
             live_data = {}
           if query.IQ_DISKUSAGE in self.requested_data:
             gmi = ganeti.masterd.instance
             disk_usage = dict((inst.name,
                                gmi.ComputeDiskSize(inst.disk_template,
                                                    [{constants.IDISK_SIZE: disk.size}
                                                     for disk in inst.disks]))
                               for inst in instance_list)
           else:
             disk_usage = None
           if query.IQ_CONSOLE in self.requested_data:
             consinfo = {}
             for inst in instance_list:
               if inst.name in live_data:
                 # Instance is running
                 consinfo[inst.name] = _GetInstanceConsole(cluster, inst)
               else:
                 consinfo[inst.name] = None
             assert set(consinfo.keys()) == set(instance_names)
           else:
             consinfo = None
           if query.IQ_NODES in self.requested_data:
             node_names = set(itertools.chain(*map(operator.attrgetter("all_nodes"),
                                                   instance_list)))
             nodes = dict(lu.cfg.GetMultiNodeInfo(node_names))
             groups = dict((uuid, lu.cfg.GetNodeGroup(uuid))
                           for uuid in set(map(operator.attrgetter("group"),
                                               nodes.values())))
           else:
             nodes = None
             groups = None
           if query.IQ_NETWORKS in self.requested_data:
             net_uuids = itertools.chain(*(lu.cfg.GetInstanceNetworks(i.name)
                                           for i in instance_list))
             networks = dict((uuid, lu.cfg.GetNetwork(uuid)) for uuid in net_uuids)
           else:
             networks = None
           return query.InstanceQueryData(instance_list, lu.cfg.GetClusterInfo(),
                                          disk_usage, offline_nodes, bad_nodes,
                                          live_data, wrongnode_inst, consinfo,
                                          nodes, groups, networks)
       class LUQuery(NoHooksLU):
         """Query for resources/items of a certain kind.
         """
         # pylint: disable=W0142
         REQ_BGL = False
         def CheckArguments(self):
           qcls = _GetQueryImplementation(self.op.what)
           self.impl = qcls(self.op.qfilter, self.op.fields, self.op.use_locking)
         def ExpandNames(self):
           self.impl.ExpandNames(self)
         def DeclareLocks(self, level):
           self.impl.DeclareLocks(self, level)
         def Exec(self, feedback_fn):
           return self.impl.NewStyleQuery(self)
       class LUQueryFields(NoHooksLU):
         """Query for resources/items of a certain kind.
         """
         # pylint: disable=W0142
         REQ_BGL = False
         def CheckArguments(self):
           self.qcls = _GetQueryImplementation(self.op.what)
         def ExpandNames(self):
           self.needed_locks = {}
         def Exec(self, feedback_fn):
           return query.QueryFields(self.qcls.FIELDS, self.op.fields)
       class LUInstanceActivateDisks(NoHooksLU):
         """Bring up an instance's disks.
         """
         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 CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster.
           """
           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.instance.primary_node)
         def Exec(self, feedback_fn):
           """Activate the disks.
           """
           disks_ok, disks_info = \
                     _AssembleInstanceDisks(self, self.instance,
                                            ignore_size=self.op.ignore_size)
           if not disks_ok:
             raise errors.OpExecError("Cannot activate block devices")
           if self.op.wait_for_sync:
             if not _WaitForSync(self, self.instance):
               raise errors.OpExecError("Some disks of the instance are degraded!")
           return disks_info
       def _AssembleInstanceDisks(lu, instance, disks=None, ignore_secondaries=False,
                                  ignore_size=False):
         """Prepare the block devices for an instance.
         This sets up the block devices on all nodes.
         @type lu: L{LogicalUnit}
         @param lu: the logical unit on whose behalf we execute
         @type instance: L{objects.Instance}
         @param instance: the instance for whose disks we assemble
         @type disks: list of L{objects.Disk} or None
         @param disks: which disks to assemble (or all, if None)
         @type ignore_secondaries: boolean
         @param ignore_secondaries: if true, errors on secondary nodes
             won't result in an error return from the function
         @type ignore_size: boolean
         @param ignore_size: if true, the current known size of the disk
             will not be used during the disk activation, useful for cases
             when the size is wrong
         @return: False if the operation failed, otherwise a list of
             (host, instance_visible_name, node_visible_name)
             with the mapping from node devices to instance devices
         """
         device_info = []
         disks_ok = True
         iname = instance.name
         disks = _ExpandCheckDisks(instance, disks)
         # 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 idx, inst_disk in enumerate(disks):
           for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
             if ignore_size:
               node_disk = node_disk.Copy()
               node_disk.UnsetSize()
             lu.cfg.SetDiskID(node_disk, node)
             result = lu.rpc.call_blockdev_assemble(node, (node_disk, instance), iname,
                                                    False, idx)
             msg = result.fail_msg
             if msg:
               is_offline_secondary = (node in instance.secondary_nodes and
                                       result.offline)
               lu.LogWarning("Could not prepare block device %s on node %s"
                             " (is_primary=False, pass=1): %s",
                             inst_disk.iv_name, node, msg)
               if not (ignore_secondaries or is_offline_secondary):
                 disks_ok = False
         # FIXME: race condition on drbd migration to primary
         # 2nd pass, do only the primary node
         for idx, inst_disk in enumerate(disks):
           dev_path = None
           for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
             if node != instance.primary_node:
               continue
             if ignore_size:
               node_disk = node_disk.Copy()
               node_disk.UnsetSize()
             lu.cfg.SetDiskID(node_disk, node)
             result = lu.rpc.call_blockdev_assemble(node, (node_disk, instance), iname,
                                                    True, idx)
             msg = result.fail_msg
             if msg:
               lu.LogWarning("Could not prepare block device %s on node %s"
                             " (is_primary=True, pass=2): %s",
                             inst_disk.iv_name, node, msg)
               disks_ok = False
             else:
               dev_path = result.payload
           device_info.append((instance.primary_node, inst_disk.iv_name, dev_path))
         # leave the disks configured for the primary node
         # this is a workaround that would be fixed better by
         # improving the logical/physical id handling
         for disk in disks:
           lu.cfg.SetDiskID(disk, instance.primary_node)
         return disks_ok, device_info
       def _StartInstanceDisks(lu, instance, force):
         """Start the disks of an instance.
         """
         disks_ok, _ = _AssembleInstanceDisks(lu, instance,
                                                  ignore_secondaries=force)
         if not disks_ok:
           _ShutdownInstanceDisks(lu, instance)
           if force is not None and not force:
             lu.LogWarning("",
                           hint=("If the message above refers to a secondary node,"
                                 " you can retry the operation using '--force'"))
           raise errors.OpExecError("Disk consistency error")
       class LUInstanceDeactivateDisks(NoHooksLU):
         """Shutdown an instance's disks.
         """
         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 CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster.
           """
           self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
           assert self.instance is not None, \
             "Cannot retrieve locked instance %s" % self.op.instance_name
         def Exec(self, feedback_fn):
           """Deactivate the disks
           """
           instance = self.instance
           if self.op.force:
             _ShutdownInstanceDisks(self, instance)
           else:
             _SafeShutdownInstanceDisks(self, instance)
       def _SafeShutdownInstanceDisks(lu, instance, disks=None):
         """Shutdown block devices of an instance.
         This function checks if an instance is running, before calling
         _ShutdownInstanceDisks.
         """
         _CheckInstanceState(lu, instance, INSTANCE_DOWN, msg="cannot shutdown disks")
         _ShutdownInstanceDisks(lu, instance, disks=disks)
       def _ExpandCheckDisks(instance, disks):
         """Return the instance disks selected by the disks list
         @type disks: list of L{objects.Disk} or None
         @param disks: selected disks
         @rtype: list of L{objects.Disk}
         @return: selected instance disks to act on
         """
         if disks is None:
           return instance.disks
         else:
           if not set(disks).issubset(instance.disks):
             raise errors.ProgrammerError("Can only act on disks belonging to the"
                                          " target instance")
           return disks
       def _ShutdownInstanceDisks(lu, instance, disks=None, ignore_primary=False):
         """Shutdown block devices of an instance.
         This does the shutdown on all nodes of the instance.
         If the ignore_primary is false, errors on the primary node are
         ignored.
         """
         all_result = True
         disks = _ExpandCheckDisks(instance, disks)
         for disk in disks:
           for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
             lu.cfg.SetDiskID(top_disk, node)
             result = lu.rpc.call_blockdev_shutdown(node, (top_disk, instance))
             msg = result.fail_msg
             if msg:
               lu.LogWarning("Could not shutdown block device %s on node %s: %s",
                             disk.iv_name, node, msg)
               if ((node == instance.primary_node and not ignore_primary) or
                   (node != instance.primary_node and not result.offline)):
                 all_result = False
         return all_result
       def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
         """Checks if a node has enough free memory.
         This function checks 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 raises an OpPrereqError
         exception.
         @type lu: C{LogicalUnit}
         @param lu: a logical unit from which we get configuration data
         @type node: C{str}
         @param node: the node to check
         @type reason: C{str}
         @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_name: C{str}
         @param hypervisor_name: the hypervisor to ask for memory stats
         @rtype: integer
         @return: node current free memory
         @raise errors.OpPrereqError: if the node doesn't have enough memory, or
             we cannot check the node
         """
         nodeinfo = lu.rpc.call_node_info([node], None, [hypervisor_name], False)
         nodeinfo[node].Raise("Can't get data from node %s" % node,
                              prereq=True, ecode=errors.ECODE_ENVIRON)
         (_, _, (hv_info, )) = nodeinfo[node].payload
         free_mem = hv_info.get("memory_free", None)
         if not isinstance(free_mem, int):
           raise errors.OpPrereqError("Can't compute free memory on node %s, result"
                                      " was '%s'" % (node, free_mem),
                                      errors.ECODE_ENVIRON)
         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),
                                      errors.ECODE_NORES)
         return free_mem
       def _CheckNodesFreeDiskPerVG(lu, nodenames, req_sizes):
         """Checks if nodes have enough free disk space in all the VGs.
         This function checks if all given nodes have the needed amount of
         free disk. In case any node has less disk or we cannot get the
         information from the node, this function raises an OpPrereqError
         exception.
         @type lu: C{LogicalUnit}
         @param lu: a logical unit from which we get configuration data
         @type nodenames: C{list}
         @param nodenames: the list of node names to check
         @type req_sizes: C{dict}
         @param req_sizes: the hash of vg and corresponding amount of disk in
             MiB to check for
         @raise errors.OpPrereqError: if the node doesn't have enough disk,
             or we cannot check the node
         """
         for vg, req_size in req_sizes.items():
           _CheckNodesFreeDiskOnVG(lu, nodenames, vg, req_size)
       def _CheckNodesFreeDiskOnVG(lu, nodenames, vg, requested):
         """Checks if nodes have enough free disk space in the specified VG.
         This function checks if all given nodes have the needed amount of
         free disk. In case any node has less disk or we cannot get the
         information from the node, this function raises an OpPrereqError
         exception.
         @type lu: C{LogicalUnit}
         @param lu: a logical unit from which we get configuration data
         @type nodenames: C{list}
         @param nodenames: the list of node names to check
         @type vg: C{str}
         @param vg: the volume group to check
         @type requested: C{int}
         @param requested: the amount of disk in MiB to check for
         @raise errors.OpPrereqError: if the node doesn't have enough disk,
             or we cannot check the node
         """
         es_flags = rpc.GetExclusiveStorageForNodeNames(lu.cfg, nodenames)
         nodeinfo = lu.rpc.call_node_info(nodenames, [vg], None, es_flags)
         for node in nodenames:
           info = nodeinfo[node]
           info.Raise("Cannot get current information from node %s" % node,
                      prereq=True, ecode=errors.ECODE_ENVIRON)
           (_, (vg_info, ), _) = info.payload
           vg_free = vg_info.get("vg_free", None)
           if not isinstance(vg_free, int):
             raise errors.OpPrereqError("Can't compute free disk space on node"
                                        " %s for vg %s, result was '%s'" %
                                        (node, vg, vg_free), errors.ECODE_ENVIRON)
           if requested > vg_free:
             raise errors.OpPrereqError("Not enough disk space on target node %s"
                                        " vg %s: required %d MiB, available %d MiB" %
                                        (node, vg, requested, vg_free),
                                        errors.ECODE_NORES)
       def _CheckNodesPhysicalCPUs(lu, nodenames, requested, hypervisor_name):
         """Checks if nodes have enough physical CPUs
         This function checks if all given nodes have the needed number of
         physical CPUs. In case any node has less CPUs or we cannot get the
         information from the node, this function raises an OpPrereqError
         exception.
         @type lu: C{LogicalUnit}
         @param lu: a logical unit from which we get configuration data
         @type nodenames: C{list}
         @param nodenames: the list of node names to check
         @type requested: C{int}
         @param requested: the minimum acceptable number of physical CPUs
         @raise errors.OpPrereqError: if the node doesn't have enough CPUs,
             or we cannot check the node
         """
         nodeinfo = lu.rpc.call_node_info(nodenames, None, [hypervisor_name], None)
         for node in nodenames:
           info = nodeinfo[node]
           info.Raise("Cannot get current information from node %s" % node,
                      prereq=True, ecode=errors.ECODE_ENVIRON)
           (_, _, (hv_info, )) = info.payload
           num_cpus = hv_info.get("cpu_total", None)
           if not isinstance(num_cpus, int):
             raise errors.OpPrereqError("Can't compute the number of physical CPUs"
                                        " on node %s, result was '%s'" %
                                        (node, num_cpus), errors.ECODE_ENVIRON)
           if requested > num_cpus:
             raise errors.OpPrereqError("Node %s has %s physical CPUs, but %s are "
                                        "required" % (node, num_cpus, requested),
                                        errors.ECODE_NORES)
       class LUInstanceStartup(LogicalUnit):
         """Starts an instance.
         """
         HPATH = "instance-start"
         HTYPE = constants.HTYPE_INSTANCE
         REQ_BGL = False
         def CheckArguments(self):
           # extra beparams
           if self.op.beparams:
             # fill the beparams dict
             objects.UpgradeBeParams(self.op.beparams)
             utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
         def ExpandNames(self):
           self._ExpandAndLockInstance()
           self.recalculate_locks[locking.LEVEL_NODE_RES] = constants.LOCKS_REPLACE
         def DeclareLocks(self, level):
           if level == locking.LEVEL_NODE_RES:
             self._LockInstancesNodes(primary_only=True, level=locking.LEVEL_NODE_RES)
         def BuildHooksEnv(self):
           """Build hooks env.
           This runs on master, primary and secondary nodes of the instance.
           """
           env = {
             "FORCE": self.op.force,
+            }
           env.update(_BuildInstanceHookEnvByObject(self, self.instance))
           return env
         def BuildHooksNodes(self):
           """Build hooks nodes.
           """
           nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
           return (nl, nl)
         def CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster.
           """
           self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
           assert self.instance is not None, \
             "Cannot retrieve locked instance %s" % self.op.instance_name
           # extra hvparams
           if self.op.hvparams:
             # check hypervisor parameter syntax (locally)
             cluster = self.cfg.GetClusterInfo()
             utils.ForceDictType(self.op.hvparams, constants.HVS_PARAMETER_TYPES)
             filled_hvp = cluster.FillHV(instance)
             filled_hvp.update(self.op.hvparams)
             hv_type = hypervisor.GetHypervisorClass(instance.hypervisor)
             hv_type.CheckParameterSyntax(filled_hvp)
             _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
           _CheckInstanceState(self, instance, INSTANCE_ONLINE)
           self.primary_offline = self.cfg.GetNodeInfo(instance.primary_node).offline
           if self.primary_offline and self.op.ignore_offline_nodes:
             self.LogWarning("Ignoring offline primary node")
             if self.op.hvparams or self.op.beparams:
               self.LogWarning("Overridden parameters are ignored")
           else:
             _CheckNodeOnline(self, instance.primary_node)
             bep = self.cfg.GetClusterInfo().FillBE(instance)
             bep.update(self.op.beparams)
             # check bridges existence
             _CheckInstanceBridgesExist(self, instance)
             remote_info = self.rpc.call_instance_info(instance.primary_node,
                                                       instance.name,
                                                       instance.hypervisor)
             remote_info.Raise("Error checking node %s" % instance.primary_node,
                               prereq=True, ecode=errors.ECODE_ENVIRON)
             if not remote_info.payload: # not running already
               _CheckNodeFreeMemory(self, instance.primary_node,
                                    "starting instance %s" % instance.name,
                                    bep[constants.BE_MINMEM], instance.hypervisor)
         def Exec(self, feedback_fn):
           """Start the instance.
           """
           instance = self.instance
           force = self.op.force
           reason = self.op.reason
           if not self.op.no_remember:
             self.cfg.MarkInstanceUp(instance.name)
           if self.primary_offline:
             assert self.op.ignore_offline_nodes
             self.LogInfo("Primary node offline, marked instance as started")
           else:
             node_current = instance.primary_node
             _StartInstanceDisks(self, instance, force)
             result = \
               self.rpc.call_instance_start(node_current,
                                            (instance, self.op.hvparams,
                                             self.op.beparams),
                                            self.op.startup_paused, reason)
             msg = result.fail_msg
             if msg:
               _ShutdownInstanceDisks(self, instance)
               raise errors.OpExecError("Could not start instance: %s" % msg)
       class LUInstanceReboot(LogicalUnit):
         """Reboot an instance.
         """
         HPATH = "instance-reboot"
         HTYPE = constants.HTYPE_INSTANCE
         REQ_BGL = False
         def ExpandNames(self):
           self._ExpandAndLockInstance()
         def BuildHooksEnv(self):
           """Build hooks env.
           This runs on master, primary and secondary nodes of the instance.
           """
           env = {
             "IGNORE_SECONDARIES": self.op.ignore_secondaries,
             "REBOOT_TYPE": self.op.reboot_type,
             "SHUTDOWN_TIMEOUT": self.op.shutdown_timeout,
+            }
           env.update(_BuildInstanceHookEnvByObject(self, self.instance))
           return env
         def BuildHooksNodes(self):
           """Build hooks nodes.
           """
           nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
           return (nl, nl)
         def CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster.
           """
           self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
           assert self.instance is not None, \
             "Cannot retrieve locked instance %s" % self.op.instance_name
           _CheckInstanceState(self, instance, INSTANCE_ONLINE)
           _CheckNodeOnline(self, instance.primary_node)
           # check bridges existence
           _CheckInstanceBridgesExist(self, instance)
         def Exec(self, feedback_fn):
           """Reboot the instance.
           """
           instance = self.instance
           ignore_secondaries = self.op.ignore_secondaries
           reboot_type = self.op.reboot_type
           reason = self.op.reason
           remote_info = self.rpc.call_instance_info(instance.primary_node,
                                                     instance.name,
                                                     instance.hypervisor)
           remote_info.Raise("Error checking node %s" % instance.primary_node)
           instance_running = bool(remote_info.payload)
           node_current = instance.primary_node
           if instance_running and reboot_type in [constants.INSTANCE_REBOOT_SOFT,
                                                   constants.INSTANCE_REBOOT_HARD]:
             for disk in instance.disks:
               self.cfg.SetDiskID(disk, node_current)
             result = self.rpc.call_instance_reboot(node_current, instance,
                                                    reboot_type,
                                                    self.op.shutdown_timeout, reason)
             result.Raise("Could not reboot instance")
           else:
             if instance_running:
               result = self.rpc.call_instance_shutdown(node_current, instance,
                                                        self.op.shutdown_timeout,
                                                        reason)
               result.Raise("Could not shutdown instance for full reboot")
               _ShutdownInstanceDisks(self, instance)
             else:
               self.LogInfo("Instance %s was already stopped, starting now",
                            instance.name)
             _StartInstanceDisks(self, instance, ignore_secondaries)
             result = self.rpc.call_instance_start(node_current,
                                                   (instance, None, None), False,
                                                    reason)
             msg = result.fail_msg
             if msg:
               _ShutdownInstanceDisks(self, instance)
               raise errors.OpExecError("Could not start instance for"
                                        " full reboot: %s" % msg)
           self.cfg.MarkInstanceUp(instance.name)
       class LUInstanceShutdown(LogicalUnit):
         """Shutdown an instance.
         """
         HPATH = "instance-stop"
         HTYPE = constants.HTYPE_INSTANCE
         REQ_BGL = False
         def ExpandNames(self):
           self._ExpandAndLockInstance()
         def BuildHooksEnv(self):
           """Build hooks env.
           This runs on master, primary and secondary nodes of the instance.
           """
           env = _BuildInstanceHookEnvByObject(self, self.instance)
           env["TIMEOUT"] = self.op.timeout
           return env
         def BuildHooksNodes(self):
           """Build hooks nodes.
           """
           nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
           return (nl, nl)
         def CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster.
           """
           self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
           assert self.instance is not None, \
             "Cannot retrieve locked instance %s" % self.op.instance_name
           if not self.op.force:
             _CheckInstanceState(self, self.instance, INSTANCE_ONLINE)
           else:
             self.LogWarning("Ignoring offline instance check")
           self.primary_offline = \
             self.cfg.GetNodeInfo(self.instance.primary_node).offline
           if self.primary_offline and self.op.ignore_offline_nodes:
             self.LogWarning("Ignoring offline primary node")
           else:
             _CheckNodeOnline(self, self.instance.primary_node)
         def Exec(self, feedback_fn):
           """Shutdown the instance.
           """
           instance = self.instance
           node_current = instance.primary_node
           timeout = self.op.timeout
           reason = self.op.reason
           # If the instance is offline we shouldn't mark it as down, as that
           # resets the offline flag.
           if not self.op.no_remember and instance.admin_state in INSTANCE_ONLINE:
             self.cfg.MarkInstanceDown(instance.name)
           if self.primary_offline:
             assert self.op.ignore_offline_nodes
             self.LogInfo("Primary node offline, marked instance as stopped")
           else:
             result = self.rpc.call_instance_shutdown(node_current, instance, timeout,
                                                      reason)
             msg = result.fail_msg
             if msg:
               self.LogWarning("Could not shutdown instance: %s", msg)
             _ShutdownInstanceDisks(self, instance)
       class LUInstanceReinstall(LogicalUnit):
         """Reinstall an instance.
         """
         HPATH = "instance-reinstall"
         HTYPE = constants.HTYPE_INSTANCE
         REQ_BGL = False
         def ExpandNames(self):
           self._ExpandAndLockInstance()
         def BuildHooksEnv(self):
           """Build hooks env.
           This runs on master, primary and secondary nodes of the instance.
           """
           return _BuildInstanceHookEnvByObject(self, self.instance)
         def BuildHooksNodes(self):
           """Build hooks nodes.
           """
           nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
           return (nl, nl)
         def CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster and is not running.
           """
           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, "Instance primary node"
                            " offline, cannot reinstall")
           if instance.disk_template == constants.DT_DISKLESS:
             raise errors.OpPrereqError("Instance '%s' has no disks" %
                                        self.op.instance_name,
                                        errors.ECODE_INVAL)
           _CheckInstanceState(self, instance, INSTANCE_DOWN, msg="cannot reinstall")
           if self.op.os_type is not None:
             # OS verification
             pnode = _ExpandNodeName(self.cfg, instance.primary_node)
             _CheckNodeHasOS(self, pnode, self.op.os_type, self.op.force_variant)
             instance_os = self.op.os_type
           else:
             instance_os = instance.os
           nodelist = list(instance.all_nodes)
           if self.op.osparams:
             i_osdict = _GetUpdatedParams(instance.osparams, self.op.osparams)
             _CheckOSParams(self, True, nodelist, instance_os, i_osdict)
             self.os_inst = i_osdict # the new dict (without defaults)
           else:
             self.os_inst = None
           self.instance = instance
         def Exec(self, feedback_fn):
           """Reinstall the instance.
           """
           inst = self.instance
           if self.op.os_type is not None:
             feedback_fn("Changing OS to '%s'..." % self.op.os_type)
             inst.os = self.op.os_type
             # Write to configuration
             self.cfg.Update(inst, feedback_fn)
           _StartInstanceDisks(self, inst, None)
           try:
             feedback_fn("Running the instance OS create scripts...")
             # FIXME: pass debug option from opcode to backend
             result = self.rpc.call_instance_os_add(inst.primary_node,
                                                    (inst, self.os_inst), True,
                                                    self.op.debug_level)
             result.Raise("Could not install OS for instance %s on node %s" %
                          (inst.name, inst.primary_node))
           finally:
             _ShutdownInstanceDisks(self, inst)
       class LUInstanceRecreateDisks(LogicalUnit):
         """Recreate an instance's missing disks.
         """
         HPATH = "instance-recreate-disks"
         HTYPE = constants.HTYPE_INSTANCE
         REQ_BGL = False
         _MODIFYABLE = compat.UniqueFrozenset([
           constants.IDISK_SIZE,
           constants.IDISK_MODE,
           ])
         # New or changed disk parameters may have different semantics
         assert constants.IDISK_PARAMS == (_MODIFYABLE | frozenset([
           constants.IDISK_ADOPT,
           # TODO: Implement support changing VG while recreating
           constants.IDISK_VG,
           constants.IDISK_METAVG,
           constants.IDISK_PROVIDER,
           constants.IDISK_NAME,
           ]))
         def _RunAllocator(self):
           """Run the allocator based on input opcode.
           """
           be_full = self.cfg.GetClusterInfo().FillBE(self.instance)
           # FIXME
           # The allocator should actually run in "relocate" mode, but current
           # allocators don't support relocating all the nodes of an instance at
           # the same time. As a workaround we use "allocate" mode, but this is
           # suboptimal for two reasons:
           # - The instance name passed to the allocator is present in the list of
           #   existing instances, so there could be a conflict within the
           #   internal structures of the allocator. This doesn't happen with the
           #   current allocators, but it's a liability.
           # - The allocator counts the resources used by the instance twice: once
           #   because the instance exists already, and once because it tries to
           #   allocate a new instance.
           # The allocator could choose some of the nodes on which the instance is
           # running, but that's not a problem. If the instance nodes are broken,
           # they should be already be marked as drained or offline, and hence
           # skipped by the allocator. If instance disks have been lost for other
           # reasons, then recreating the disks on the same nodes should be fine.
           disk_template = self.instance.disk_template
           spindle_use = be_full[constants.BE_SPINDLE_USE]
           req = iallocator.IAReqInstanceAlloc(name=self.op.instance_name,
                                               disk_template=disk_template,
                                               tags=list(self.instance.GetTags()),
                                               os=self.instance.os,
                                               nics=[{}],
                                               vcpus=be_full[constants.BE_VCPUS],
                                               memory=be_full[constants.BE_MAXMEM],
                                               spindle_use=spindle_use,
                                               disks=[{constants.IDISK_SIZE: d.size,
                                                       constants.IDISK_MODE: d.mode}
                                                       for d in self.instance.disks],
                                               hypervisor=self.instance.hypervisor,
                                               node_whitelist=None)
           ial = iallocator.IAllocator(self.cfg, self.rpc, req)
           ial.Run(self.op.iallocator)
           assert req.RequiredNodes() == len(self.instance.all_nodes)
           if not ial.success:
             raise errors.OpPrereqError("Can't compute nodes using iallocator '%s':"
                                        " %s" % (self.op.iallocator, ial.info),
                                        errors.ECODE_NORES)
           self.op.nodes = ial.result
           self.LogInfo("Selected nodes for instance %s via iallocator %s: %s",
                        self.op.instance_name, self.op.iallocator,
                        utils.CommaJoin(ial.result))
         def CheckArguments(self):
           if self.op.disks and ht.TNonNegativeInt(self.op.disks[0]):
             # Normalize and convert deprecated list of disk indices
             self.op.disks = [(idx, {}) for idx in sorted(frozenset(self.op.disks))]
           duplicates = utils.FindDuplicates(map(compat.fst, self.op.disks))
           if duplicates:
             raise errors.OpPrereqError("Some disks have been specified more than"
                                        " once: %s" % utils.CommaJoin(duplicates),
                                        errors.ECODE_INVAL)
           # We don't want _CheckIAllocatorOrNode selecting the default iallocator
           # when neither iallocator nor nodes are specified
           if self.op.iallocator or self.op.nodes:
             _CheckIAllocatorOrNode(self, "iallocator", "nodes")
           for (idx, params) in self.op.disks:
             utils.ForceDictType(params, constants.IDISK_PARAMS_TYPES)
             unsupported = frozenset(params.keys()) - self._MODIFYABLE
             if unsupported:
               raise errors.OpPrereqError("Parameters for disk %s try to change"
                                          " unmodifyable parameter(s): %s" %
                                          (idx, utils.CommaJoin(unsupported)),
                                          errors.ECODE_INVAL)
         def ExpandNames(self):
           self._ExpandAndLockInstance()
           self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
           if self.op.nodes:
             self.op.nodes = [_ExpandNodeName(self.cfg, n) for n in self.op.nodes]
             self.needed_locks[locking.LEVEL_NODE] = list(self.op.nodes)
           else:
             self.needed_locks[locking.LEVEL_NODE] = []
             if self.op.iallocator:
               # iallocator will select a new node in the same group
               self.needed_locks[locking.LEVEL_NODEGROUP] = []
               self.needed_locks[locking.LEVEL_NODE_ALLOC] = locking.ALL_SET
           self.needed_locks[locking.LEVEL_NODE_RES] = []
         def DeclareLocks(self, level):
           if level == locking.LEVEL_NODEGROUP:
             assert self.op.iallocator is not None
             assert not self.op.nodes
             assert not self.needed_locks[locking.LEVEL_NODEGROUP]
             self.share_locks[locking.LEVEL_NODEGROUP] = 1
             # Lock the primary group used by the instance optimistically; this
             # requires going via the node before it's locked, requiring
             # verification later on
             self.needed_locks[locking.LEVEL_NODEGROUP] = \
               self.cfg.GetInstanceNodeGroups(self.op.instance_name, primary_only=True)
           elif level == locking.LEVEL_NODE:
             # If an allocator is used, then we lock all the nodes in the current
             # instance group, as we don't know yet which ones will be selected;
             # if we replace the nodes without using an allocator, locks are
             # already declared in ExpandNames; otherwise, we need to lock all the
             # instance nodes for disk re-creation
             if self.op.iallocator:
               assert not self.op.nodes
               assert not self.needed_locks[locking.LEVEL_NODE]
               assert len(self.owned_locks(locking.LEVEL_NODEGROUP)) == 1
               # Lock member nodes of the group of the primary node
               for group_uuid in self.owned_locks(locking.LEVEL_NODEGROUP):
                 self.needed_locks[locking.LEVEL_NODE].extend(
                   self.cfg.GetNodeGroup(group_uuid).members)
               assert locking.NAL in self.owned_locks(locking.LEVEL_NODE_ALLOC)
             elif not self.op.nodes:
               self._LockInstancesNodes(primary_only=False)
           elif level == locking.LEVEL_NODE_RES:
             # Copy node locks
             self.needed_locks[locking.LEVEL_NODE_RES] = \
               _CopyLockList(self.needed_locks[locking.LEVEL_NODE])
         def BuildHooksEnv(self):
           """Build hooks env.
           This runs on master, primary and secondary nodes of the instance.
           """
           return _BuildInstanceHookEnvByObject(self, self.instance)
         def BuildHooksNodes(self):
           """Build hooks nodes.
           """
           nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
           return (nl, nl)
         def CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster and is not running.
           """
           instance = self.cfg.GetInstanceInfo(self.op.instance_name)
           assert instance is not None, \
             "Cannot retrieve locked instance %s" % self.op.instance_name
           if self.op.nodes:
             if len(self.op.nodes) != len(instance.all_nodes):
               raise errors.OpPrereqError("Instance %s currently has %d nodes, but"
                                          " %d replacement nodes were specified" %
                                          (instance.name, len(instance.all_nodes),
                                           len(self.op.nodes)),
                                          errors.ECODE_INVAL)
             assert instance.disk_template != constants.DT_DRBD8 or \
                 len(self.op.nodes) == 2
             assert instance.disk_template != constants.DT_PLAIN or \
                 len(self.op.nodes) == 1
             primary_node = self.op.nodes[0]
           else:
             primary_node = instance.primary_node
           if not self.op.iallocator:
             _CheckNodeOnline(self, primary_node)
           if instance.disk_template == constants.DT_DISKLESS:
             raise errors.OpPrereqError("Instance '%s' has no disks" %
                                        self.op.instance_name, errors.ECODE_INVAL)
           # Verify if node group locks are still correct
           owned_groups = self.owned_locks(locking.LEVEL_NODEGROUP)
           if owned_groups:
             # Node group locks are acquired only for the primary node (and only
             # when the allocator is used)
             _CheckInstanceNodeGroups(self.cfg, self.op.instance_name, owned_groups,
                                      primary_only=True)
           # if we replace nodes *and* the old primary is offline, we don't
           # check the instance state
           old_pnode = self.cfg.GetNodeInfo(instance.primary_node)
           if not ((self.op.iallocator or self.op.nodes) and old_pnode.offline):
             _CheckInstanceState(self, instance, INSTANCE_NOT_RUNNING,
                                 msg="cannot recreate disks")
           if self.op.disks:
             self.disks = dict(self.op.disks)
           else:
             self.disks = dict((idx, {}) for idx in range(len(instance.disks)))
           maxidx = max(self.disks.keys())
           if maxidx >= len(instance.disks):
             raise errors.OpPrereqError("Invalid disk index '%s'" % maxidx,
                                        errors.ECODE_INVAL)
           if ((self.op.nodes or self.op.iallocator) and
               sorted(self.disks.keys()) != range(len(instance.disks))):
             raise errors.OpPrereqError("Can't recreate disks partially and"
                                        " change the nodes at the same time",
                                        errors.ECODE_INVAL)
           self.instance = instance
           if self.op.iallocator:
             self._RunAllocator()
             # Release unneeded node and node resource locks
             _ReleaseLocks(self, locking.LEVEL_NODE, keep=self.op.nodes)
             _ReleaseLocks(self, locking.LEVEL_NODE_RES, keep=self.op.nodes)
             _ReleaseLocks(self, locking.LEVEL_NODE_ALLOC)
           assert not self.glm.is_owned(locking.LEVEL_NODE_ALLOC)
         def Exec(self, feedback_fn):
           """Recreate the disks.
           """
           instance = self.instance
           assert (self.owned_locks(locking.LEVEL_NODE) ==
                   self.owned_locks(locking.LEVEL_NODE_RES))
           to_skip = []
           mods = [] # keeps track of needed changes
           for idx, disk in enumerate(instance.disks):
             try:
               changes = self.disks[idx]
             except KeyError:
               # Disk should not be recreated
               to_skip.append(idx)
               continue
             # update secondaries for disks, if needed
             if self.op.nodes and disk.dev_type == constants.LD_DRBD8:
               # need to update the nodes and minors
               assert len(self.op.nodes) == 2
               assert len(disk.logical_id) == 6 # otherwise disk internals
                                                # have changed
               (_, _, old_port, _, _, old_secret) = disk.logical_id
               new_minors = self.cfg.AllocateDRBDMinor(self.op.nodes, instance.name)
               new_id = (self.op.nodes[0], self.op.nodes[1], old_port,
                         new_minors[0], new_minors[1], old_secret)
               assert len(disk.logical_id) == len(new_id)
             else:
               new_id = None
             mods.append((idx, new_id, changes))
           # now that we have passed all asserts above, we can apply the mods
           # in a single run (to avoid partial changes)
           for idx, new_id, changes in mods:
             disk = instance.disks[idx]
             if new_id is not None:
               assert disk.dev_type == constants.LD_DRBD8
               disk.logical_id = new_id
             if changes:
               disk.Update(size=changes.get(constants.IDISK_SIZE, None),
                           mode=changes.get(constants.IDISK_MODE, None))
           # change primary node, if needed
           if self.op.nodes:
             instance.primary_node = self.op.nodes[0]
             self.LogWarning("Changing the instance's nodes, you will have to"
                             " remove any disks left on the older nodes manually")
           if self.op.nodes:
             self.cfg.Update(instance, feedback_fn)
           # All touched nodes must be locked
           mylocks = self.owned_locks(locking.LEVEL_NODE)
           assert mylocks.issuperset(frozenset(instance.all_nodes))
           _CreateDisks(self, instance, to_skip=to_skip)
       class LUInstanceRename(LogicalUnit):
         """Rename an instance.
         """
         HPATH = "instance-rename"
         HTYPE = constants.HTYPE_INSTANCE
         def CheckArguments(self):
           """Check arguments.
           """
           if self.op.ip_check and not self.op.name_check:
             # TODO: make the ip check more flexible and not depend on the name check
             raise errors.OpPrereqError("IP address check requires a name check",
                                        errors.ECODE_INVAL)
         def BuildHooksEnv(self):
           """Build hooks env.
           This runs on master, primary and secondary nodes of the instance.
           """
           env = _BuildInstanceHookEnvByObject(self, self.instance)
           env["INSTANCE_NEW_NAME"] = self.op.new_name
           return env
         def BuildHooksNodes(self):
           """Build hooks nodes.
           """
           nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
           return (nl, nl)
         def CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster and is not running.
           """
           self.op.instance_name = _ExpandInstanceName(self.cfg,
                                                       self.op.instance_name)
           instance = self.cfg.GetInstanceInfo(self.op.instance_name)
           assert instance is not None
           _CheckNodeOnline(self, instance.primary_node)
           _CheckInstanceState(self, instance, INSTANCE_NOT_RUNNING,
                               msg="cannot rename")
           self.instance = instance
           new_name = self.op.new_name
           if self.op.name_check:
             hostname = _CheckHostnameSane(self, new_name)
             new_name = self.op.new_name = hostname.name
             if (self.op.ip_check and
                 netutils.TcpPing(hostname.ip, constants.DEFAULT_NODED_PORT)):
               raise errors.OpPrereqError("IP %s of instance %s already in use" %
                                          (hostname.ip, new_name),
                                          errors.ECODE_NOTUNIQUE)
           instance_list = self.cfg.GetInstanceList()
           if new_name in instance_list and new_name != instance.name:
             raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
                                        new_name, errors.ECODE_EXISTS)
         def Exec(self, feedback_fn):
           """Rename the instance.
           """
           inst = self.instance
           old_name = inst.name
           rename_file_storage = False
           if (inst.disk_template in constants.DTS_FILEBASED and
               self.op.new_name != inst.name):
             old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
             rename_file_storage = True
           self.cfg.RenameInstance(inst.name, self.op.new_name)
           # Change the instance lock. This is definitely safe while we hold the BGL.
           # Otherwise the new lock would have to be added in acquired mode.
           assert self.REQ_BGL
           assert locking.BGL in self.owned_locks(locking.LEVEL_CLUSTER)
           self.glm.remove(locking.LEVEL_INSTANCE, old_name)
           self.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
           # re-read the instance from the configuration after rename
           inst = self.cfg.GetInstanceInfo(self.op.new_name)
           if rename_file_storage:
             new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
             result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
                                                            old_file_storage_dir,
                                                            new_file_storage_dir)
             result.Raise("Could not rename on node %s directory '%s' to '%s'"
                          " (but the instance has been renamed in Ganeti)" %
                          (inst.primary_node, old_file_storage_dir,
                           new_file_storage_dir))
           _StartInstanceDisks(self, inst, None)
           # update info on disks
           info = _GetInstanceInfoText(inst)
           for (idx, disk) in enumerate(inst.disks):
             for node in inst.all_nodes:
               self.cfg.SetDiskID(disk, node)
               result = self.rpc.call_blockdev_setinfo(node, disk, info)
               if result.fail_msg:
                 self.LogWarning("Error setting info on node %s for disk %s: %s",
                                 node, idx, result.fail_msg)
           try:
             result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
                                                        old_name, self.op.debug_level)
             msg = result.fail_msg
             if msg:
               msg = ("Could not run OS rename script for instance %s on node %s"
                      " (but the instance has been renamed in Ganeti): %s" %
                      (inst.name, inst.primary_node, msg))
               self.LogWarning(msg)
           finally:
             _ShutdownInstanceDisks(self, inst)
           return inst.name
       class LUInstanceRemove(LogicalUnit):
         """Remove an instance.
         """
         HPATH = "instance-remove"
         HTYPE = constants.HTYPE_INSTANCE
         REQ_BGL = False
         def ExpandNames(self):
           self._ExpandAndLockInstance()
           self.needed_locks[locking.LEVEL_NODE] = []
           self.needed_locks[locking.LEVEL_NODE_RES] = []
           self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
         def DeclareLocks(self, level):
           if level == locking.LEVEL_NODE:
             self._LockInstancesNodes()
           elif level == locking.LEVEL_NODE_RES:
             # Copy node locks
             self.needed_locks[locking.LEVEL_NODE_RES] = \
               _CopyLockList(self.needed_locks[locking.LEVEL_NODE])
         def BuildHooksEnv(self):
           """Build hooks env.
           This runs on master, primary and secondary nodes of the instance.
           """
           env = _BuildInstanceHookEnvByObject(self, self.instance)
           env["SHUTDOWN_TIMEOUT"] = self.op.shutdown_timeout
           return env
         def BuildHooksNodes(self):
           """Build hooks nodes.
           """
           nl = [self.cfg.GetMasterNode()]
           nl_post = list(self.instance.all_nodes) + nl
           return (nl, nl_post)
         def CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster.
           """
           self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
           assert self.instance is not None, \
             "Cannot retrieve locked instance %s" % self.op.instance_name
         def Exec(self, feedback_fn):
           """Remove the instance.
           """
           instance = self.instance
           logging.info("Shutting down instance %s on node %s",
                        instance.name, instance.primary_node)
           result = self.rpc.call_instance_shutdown(instance.primary_node, instance,
                                                    self.op.shutdown_timeout,
                                                    self.op.reason)
           msg = result.fail_msg
           if msg:
             if self.op.ignore_failures:
               feedback_fn("Warning: can't shutdown instance: %s" % msg)
             else:
               raise errors.OpExecError("Could not shutdown instance %s on"
                                        " node %s: %s" %
                                        (instance.name, instance.primary_node, msg))
           assert (self.owned_locks(locking.LEVEL_NODE) ==
                   self.owned_locks(locking.LEVEL_NODE_RES))
           assert not (set(instance.all_nodes) -
                       self.owned_locks(locking.LEVEL_NODE)), \
             "Not owning correct locks"
           _RemoveInstance(self, feedback_fn, instance, self.op.ignore_failures)
       def _RemoveInstance(lu, feedback_fn, instance, ignore_failures):
         """Utility function to remove an instance.
         """
         logging.info("Removing block devices for instance %s", instance.name)
         if not _RemoveDisks(lu, instance, ignore_failures=ignore_failures):
           if not ignore_failures:
             raise errors.OpExecError("Can't remove instance's disks")
           feedback_fn("Warning: can't remove instance's disks")
         logging.info("Removing instance %s out of cluster config", instance.name)
         lu.cfg.RemoveInstance(instance.name)
         assert not lu.remove_locks.get(locking.LEVEL_INSTANCE), \
           "Instance lock removal conflict"
         # Remove lock for the instance
         lu.remove_locks[locking.LEVEL_INSTANCE] = instance.name
       class LUInstanceQuery(NoHooksLU):
         """Logical unit for querying instances.
         """
         # pylint: disable=W0142
         REQ_BGL = False
         def CheckArguments(self):
           self.iq = _InstanceQuery(qlang.MakeSimpleFilter("name", self.op.names),
                                    self.op.output_fields, self.op.use_locking)
         def ExpandNames(self):
           self.iq.ExpandNames(self)
         def DeclareLocks(self, level):
           self.iq.DeclareLocks(self, level)
         def Exec(self, feedback_fn):
           return self.iq.OldStyleQuery(self)
       def _ExpandNamesForMigration(lu):
         """Expands names for use with L{TLMigrateInstance}.
         @type lu: L{LogicalUnit}
         """
         if lu.op.target_node is not None:
           lu.op.target_node = _ExpandNodeName(lu.cfg, lu.op.target_node)
         lu.needed_locks[locking.LEVEL_NODE] = []
         lu.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
         lu.needed_locks[locking.LEVEL_NODE_RES] = []
         lu.recalculate_locks[locking.LEVEL_NODE_RES] = constants.LOCKS_REPLACE
         # The node allocation lock is actually only needed for externally replicated
         # instances (e.g. sharedfile or RBD) and if an iallocator is used.
         lu.needed_locks[locking.LEVEL_NODE_ALLOC] = []
       def _DeclareLocksForMigration(lu, level):
         """Declares locks for L{TLMigrateInstance}.
         @type lu: L{LogicalUnit}
         @param level: Lock level
         """
         if level == locking.LEVEL_NODE_ALLOC:
           assert lu.op.instance_name in lu.owned_locks(locking.LEVEL_INSTANCE)
           instance = lu.cfg.GetInstanceInfo(lu.op.instance_name)
           # Node locks are already declared here rather than at LEVEL_NODE as we need
           # the instance object anyway to declare the node allocation lock.
           if instance.disk_template in constants.DTS_EXT_MIRROR:
             if lu.op.target_node is None:
               lu.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
               lu.needed_locks[locking.LEVEL_NODE_ALLOC] = locking.ALL_SET
             else:
               lu.needed_locks[locking.LEVEL_NODE] = [instance.primary_node,
                                                      lu.op.target_node]
             del lu.recalculate_locks[locking.LEVEL_NODE]
           else:
             lu._LockInstancesNodes() # pylint: disable=W0212
         elif level == locking.LEVEL_NODE:
           # Node locks are declared together with the node allocation lock
           assert (lu.needed_locks[locking.LEVEL_NODE] or
                   lu.needed_locks[locking.LEVEL_NODE] is locking.ALL_SET)
         elif level == locking.LEVEL_NODE_RES:
           # Copy node locks
           lu.needed_locks[locking.LEVEL_NODE_RES] = \
             _CopyLockList(lu.needed_locks[locking.LEVEL_NODE])
       class LUInstanceFailover(LogicalUnit):
         """Failover an instance.
         """
         HPATH = "instance-failover"
         HTYPE = constants.HTYPE_INSTANCE
         REQ_BGL = False
         def CheckArguments(self):
           """Check the arguments.
           """
           self.iallocator = getattr(self.op, "iallocator", None)
           self.target_node = getattr(self.op, "target_node", None)
         def ExpandNames(self):
           self._ExpandAndLockInstance()
           _ExpandNamesForMigration(self)
           self._migrater = \
             TLMigrateInstance(self, self.op.instance_name, False, True, False,
                               self.op.ignore_consistency, True,
                               self.op.shutdown_timeout, self.op.ignore_ipolicy)
           self.tasklets = [self._migrater]
         def DeclareLocks(self, level):
           _DeclareLocksForMigration(self, level)
         def BuildHooksEnv(self):
           """Build hooks env.
           This runs on master, primary and secondary nodes of the instance.
           """
           instance = self._migrater.instance
           source_node = instance.primary_node
           target_node = self.op.target_node
           env = {
             "IGNORE_CONSISTENCY": self.op.ignore_consistency,
             "SHUTDOWN_TIMEOUT": self.op.shutdown_timeout,
             "OLD_PRIMARY": source_node,
             "NEW_PRIMARY": target_node,
+            }
           if instance.disk_template in constants.DTS_INT_MIRROR:
             env["OLD_SECONDARY"] = instance.secondary_nodes[0]
             env["NEW_SECONDARY"] = source_node
           else:
             env["OLD_SECONDARY"] = env["NEW_SECONDARY"] = ""
           env.update(_BuildInstanceHookEnvByObject(self, instance))
           return env
         def BuildHooksNodes(self):
           """Build hooks nodes.
           """
           instance = self._migrater.instance
           nl = [self.cfg.GetMasterNode()] + list(instance.secondary_nodes)
           return (nl, nl + [instance.primary_node])
       class LUInstanceMigrate(LogicalUnit):
         """Migrate an instance.
         This is migration without shutting down, compared to the failover,
         which is done with shutdown.
         """
         HPATH = "instance-migrate"
         HTYPE = constants.HTYPE_INSTANCE
         REQ_BGL = False
         def ExpandNames(self):
           self._ExpandAndLockInstance()
           _ExpandNamesForMigration(self)
           self._migrater = \
             TLMigrateInstance(self, self.op.instance_name, self.op.cleanup,
                               False, self.op.allow_failover, False,
                               self.op.allow_runtime_changes,
                               constants.DEFAULT_SHUTDOWN_TIMEOUT,
                               self.op.ignore_ipolicy)
           self.tasklets = [self._migrater]
         def DeclareLocks(self, level):
           _DeclareLocksForMigration(self, level)
         def BuildHooksEnv(self):
           """Build hooks env.
           This runs on master, primary and secondary nodes of the instance.
           """
           instance = self._migrater.instance
           source_node = instance.primary_node
           target_node = self.op.target_node
           env = _BuildInstanceHookEnvByObject(self, instance)
           env.update({
             "MIGRATE_LIVE": self._migrater.live,
             "MIGRATE_CLEANUP": self.op.cleanup,
             "OLD_PRIMARY": source_node,
             "NEW_PRIMARY": target_node,
             "ALLOW_RUNTIME_CHANGES": self.op.allow_runtime_changes,
             })
           if instance.disk_template in constants.DTS_INT_MIRROR:
             env["OLD_SECONDARY"] = target_node
             env["NEW_SECONDARY"] = source_node
           else:
             env["OLD_SECONDARY"] = env["NEW_SECONDARY"] = None
           return env
         def BuildHooksNodes(self):
           """Build hooks nodes.
           """
           instance = self._migrater.instance
           snodes = list(instance.secondary_nodes)
           nl = [self.cfg.GetMasterNode(), instance.primary_node] + snodes
           return (nl, nl)
       class LUInstanceMove(LogicalUnit):
         """Move an instance by data-copying.
         """
         HPATH = "instance-move"
         HTYPE = constants.HTYPE_INSTANCE
         REQ_BGL = False
         def ExpandNames(self):
           self._ExpandAndLockInstance()
           target_node = _ExpandNodeName(self.cfg, self.op.target_node)
           self.op.target_node = target_node
           self.needed_locks[locking.LEVEL_NODE] = [target_node]
           self.needed_locks[locking.LEVEL_NODE_RES] = []
           self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_APPEND
         def DeclareLocks(self, level):
           if level == locking.LEVEL_NODE:
             self._LockInstancesNodes(primary_only=True)
           elif level == locking.LEVEL_NODE_RES:
             # Copy node locks
             self.needed_locks[locking.LEVEL_NODE_RES] = \
               _CopyLockList(self.needed_locks[locking.LEVEL_NODE])
         def BuildHooksEnv(self):
           """Build hooks env.
           This runs on master, primary and secondary nodes of the instance.
           """
           env = {
             "TARGET_NODE": self.op.target_node,
             "SHUTDOWN_TIMEOUT": self.op.shutdown_timeout,
+            }
           env.update(_BuildInstanceHookEnvByObject(self, self.instance))
           return env
         def BuildHooksNodes(self):
           """Build hooks nodes.
           """
           nl = [
             self.cfg.GetMasterNode(),
             self.instance.primary_node,
             self.op.target_node,
+            ]
           return (nl, nl)
         def CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster.
           """
           self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
           assert self.instance is not None, \
             "Cannot retrieve locked instance %s" % self.op.instance_name
           if instance.disk_template not in constants.DTS_COPYABLE:
             raise errors.OpPrereqError("Disk template %s not suitable for copying" %
                                        instance.disk_template, errors.ECODE_STATE)
           node = self.cfg.GetNodeInfo(self.op.target_node)
           assert node is not None, \
             "Cannot retrieve locked node %s" % self.op.target_node
           self.target_node = target_node = node.name
           if target_node == instance.primary_node:
             raise errors.OpPrereqError("Instance %s is already on the node %s" %
                                        (instance.name, target_node),
                                        errors.ECODE_STATE)
           bep = self.cfg.GetClusterInfo().FillBE(instance)
           for idx, dsk in enumerate(instance.disks):
             if dsk.dev_type not in (constants.LD_LV, constants.LD_FILE):
               raise errors.OpPrereqError("Instance disk %d has a complex layout,"
                                          " cannot copy" % idx, errors.ECODE_STATE)
           _CheckNodeOnline(self, target_node)
           _CheckNodeNotDrained(self, target_node)
           _CheckNodeVmCapable(self, target_node)
           cluster = self.cfg.GetClusterInfo()
           group_info = self.cfg.GetNodeGroup(node.group)
           ipolicy = ganeti.masterd.instance.CalculateGroupIPolicy(cluster, group_info)
           _CheckTargetNodeIPolicy(self, ipolicy, instance, node, self.cfg,
                                   ignore=self.op.ignore_ipolicy)
           if instance.admin_state == constants.ADMINST_UP:
             # check memory requirements on the secondary node
             _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
                                  instance.name, bep[constants.BE_MAXMEM],
                                  instance.hypervisor)
           else:
             self.LogInfo("Not checking memory on the secondary node as"
                          " instance will not be started")
           # check bridge existance
           _CheckInstanceBridgesExist(self, instance, node=target_node)
         def Exec(self, feedback_fn):
           """Move an instance.
           The move is done by shutting it down on its present node, copying
           the data over (slow) and starting it on the new node.
           """
           instance = self.instance
           source_node = instance.primary_node
           target_node = self.target_node
           self.LogInfo("Shutting down instance %s on source node %s",
                        instance.name, source_node)
           assert (self.owned_locks(locking.LEVEL_NODE) ==
                   self.owned_locks(locking.LEVEL_NODE_RES))
           result = self.rpc.call_instance_shutdown(source_node, instance,
                                                    self.op.shutdown_timeout,
                                                    self.op.reason)
           msg = result.fail_msg
           if msg:
             if self.op.ignore_consistency:
               self.LogWarning("Could not shutdown instance %s on node %s."
                               " Proceeding anyway. Please make sure node"
                               " %s is down. Error details: %s",
                               instance.name, source_node, source_node, msg)
             else:
               raise errors.OpExecError("Could not shutdown instance %s on"
                                        " node %s: %s" %
                                        (instance.name, source_node, msg))
           # create the target disks
           try:
             _CreateDisks(self, instance, target_node=target_node)
           except errors.OpExecError:
             self.LogWarning("Device creation failed")
             self.cfg.ReleaseDRBDMinors(instance.name)
             raise
           cluster_name = self.cfg.GetClusterInfo().cluster_name
           errs = []
           # activate, get path, copy the data over
           for idx, disk in enumerate(instance.disks):
             self.LogInfo("Copying data for disk %d", idx)
             result = self.rpc.call_blockdev_assemble(target_node, (disk, instance),
                                                      instance.name, True, idx)
             if result.fail_msg:
               self.LogWarning("Can't assemble newly created disk %d: %s",
                               idx, result.fail_msg)
               errs.append(result.fail_msg)
               break
             dev_path = result.payload
             result = self.rpc.call_blockdev_export(source_node, (disk, instance),
                                                    target_node, dev_path,
                                                    cluster_name)
             if result.fail_msg:
               self.LogWarning("Can't copy data over for disk %d: %s",
                               idx, result.fail_msg)
               errs.append(result.fail_msg)
               break
           if errs:
             self.LogWarning("Some disks failed to copy, aborting")
             try:
               _RemoveDisks(self, instance, target_node=target_node)
             finally:
               self.cfg.ReleaseDRBDMinors(instance.name)
               raise errors.OpExecError("Errors during disk copy: %s" %
                                        (",".join(errs),))
           instance.primary_node = target_node
           self.cfg.Update(instance, feedback_fn)
           self.LogInfo("Removing the disks on the original node")
           _RemoveDisks(self, instance, target_node=source_node)
           # Only start the instance if it's marked as up
           if instance.admin_state == constants.ADMINST_UP:
             self.LogInfo("Starting instance %s on node %s",
                          instance.name, target_node)
             disks_ok, _ = _AssembleInstanceDisks(self, instance,
                                                  ignore_secondaries=True)
             if not disks_ok:
               _ShutdownInstanceDisks(self, instance)
               raise errors.OpExecError("Can't activate the instance's disks")
             result = self.rpc.call_instance_start(target_node,
                                                   (instance, None, None), False,
                                                    self.op.reason)
             msg = result.fail_msg
             if msg:
               _ShutdownInstanceDisks(self, instance)
               raise errors.OpExecError("Could not start instance %s on node %s: %s" %
                                        (instance.name, target_node, msg))
       class TLMigrateInstance(Tasklet):
         """Tasklet class for instance migration.
         @type live: boolean
         @ivar live: whether the migration will be done live or non-live;
             this variable is initalized only after CheckPrereq has run
         @type cleanup: boolean
         @ivar cleanup: Wheater we cleanup from a failed migration
         @type iallocator: string
         @ivar iallocator: The iallocator used to determine target_node
         @type target_node: string
         @ivar target_node: If given, the target_node to reallocate the instance to
         @type failover: boolean
         @ivar failover: Whether operation results in failover or migration
         @type fallback: boolean
         @ivar fallback: Whether fallback to failover is allowed if migration not
                         possible
         @type ignore_consistency: boolean
         @ivar ignore_consistency: Wheter we should ignore consistency between source
                                   and target node
         @type shutdown_timeout: int
         @ivar shutdown_timeout: In case of failover timeout of the shutdown
         @type ignore_ipolicy: bool
         @ivar ignore_ipolicy: If true, we can ignore instance policy when migrating
         """
         # Constants
         _MIGRATION_POLL_INTERVAL = 1      # seconds
         _MIGRATION_FEEDBACK_INTERVAL = 10 # seconds
         def __init__(self, lu, instance_name, cleanup, failover, fallback,
                      ignore_consistency, allow_runtime_changes, shutdown_timeout,
                      ignore_ipolicy):
           """Initializes this class.
           """
           Tasklet.__init__(self, lu)
           # Parameters
           self.instance_name = instance_name
           self.cleanup = cleanup
           self.live = False # will be overridden later
           self.failover = failover
           self.fallback = fallback
           self.ignore_consistency = ignore_consistency
           self.shutdown_timeout = shutdown_timeout
           self.ignore_ipolicy = ignore_ipolicy
           self.allow_runtime_changes = allow_runtime_changes
         def CheckPrereq(self):
           """Check prerequisites.
           This checks that the instance is in the cluster.
           """
           instance_name = _ExpandInstanceName(self.lu.cfg, self.instance_name)
           instance = self.cfg.GetInstanceInfo(instance_name)
           assert instance is not None
           self.instance = instance
           cluster = self.cfg.GetClusterInfo()
           if (not self.cleanup and
               not instance.admin_state == constants.ADMINST_UP and
               not self.failover and self.fallback):
             self.lu.LogInfo("Instance is marked down or offline, fallback allowed,"
                             " switching to failover")
             self.failover = True
           if instance.disk_template not in constants.DTS_MIRRORED:
             if self.failover:
               text = "failovers"
             else:
               text = "migrations"
             raise errors.OpPrereqError("Instance's disk layout '%s' does not allow"
                                        " %s" % (instance.disk_template, text),
                                        errors.ECODE_STATE)
           if instance.disk_template in constants.DTS_EXT_MIRROR:
             _CheckIAllocatorOrNode(self.lu, "iallocator", "target_node")
             if self.lu.op.iallocator:
               assert locking.NAL in self.lu.owned_locks(locking.LEVEL_NODE_ALLOC)
               self._RunAllocator()
             else:
               # We set set self.target_node as it is required by
               # BuildHooksEnv
               self.target_node = self.lu.op.target_node
             # Check that the target node is correct in terms of instance policy
             nodeinfo = self.cfg.GetNodeInfo(self.target_node)
             group_info = self.cfg.GetNodeGroup(nodeinfo.group)
             ipolicy = ganeti.masterd.instance.CalculateGroupIPolicy(cluster,
                                                                     group_info)
             _CheckTargetNodeIPolicy(self.lu, ipolicy, instance, nodeinfo, self.cfg,
                                     ignore=self.ignore_ipolicy)
             # self.target_node is already populated, either directly or by the
             # iallocator run
             target_node = self.target_node
             if self.target_node == instance.primary_node:
               raise errors.OpPrereqError("Cannot migrate instance %s"
                                          " to its primary (%s)" %
                                          (instance.name, instance.primary_node),
                                          errors.ECODE_STATE)
             if len(self.lu.tasklets) == 1:
               # It is safe to release locks only when we're the only tasklet
               # in the LU
               _ReleaseLocks(self.lu, locking.LEVEL_NODE,
                             keep=[instance.primary_node, self.target_node])
               _ReleaseLocks(self.lu, locking.LEVEL_NODE_ALLOC)
           else:
             assert not self.lu.glm.is_owned(locking.LEVEL_NODE_ALLOC)
             secondary_nodes = instance.secondary_nodes
             if not secondary_nodes:
               raise errors.ConfigurationError("No secondary node but using"
                                               " %s disk template" %
                                               instance.disk_template)
             target_node = secondary_nodes[0]
             if self.lu.op.iallocator or (self.lu.op.target_node and
                                          self.lu.op.target_node != target_node):
               if self.failover:
                 text = "failed over"
               else:
                 text = "migrated"
               raise errors.OpPrereqError("Instances with disk template %s cannot"
                                          " be %s to arbitrary nodes"
                                          " (neither an iallocator nor a target"
                                          " node can be passed)" %
                                          (instance.disk_template, text),
                                          errors.ECODE_INVAL)
             nodeinfo = self.cfg.GetNodeInfo(target_node)
             group_info = self.cfg.GetNodeGroup(nodeinfo.group)
             ipolicy = ganeti.masterd.instance.CalculateGroupIPolicy(cluster,
                                                                     group_info)
             _CheckTargetNodeIPolicy(self.lu, ipolicy, instance, nodeinfo, self.cfg,
                                     ignore=self.ignore_ipolicy)
           i_be = cluster.FillBE(instance)
           # check memory requirements on the secondary node
           if (not self.cleanup and
                (not self.failover or instance.admin_state == constants.ADMINST_UP)):
             self.tgt_free_mem = _CheckNodeFreeMemory(self.lu, target_node,
                                                      "migrating instance %s" %
                                                      instance.name,
                                                      i_be[constants.BE_MINMEM],
                                                      instance.hypervisor)
           else:
             self.lu.LogInfo("Not checking memory on the secondary node as"
                             " instance will not be started")
           # check if failover must be forced instead of migration
           if (not self.cleanup and not self.failover and
               i_be[constants.BE_ALWAYS_FAILOVER]):
             self.lu.LogInfo("Instance configured to always failover; fallback"
                             " to failover")
             self.failover = True
           # check bridge existance
           _CheckInstanceBridgesExist(self.lu, instance, node=target_node)
           if not self.cleanup:
             _CheckNodeNotDrained(self.lu, target_node)
             if not self.failover:
               result = self.rpc.call_instance_migratable(instance.primary_node,
                                                          instance)
               if result.fail_msg and self.fallback:
                 self.lu.LogInfo("Can't migrate, instance offline, fallback to"
                                 " failover")
                 self.failover = True
               else:
                 result.Raise("Can't migrate, please use failover",
                              prereq=True, ecode=errors.ECODE_STATE)
           assert not (self.failover and self.cleanup)
           if not self.failover:
             if self.lu.op.live is not None and self.lu.op.mode is not None:
               raise errors.OpPrereqError("Only one of the 'live' and 'mode'"
                                          " parameters are accepted",
                                          errors.ECODE_INVAL)
             if self.lu.op.live is not None:
               if self.lu.op.live:
                 self.lu.op.mode = constants.HT_MIGRATION_LIVE
               else:
                 self.lu.op.mode = constants.HT_MIGRATION_NONLIVE
               # reset the 'live' parameter to None so that repeated
               # invocations of CheckPrereq do not raise an exception
               self.lu.op.live = None
             elif self.lu.op.mode is None:
               # read the default value from the hypervisor
               i_hv = cluster.FillHV(self.instance, skip_globals=False)
               self.lu.op.mode = i_hv[constants.HV_MIGRATION_MODE]
             self.live = self.lu.op.mode == constants.HT_MIGRATION_LIVE
           else:
             # Failover is never live
             self.live = False
           if not (self.failover or self.cleanup):
             remote_info = self.rpc.call_instance_info(instance.primary_node,
                                                       instance.name,
                                                       instance.hypervisor)
             remote_info.Raise("Error checking instance on node %s" %
                               instance.primary_node)
             instance_running = bool(remote_info.payload)
             if instance_running:
               self.current_mem = int(remote_info.payload["memory"])
         def _RunAllocator(self):
           """Run the allocator based on input opcode.
           """
           assert locking.NAL in self.lu.owned_locks(locking.LEVEL_NODE_ALLOC)
           # FIXME: add a self.ignore_ipolicy option
           req = iallocator.IAReqRelocate(name=self.instance_name,
                                          relocate_from=[self.instance.primary_node])
           ial = iallocator.IAllocator(self.cfg, self.rpc, req)
           ial.Run(self.lu.op.iallocator)
           if not ial.success:
             raise errors.OpPrereqError("Can't compute nodes using"
                                        " iallocator '%s': %s" %
                                        (self.lu.op.iallocator, ial.info),
                                        errors.ECODE_NORES)
           self.target_node = ial.result[0]
           self.lu.LogInfo("Selected nodes for instance %s via iallocator %s: %s",
                           self.instance_name, self.lu.op.iallocator,
                           utils.CommaJoin(ial.result))
         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,
                                                    self.instance))
             min_percent = 100
             for node, nres in result.items():
               nres.Raise("Cannot resync disks on node %s" % node)
               node_done, node_percent = nres.payload
               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)
           result.Raise("Cannot change disk to secondary on node %s" % node)
         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():
             nres.Raise("Cannot disconnect disks node %s" % node)
         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),
                                                  self.instance.name, multimaster)
           for node, nres in result.items():
             nres.Raise("Cannot change disks config on node %s" % node)
         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
           """
           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("Can't contact node %s" % node)
           runningon_source = instance.name in ins_l[source_node].payload
           runningon_target = instance.name in ins_l[target_node].payload
           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, self.feedback_fn)
             demoted_node = source_node
           else:
             self.feedback_fn("* instance confirmed to be running on its"
                              " primary node (%s)" % source_node)
             demoted_node = target_node
           if instance.disk_template in constants.DTS_INT_MIRROR:
             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
           if self.instance.disk_template in constants.DTS_EXT_MIRROR:
             return
           try:
             self._EnsureSecondary(target_node)
             self._GoStandalone()
             self._GoReconnect(False)
             self._WaitUntilSync()
           except errors.OpExecError, err:
             self.lu.LogWarning("Migration failed and I can't reconnect the drives,"
                                " please try to recover the instance manually;"
                                " error '%s'" % str(err))
         def _AbortMigration(self):
           """Call the hypervisor code to abort a started migration.
           """
           instance = self.instance
           target_node = self.target_node
           source_node = self.source_node
           migration_info = self.migration_info
           abort_result = self.rpc.call_instance_finalize_migration_dst(target_node,
                                                                        instance,
                                                                        migration_info,
                                                                        False)
           abort_msg = abort_result.fail_msg
           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.
           abort_result = self.rpc.call_instance_finalize_migration_src(
             source_node, instance, False, self.live)
           abort_msg = abort_result.fail_msg
           if abort_msg:
             logging.error("Aborting migration failed on source node %s: %s",
                           source_node, abort_msg)
         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
           # Check for hypervisor version mismatch and warn the user.
           nodeinfo = self.rpc.call_node_info([source_node, target_node],
                                              None, [self.instance.hypervisor], False)
           for ninfo in nodeinfo.values():
             ninfo.Raise("Unable to retrieve node information from node '%s'" %
                         ninfo.node)
           (_, _, (src_info, )) = nodeinfo[source_node].payload
           (_, _, (dst_info, )) = nodeinfo[target_node].payload
           if ((constants.HV_NODEINFO_KEY_VERSION in src_info) and
               (constants.HV_NODEINFO_KEY_VERSION in dst_info)):
             src_version = src_info[constants.HV_NODEINFO_KEY_VERSION]
             dst_version = dst_info[constants.HV_NODEINFO_KEY_VERSION]
             if src_version != dst_version:
               self.feedback_fn("* warning: hypervisor version mismatch between"
                                " source (%s) and target (%s) node" %
                                (src_version, dst_version))
           self.feedback_fn("* checking disk consistency between source and target")
           for (idx, dev) in enumerate(instance.disks):
             if not _CheckDiskConsistency(self.lu, instance, dev, target_node, False):
               raise errors.OpExecError("Disk %s is degraded or not fully"
                                        " synchronized on target node,"
                                        " aborting migration" % idx)
           if self.current_mem > self.tgt_free_mem:
             if not self.allow_runtime_changes:
               raise errors.OpExecError("Memory ballooning not allowed and not enough"
                                        " free memory to fit instance %s on target"
                                        " node %s (have %dMB, need %dMB)" %
                                        (instance.name, target_node,
                                         self.tgt_free_mem, self.current_mem))
             self.feedback_fn("* setting instance memory to %s" % self.tgt_free_mem)
             rpcres = self.rpc.call_instance_balloon_memory(instance.primary_node,
                                                            instance,
                                                            self.tgt_free_mem)
             rpcres.Raise("Cannot modify instance runtime memory")
           # First get the migration information from the remote node
           result = self.rpc.call_migration_info(source_node, instance)
           msg = result.fail_msg
           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.payload
           if self.instance.disk_template not in constants.DTS_EXT_MIRROR:
             # 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.fail_msg
           if msg:
             logging.error("Instance pre-migration failed, trying to revert"
                           " disk status: %s", msg)
             self.feedback_fn("Pre-migration failed, aborting")
             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)
           result = self.rpc.call_instance_migrate(source_node, instance,
                                                   self.nodes_ip[target_node],
                                                   self.live)
           msg = result.fail_msg
           if msg:
             logging.error("Instance migration failed, trying to revert"
                           " disk status: %s", msg)
             self.feedback_fn("Migration failed, aborting")
             self._AbortMigration()
             self._RevertDiskStatus()
             raise errors.OpExecError("Could not migrate instance %s: %s" %
                                      (instance.name, msg))
           self.feedback_fn("* starting memory transfer")
           last_feedback = time.time()
           while True:
             result = self.rpc.call_instance_get_migration_status(source_node,
                                                                  instance)
             msg = result.fail_msg
             ms = result.payload   # MigrationStatus instance
             if msg or (ms.status in constants.HV_MIGRATION_FAILED_STATUSES):
               logging.error("Instance migration failed, trying to revert"
                             " disk status: %s", msg)
               self.feedback_fn("Migration failed, aborting")
               self._AbortMigration()
               self._RevertDiskStatus()
               if not msg:
                 msg = "hypervisor returned failure"
               raise errors.OpExecError("Could not migrate instance %s: %s" %
                                        (instance.name, msg))
             if result.payload.status != constants.HV_MIGRATION_ACTIVE:
               self.feedback_fn("* memory transfer complete")
               break
             if (utils.TimeoutExpired(last_feedback,
                                      self._MIGRATION_FEEDBACK_INTERVAL) and
                 ms.transferred_ram is not None):
               mem_progress = 100 * float(ms.transferred_ram) / float(ms.total_ram)
               self.feedback_fn("* memory transfer progress: %.2f %%" % mem_progress)
               last_feedback = time.time()
             time.sleep(self._MIGRATION_POLL_INTERVAL)
           result = self.rpc.call_instance_finalize_migration_src(source_node,
                                                                  instance,
                                                                  True,
                                                                  self.live)
           msg = result.fail_msg
           if msg:
             logging.error("Instance migration succeeded, but finalization failed"
                           " on the source node: %s", msg)
             raise errors.OpExecError("Could not finalize instance migration: %s" %
                                      msg)
           instance.primary_node = target_node
           # distribute new instance config to the other nodes
           self.cfg.Update(instance, self.feedback_fn)
           result = self.rpc.call_instance_finalize_migration_dst(target_node,
                                                                  instance,
                                                                  migration_info,
                                                                  True)
           msg = result.fail_msg
           if msg:
             logging.error("Instance migration succeeded, but finalization failed"
                           " on the target node: %s", msg)
             raise errors.OpExecError("Could not finalize instance migration: %s" %
                                      msg)
           if self.instance.disk_template not in constants.DTS_EXT_MIRROR:
             self._EnsureSecondary(source_node)
             self._WaitUntilSync()
             self._GoStandalone()
             self._GoReconnect(False)
             self._WaitUntilSync()
           # If the instance's disk template is `rbd' or `ext' and there was a
           # successful migration, unmap the device from the source node.
           if self.instance.disk_template in (constants.DT_RBD, constants.DT_EXT):
             disks = _ExpandCheckDisks(instance, instance.disks)
             self.feedback_fn("* unmapping instance's disks from %s" % source_node)
             for disk in disks:
               result = self.rpc.call_blockdev_shutdown(source_node, (disk, instance))
               msg = result.fail_msg
               if msg:
                 logging.error("Migration was successful, but couldn't unmap the"
                               " block device %s on source node %s: %s",
                               disk.iv_name, source_node, msg)
                 logging.error("You need to unmap the device %s manually on %s",
                               disk.iv_name, source_node)
           self.feedback_fn("* done")
         def _ExecFailover(self):
           """Failover an instance.
           The failover is done by shutting it down on its present node and
           starting it on the secondary.
           """
           instance = self.instance
           primary_node = self.cfg.GetNodeInfo(instance.primary_node)
           source_node = instance.primary_node
           target_node = self.target_node
           if instance.admin_state == constants.ADMINST_UP:
             self.feedback_fn("* checking disk consistency between source and target")
             for (idx, dev) in enumerate(instance.disks):
               # for drbd, these are drbd over lvm
               if not _CheckDiskConsistency(self.lu, instance, dev, target_node,
                                            False):
                 if primary_node.offline:
                   self.feedback_fn("Node %s is offline, ignoring degraded disk %s on"
                                    " target node %s" %
                                    (primary_node.name, idx, target_node))
                 elif not self.ignore_consistency:
                   raise errors.OpExecError("Disk %s is degraded on target node,"
                                            " aborting failover" % idx)
           else:
             self.feedback_fn("* not checking disk consistency as instance is not"
                              " running")
           self.feedback_fn("* shutting down instance on source node")
           logging.info("Shutting down instance %s on node %s",
                        instance.name, source_node)
           result = self.rpc.call_instance_shutdown(source_node, instance,
                                                    self.shutdown_timeout,
                                                    self.lu.op.reason)
           msg = result.fail_msg
           if msg:
             if self.ignore_consistency or primary_node.offline:
               self.lu.LogWarning("Could not shutdown instance %s on node %s,"
                                  " proceeding anyway; please make sure node"
                                  " %s is down; error details: %s",
                                  instance.name, source_node, source_node, msg)
             else:
               raise errors.OpExecError("Could not shutdown instance %s on"
                                        " node %s: %s" %
                                        (instance.name, source_node, msg))
           self.feedback_fn("* deactivating the instance's disks on source node")
           if not _ShutdownInstanceDisks(self.lu, instance, ignore_primary=True):
             raise errors.OpExecError("Can't shut down the instance's disks")
           instance.primary_node = target_node
           # distribute new instance config to the other nodes
           self.cfg.Update(instance, self.feedback_fn)
           # Only start the instance if it's marked as up
           if instance.admin_state == constants.ADMINST_UP:
             self.feedback_fn("* activating the instance's disks on target node %s" %
                              target_node)
             logging.info("Starting instance %s on node %s",
                          instance.name, target_node)
             disks_ok, _ = _AssembleInstanceDisks(self.lu, instance,
                                                  ignore_secondaries=True)
             if not disks_ok:
               _ShutdownInstanceDisks(self.lu, instance)
               raise errors.OpExecError("Can't activate the instance's disks")
             self.feedback_fn("* starting the instance on the target node %s" %
                              target_node)
             result = self.rpc.call_instance_start(target_node, (instance, None, None),
                                                   False, self.lu.op.reason)
             msg = result.fail_msg
             if msg:
               _ShutdownInstanceDisks(self.lu, instance)
               raise errors.OpExecError("Could not start instance %s on node %s: %s" %
                                        (instance.name, target_node, msg))
         def Exec(self, feedback_fn):
           """Perform the migration.
           """
           self.feedback_fn = feedback_fn
           self.source_node = self.instance.primary_node
           # FIXME: if we implement migrate-to-any in DRBD, this needs fixing
           if self.instance.disk_template in constants.DTS_INT_MIRROR:
             self.target_node = self.instance.secondary_nodes[0]
             # Otherwise self.target_node has been populated either
             # directly, or through an iallocator.
           self.all_nodes = [self.source_node, self.target_node]
           self.nodes_ip = dict((name, node.secondary_ip) for (name, node)
                                in self.cfg.GetMultiNodeInfo(self.all_nodes))
           if self.failover:
             feedback_fn("Failover instance %s" % self.instance.name)
             self._ExecFailover()
           else:
             feedback_fn("Migrating instance %s" % self.instance.name)
             if self.cleanup:
               return self._ExecCleanup()
             else:
               return self._ExecMigration()
       def _CreateBlockDev(lu, node, instance, device, force_create, info,
                           force_open):
         """Wrapper around L{_CreateBlockDevInner}.
         This method annotates the root device first.
         """
         (disk,) = _AnnotateDiskParams(instance, [device], lu.cfg)
         excl_stor = _IsExclusiveStorageEnabledNodeName(lu.cfg, node)
         return _CreateBlockDevInner(lu, node, instance, disk, force_create, info,
                                     force_open, excl_stor)
       def _CreateBlockDevInner(lu, node, instance, device, force_create,
                                info, force_open, excl_stor):
         """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.
         @attention: The device has to be annotated already.
         @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.BlockdevCreate} function where it specifies
             whether we run on primary or not, and it affects both
             the child assembly and the device own Open() execution
         @type excl_stor: boolean
         @param excl_stor: Whether exclusive_storage is active for the node
         @return: list of created devices
         """
         created_devices = []
         try:
           if device.CreateOnSecondary():
             force_create = True
           if device.children:
             for child in device.children:
               devs = _CreateBlockDevInner(lu, node, instance, child, force_create,
                                           info, force_open, excl_stor)
               created_devices.extend(devs)
           if not force_create:
             return created_devices
           _CreateSingleBlockDev(lu, node, instance, device, info, force_open,
                                 excl_stor)
           # The device has been completely created, so there is no point in keeping
           # its subdevices in the list. We just add the device itself instead.
           created_devices = [(node, device)]
           return created_devices
         except errors.DeviceCreationError, e:
           e.created_devices.extend(created_devices)
           raise e
         except errors.OpExecError, e:
           raise errors.DeviceCreationError(str(e), created_devices)
       def _CreateSingleBlockDev(lu, node, instance, device, info, force_open,
                                 excl_stor):
         """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.BlockdevCreate} function where it specifies
             whether we run on primary or not, and it affects both
             the child assembly and the device own Open() execution
         @type excl_stor: boolean
         @param excl_stor: Whether exclusive_storage is active for the node
         """
         lu.cfg.SetDiskID(device, node)
         result = lu.rpc.call_blockdev_create(node, device, device.size,
                                              instance.name, force_open, info,
                                              excl_stor)
         result.Raise("Can't create block device %s on"
                      " node %s for instance %s" % (device, node, instance.name))
         if device.physical_id is None:
           device.physical_id = result.payload
       def _GenerateUniqueNames(lu, exts):
         """Generate a suitable LV name.
         This will generate a logical volume name for the given instance.
         """
         results = []
         for val in exts:
           new_id = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
           results.append("%s%s" % (new_id, val))
         return results
       def _GenerateDRBD8Branch(lu, primary, secondary, size, vgnames, names,
                                iv_name, p_minor, s_minor):
         """Generate a drbd8 device complete with its children.
         """
         assert len(vgnames) == len(names) == 2
         port = lu.cfg.AllocatePort()
         shared_secret = lu.cfg.GenerateDRBDSecret(lu.proc.GetECId())
         dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
                                 logical_id=(vgnames[0], names[0]),
                                 params={})
         dev_data.uuid = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
         dev_meta = objects.Disk(dev_type=constants.LD_LV,
                                 size=constants.DRBD_META_SIZE,
                                 logical_id=(vgnames[1], names[1]),
                                 params={})
         dev_meta.uuid = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
         drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
                                 logical_id=(primary, secondary, port,
                                             p_minor, s_minor,
                                             shared_secret),
                                 children=[dev_data, dev_meta],
                                 iv_name=iv_name, params={})
         drbd_dev.uuid = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
         return drbd_dev
       _DISK_TEMPLATE_NAME_PREFIX = {
         constants.DT_PLAIN: "",
         constants.DT_RBD: ".rbd",
         constants.DT_EXT: ".ext",
+        }
       _DISK_TEMPLATE_DEVICE_TYPE = {
         constants.DT_PLAIN: constants.LD_LV,
         constants.DT_FILE: constants.LD_FILE,
         constants.DT_SHARED_FILE: constants.LD_FILE,
         constants.DT_BLOCK: constants.LD_BLOCKDEV,
         constants.DT_RBD: constants.LD_RBD,
         constants.DT_EXT: constants.LD_EXT,
+        }
       def _GenerateDiskTemplate(
         lu, template_name, instance_name, primary_node, secondary_nodes,
         disk_info, file_storage_dir, file_driver, base_index,
         feedback_fn, full_disk_params, _req_file_storage=opcodes.RequireFileStorage,
         _req_shr_file_storage=opcodes.RequireSharedFileStorage):
         """Generate the entire disk layout for a given template type.
         """
         vgname = lu.cfg.GetVGName()
         disk_count = len(disk_info)
         disks = []
         if template_name == constants.DT_DISKLESS:
           pass
         elif template_name == constants.DT_DRBD8:
           if len(secondary_nodes) != 1:
             raise errors.ProgrammerError("Wrong template configuration")
           remote_node = secondary_nodes[0]
           minors = lu.cfg.AllocateDRBDMinor(
             [primary_node, remote_node] * len(disk_info), instance_name)
           (drbd_params, _, _) = objects.Disk.ComputeLDParams(template_name,
                                                              full_disk_params)
           drbd_default_metavg = drbd_params[constants.LDP_DEFAULT_METAVG]
           names = []
           for lv_prefix in _GenerateUniqueNames(lu, [".disk%d" % (base_index + 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
             data_vg = disk.get(constants.IDISK_VG, vgname)
             meta_vg = disk.get(constants.IDISK_METAVG, drbd_default_metavg)
             disk_dev = _GenerateDRBD8Branch(lu, primary_node, remote_node,
                                             disk[constants.IDISK_SIZE],
                                             [data_vg, meta_vg],
                                             names[idx * 2:idx * 2 + 2],
                                             "disk/%d" % disk_index,
                                             minors[idx * 2], minors[idx * 2 + 1])
             disk_dev.mode = disk[constants.IDISK_MODE]
             disk_dev.name = disk.get(constants.IDISK_NAME, None)
             disks.append(disk_dev)
         else:
           if secondary_nodes:
             raise errors.ProgrammerError("Wrong template configuration")
           if template_name == constants.DT_FILE:
             _req_file_storage()
           elif template_name == constants.DT_SHARED_FILE:
             _req_shr_file_storage()
           name_prefix = _DISK_TEMPLATE_NAME_PREFIX.get(template_name, None)
           if name_prefix is None:
             names = None
           else:
             names = _GenerateUniqueNames(lu, ["%s.disk%s" %
                                               (name_prefix, base_index + i)
                                               for i in range(disk_count)])
           if template_name == constants.DT_PLAIN:
             def logical_id_fn(idx, _, disk):
               vg = disk.get(constants.IDISK_VG, vgname)
               return (vg, names[idx])
           elif template_name in (constants.DT_FILE, constants.DT_SHARED_FILE):
             logical_id_fn = \
               lambda _, disk_index, disk: (file_driver,
                                            "%s/disk%d" % (file_storage_dir,
                                                           disk_index))
           elif template_name == constants.DT_BLOCK:
             logical_id_fn = \
               lambda idx, disk_index, disk: (constants.BLOCKDEV_DRIVER_MANUAL,
                                              disk[constants.IDISK_ADOPT])
           elif template_name == constants.DT_RBD:
             logical_id_fn = lambda idx, _, disk: ("rbd", names[idx])
           elif template_name == constants.DT_EXT:
             def logical_id_fn(idx, _, disk):
               provider = disk.get(constants.IDISK_PROVIDER, None)
               if provider is None:
                 raise errors.ProgrammerError("Disk template is %s, but '%s' is"
                                              " not found", constants.DT_EXT,
                                              constants.IDISK_PROVIDER)
               return (provider, names[idx])
           else:
             raise errors.ProgrammerError("Unknown disk template '%s'" % template_name)
           dev_type = _DISK_TEMPLATE_DEVICE_TYPE[template_name]
           for idx, disk in enumerate(disk_info):
             params = {}
             # Only for the Ext template add disk_info to params
             if template_name == constants.DT_EXT:
               params[constants.IDISK_PROVIDER] = disk[constants.IDISK_PROVIDER]
               for key in disk:
                 if key not in constants.IDISK_PARAMS:
                   params[key] = disk[key]
             disk_index = idx + base_index
             size = disk[constants.IDISK_SIZE]
             feedback_fn("* disk %s, size %s" %
                         (disk_index, utils.FormatUnit(size, "h")))
             disk_dev = objects.Disk(dev_type=dev_type, size=size,
                                     logical_id=logical_id_fn(idx, disk_index, disk),
                                     iv_name="disk/%d" % disk_index,
                                     mode=disk[constants.IDISK_MODE],
                                     params=params)
             disk_dev.name = disk.get(constants.IDISK_NAME, None)
             disk_dev.uuid = lu.cfg.GenerateUniqueID(lu.proc.GetECId())
             disks.append(disk_dev)
         return disks
       def _GetInstanceInfoText(instance):
         """Compute that text that should be added to the disk's metadata.
         """
         return "originstname+%s" % instance.name
       def _CalcEta(time_taken, written, total_size):
         """Calculates the ETA based on size written and total size.
         @param time_taken: The time taken so far
         @param written: amount written so far
         @param total_size: The total size of data to be written
         @return: The remaining time in seconds
         ""&qu

Synnefo » snf-ganeti

root / lib / cmdlib / __init__.py @ 31b836b8

root / lib / cmdlib / init.py @ 31b836b8