Source Code for Module ganeti.cmdlib.instance_migration

   1  # 
   2  # 
   3   
   4  # Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 Google Inc. 
   5  # All rights reserved. 
   6  # 
   7  # Redistribution and use in source and binary forms, with or without 
   8  # modification, are permitted provided that the following conditions are 
   9  # met: 
  10  # 
  11  # 1. Redistributions of source code must retain the above copyright notice, 
  12  # this list of conditions and the following disclaimer. 
  13  # 
  14  # 2. Redistributions in binary form must reproduce the above copyright 
  15  # notice, this list of conditions and the following disclaimer in the 
  16  # documentation and/or other materials provided with the distribution. 
  17  # 
  18  # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS 
  19  # IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED 
  20  # TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR 
  21  # PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR 
  22  # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
  23  # EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
  24  # PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
  25  # PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
  26  # LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
  27  # NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
  28  # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 
  29   
  30   
  31  """Logical units dealing with instance migration an failover.""" 
  32   
  33  import logging 
  34  import time 
  35   
  36  from ganeti import constants 
  37  from ganeti import errors 
  38  from ganeti import locking 
  39  from ganeti import hypervisor 
  40  from ganeti.masterd import iallocator 
  41  from ganeti import utils 
  42  from ganeti.cmdlib.base import LogicalUnit, Tasklet 
  43  from ganeti.cmdlib.common import ExpandInstanceUuidAndName, \ 
  44    CheckIAllocatorOrNode, ExpandNodeUuidAndName 
  45  from ganeti.cmdlib.instance_storage import CheckDiskConsistency, \ 
  46    ExpandCheckDisks, ShutdownInstanceDisks, AssembleInstanceDisks 
  47  from ganeti.cmdlib.instance_utils import BuildInstanceHookEnvByObject, \ 
  48    CheckTargetNodeIPolicy, ReleaseLocks, CheckNodeNotDrained, \ 
  49    CopyLockList, CheckNodeFreeMemory, CheckInstanceBridgesExist 
  50   
  51  import ganeti.masterd.instance 
  52   
  53   
  54 -def _ExpandNamesForMigration(lu): 
  55    """Expands names for use with L{TLMigrateInstance}. 
  56   
  57    @type lu: L{LogicalUnit} 
  58   
  59    """ 
  60    if lu.op.target_node is not None: 
  61      (lu.op.target_node_uuid, lu.op.target_node) = \ 
  62        ExpandNodeUuidAndName(lu.cfg, lu.op.target_node_uuid, lu.op.target_node) 
  63   
  64    lu.needed_locks[locking.LEVEL_NODE] = [] 
  65    lu.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE 
  66    lu.dont_collate_locks[locking.LEVEL_NODE] = True 
  67   
  68    lu.needed_locks[locking.LEVEL_NODE_RES] = [] 
  69    lu.recalculate_locks[locking.LEVEL_NODE_RES] = constants.LOCKS_REPLACE 
  70    lu.dont_collate_locks[locking.LEVEL_NODE_RES] = True 
  71   
  72   
  73 -def _DeclareLocksForMigration(lu, level): 
  74    """Declares locks for L{TLMigrateInstance}. 
  75   
  76    @type lu: L{LogicalUnit} 
  77    @param level: Lock level 
  78   
  79    """ 
  80    if level == locking.LEVEL_NODE: 
  81      assert lu.op.instance_name in lu.owned_locks(locking.LEVEL_INSTANCE) 
  82   
  83      instance = lu.cfg.GetInstanceInfo(lu.op.instance_uuid) 
  84   
  85      disks = lu.cfg.GetInstanceDisks(instance.uuid) 
  86      if utils.AnyDiskOfType(disks, constants.DTS_EXT_MIRROR): 
  87        if lu.op.target_node is None: 
  88          lu.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET 
  89        else: 
  90          lu.needed_locks[locking.LEVEL_NODE] = [instance.primary_node, 
  91                                                 lu.op.target_node_uuid] 
  92      else: 
  93        lu._LockInstancesNodes() # pylint: disable=W0212 
  94   
  95      assert (lu.needed_locks[locking.LEVEL_NODE] or 
  96              lu.needed_locks[locking.LEVEL_NODE] is locking.ALL_SET) 
  97   
  98    elif level == locking.LEVEL_NODE_RES: 
  99      # Copy node locks 
 100      lu.needed_locks[locking.LEVEL_NODE_RES] = \ 
 101        CopyLockList(lu.needed_locks[locking.LEVEL_NODE]) 
 102   
 103   
 104 -class LUInstanceFailover(LogicalUnit): 
 105    """Failover an instance. 
 106   
 107    This is migration by shutting the instance down, but with the disks 
 108    of the instance already available on the new node. 
 109   
 110    See also: 
 111    L{LUInstanceMove} for moving an instance by copying the data. 
 112   
 113    L{LUInstanceMigrate} for the live migration of an instance (no shutdown 
 114    required). 
 115    """ 
 116    HPATH = "instance-failover" 
 117    HTYPE = constants.HTYPE_INSTANCE 
 118    REQ_BGL = False 
 119   
 120 -  def CheckArguments(self): 
 121      """Check the arguments. 
 122   
 123      """ 
 124      self.iallocator = getattr(self.op, "iallocator", None) 
 125      self.target_node = getattr(self.op, "target_node", None) 
 126   
 127 -  def ExpandNames(self): 
 128      self._ExpandAndLockInstance(allow_forthcoming=True) 
 129      _ExpandNamesForMigration(self) 
 130   
 131      self._migrater = \ 
 132        TLMigrateInstance(self, self.op.instance_uuid, self.op.instance_name, 
 133                          self.op.cleanup, True, False, 
 134                          self.op.ignore_consistency, True, 
 135                          self.op.shutdown_timeout, self.op.ignore_ipolicy, True) 
 136   
 137      self.tasklets = [self._migrater] 
 138   
 139 -  def DeclareLocks(self, level): 
 140      _DeclareLocksForMigration(self, level) 
 141   
 142 -  def BuildHooksEnv(self): 
 143      """Build hooks env. 
 144   
 145      This runs on master, primary and secondary nodes of the instance. 
 146   
 147      """ 
 148      instance = self._migrater.instance 
 149      source_node_uuid = instance.primary_node 
 150      target_node_uuid = self._migrater.target_node_uuid 
 151      env = { 
 152        "IGNORE_CONSISTENCY": self.op.ignore_consistency, 
 153        "SHUTDOWN_TIMEOUT": self.op.shutdown_timeout, 
 154        "OLD_PRIMARY": self.cfg.GetNodeName(source_node_uuid), 
 155        "NEW_PRIMARY": self.cfg.GetNodeName(target_node_uuid), 
 156        "FAILOVER_CLEANUP": self.op.cleanup, 
 157        } 
 158   
 159      disks = self.cfg.GetInstanceDisks(instance.uuid) 
 160      if utils.AnyDiskOfType(disks, constants.DTS_INT_MIRROR): 
 161        secondary_nodes = self.cfg.GetInstanceSecondaryNodes(instance.uuid) 
 162        env["OLD_SECONDARY"] = self.cfg.GetNodeName(secondary_nodes[0]) 
 163        env["NEW_SECONDARY"] = self.cfg.GetNodeName(source_node_uuid) 
 164      else: 
 165        env["OLD_SECONDARY"] = env["NEW_SECONDARY"] = "" 
 166   
 167      env.update(BuildInstanceHookEnvByObject(self, instance)) 
 168   
 169      return env 
 170   
 171 -  def BuildHooksNodes(self): 
 172      """Build hooks nodes. 
 173   
 174      """ 
 175      instance = self._migrater.instance 
 176      secondary_nodes = self.cfg.GetInstanceSecondaryNodes(instance.uuid) 
 177      nl = [self.cfg.GetMasterNode()] + list(secondary_nodes) 
 178      nl.append(self._migrater.target_node_uuid) 
 179      return (nl, nl + [instance.primary_node]) 
 180   
 181   
 182 -class LUInstanceMigrate(LogicalUnit): 
 183    """Migrate an instance. 
 184   
 185    This is migration without shutting down (live migration) and the disks are 
 186    already available on the new node. 
 187   
 188    See also: 
 189    L{LUInstanceMove} for moving an instance by copying the data. 
 190   
 191    L{LUInstanceFailover} for the migration of an instance where a shutdown is 
 192    required. 
 193    """ 
 194    HPATH = "instance-migrate" 
 195    HTYPE = constants.HTYPE_INSTANCE 
 196    REQ_BGL = False 
 197   
 198 -  def ExpandNames(self): 
 199      self._ExpandAndLockInstance() 
 200      _ExpandNamesForMigration(self) 
 201   
 202      self._migrater = \ 
 203        TLMigrateInstance(self, self.op.instance_uuid, self.op.instance_name, 
 204                          self.op.cleanup, False, self.op.allow_failover, False, 
 205                          self.op.allow_runtime_changes, 
 206                          constants.DEFAULT_SHUTDOWN_TIMEOUT, 
 207                          self.op.ignore_ipolicy, self.op.ignore_hvversions) 
 208   
 209      self.tasklets = [self._migrater] 
 210   
 211 -  def DeclareLocks(self, level): 
 212      _DeclareLocksForMigration(self, level) 
 213   
 214 -  def BuildHooksEnv(self): 
 215      """Build hooks env. 
 216   
 217      This runs on master, primary and secondary nodes of the instance. 
 218   
 219      """ 
 220      instance = self._migrater.instance 
 221      source_node_uuid = instance.primary_node 
 222      target_node_uuid = self._migrater.target_node_uuid 
 223      env = BuildInstanceHookEnvByObject(self, instance) 
 224      env.update({ 
 225        "MIGRATE_LIVE": self._migrater.live, 
 226        "MIGRATE_CLEANUP": self.op.cleanup, 
 227        "OLD_PRIMARY": self.cfg.GetNodeName(source_node_uuid), 
 228        "NEW_PRIMARY": self.cfg.GetNodeName(target_node_uuid), 
 229        "ALLOW_RUNTIME_CHANGES": self.op.allow_runtime_changes, 
 230        }) 
 231   
 232      disks = self.cfg.GetInstanceDisks(instance.uuid) 
 233      if utils.AnyDiskOfType(disks, constants.DTS_INT_MIRROR): 
 234        secondary_nodes = self.cfg.GetInstanceSecondaryNodes(instance.uuid) 
 235        env["OLD_SECONDARY"] = self.cfg.GetNodeName(secondary_nodes[0]) 
 236        env["NEW_SECONDARY"] = self.cfg.GetNodeName(source_node_uuid) 
 237      else: 
 238        env["OLD_SECONDARY"] = env["NEW_SECONDARY"] = "" 
 239   
 240      return env 
 241   
 242 -  def BuildHooksNodes(self): 
 243      """Build hooks nodes. 
 244   
 245      """ 
 246      instance = self._migrater.instance 
 247      secondary_nodes = self.cfg.GetInstanceSecondaryNodes(instance.uuid) 
 248      snode_uuids = list(secondary_nodes) 
 249      nl = [self.cfg.GetMasterNode(), instance.primary_node] + snode_uuids 
 250      nl.append(self._migrater.target_node_uuid) 
 251      return (nl, nl) 
 252   
 253   
 254 -class TLMigrateInstance(Tasklet): 
 255    """Tasklet class for instance migration. 
 256   
 257    @type live: boolean 
 258    @ivar live: whether the migration will be done live or non-live; 
 259        this variable is initalized only after CheckPrereq has run 
 260    @type cleanup: boolean 
 261    @ivar cleanup: Wheater we cleanup from a failed migration 
 262    @type iallocator: string 
 263    @ivar iallocator: The iallocator used to determine target_node 
 264    @type target_node_uuid: string 
 265    @ivar target_node_uuid: If given, the target node UUID to reallocate the 
 266        instance to 
 267    @type failover: boolean 
 268    @ivar failover: Whether operation results in failover or migration 
 269    @type fallback: boolean 
 270    @ivar fallback: Whether fallback to failover is allowed if migration not 
 271                    possible 
 272    @type ignore_consistency: boolean 
 273    @ivar ignore_consistency: Wheter we should ignore consistency between source 
 274                              and target node 
 275    @type shutdown_timeout: int 
 276    @ivar shutdown_timeout: In case of failover timeout of the shutdown 
 277    @type ignore_ipolicy: bool 
 278    @ivar ignore_ipolicy: If true, we can ignore instance policy when migrating 
 279    @type ignore_hvversions: bool 
 280    @ivar ignore_hvversions: If true, accept incompatible hypervisor versions 
 281   
 282    """ 
 283   
 284    # Constants 
 285    _MIGRATION_POLL_INTERVAL = 1      # seconds 
 286    _MIGRATION_FEEDBACK_INTERVAL = 10 # seconds 
 287   
 288 -  def __init__(self, lu, instance_uuid, instance_name, cleanup, failover, 
 289                 fallback, ignore_consistency, allow_runtime_changes, 
 290                 shutdown_timeout, ignore_ipolicy, ignore_hvversions): 
 291      """Initializes this class. 
 292   
 293      """ 
 294      Tasklet.__init__(self, lu) 
 295   
 296      # Parameters 
 297      self.instance_uuid = instance_uuid 
 298      self.instance_name = instance_name 
 299      self.cleanup = cleanup 
 300      self.live = False # will be overridden later 
 301      self.failover = failover 
 302      self.fallback = fallback 
 303      self.ignore_consistency = ignore_consistency 
 304      self.shutdown_timeout = shutdown_timeout 
 305      self.ignore_ipolicy = ignore_ipolicy 
 306      self.allow_runtime_changes = allow_runtime_changes 
 307      self.ignore_hvversions = ignore_hvversions 
 308   
 309 -  def CheckPrereq(self): 
 310      """Check prerequisites. 
 311   
 312      This checks that the instance is in the cluster. 
 313   
 314      """ 
 315      (self.instance_uuid, self.instance_name) = \ 
 316        ExpandInstanceUuidAndName(self.lu.cfg, self.instance_uuid, 
 317                                  self.instance_name) 
 318      self.instance = self.cfg.GetInstanceInfo(self.instance_uuid) 
 319      assert self.instance is not None 
 320      cluster = self.cfg.GetClusterInfo() 
 321   
 322      if (not self.cleanup and 
 323          not self.instance.admin_state == constants.ADMINST_UP and 
 324          not self.failover and self.fallback): 
 325        self.lu.LogInfo("Instance is marked down or offline, fallback allowed," 
 326                        " switching to failover") 
 327        self.failover = True 
 328   
 329      disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
 330   
 331      if not utils.AllDiskOfType(disks, constants.DTS_MIRRORED): 
 332        if self.failover: 
 333          text = "failovers" 
 334        else: 
 335          text = "migrations" 
 336        invalid_disks = set(d.dev_type for d in disks 
 337                               if d.dev_type not in constants.DTS_MIRRORED) 
 338        raise errors.OpPrereqError("Instance's disk layout '%s' does not allow" 
 339                                   " %s" % (utils.CommaJoin(invalid_disks), text), 
 340                                   errors.ECODE_STATE) 
 341   
 342      # TODO allow heterogeneous disk types if all are mirrored in some way. 
 343      if utils.AllDiskOfType(disks, constants.DTS_EXT_MIRROR): 
 344        CheckIAllocatorOrNode(self.lu, "iallocator", "target_node") 
 345   
 346        if self.lu.op.iallocator: 
 347          self._RunAllocator() 
 348        else: 
 349          # We set set self.target_node_uuid as it is required by 
 350          # BuildHooksEnv 
 351          self.target_node_uuid = self.lu.op.target_node_uuid 
 352   
 353        # Check that the target node is correct in terms of instance policy 
 354        nodeinfo = self.cfg.GetNodeInfo(self.target_node_uuid) 
 355        group_info = self.cfg.GetNodeGroup(nodeinfo.group) 
 356        ipolicy = ganeti.masterd.instance.CalculateGroupIPolicy(cluster, 
 357                                                                group_info) 
 358        CheckTargetNodeIPolicy(self.lu, ipolicy, self.instance, nodeinfo, 
 359                               self.cfg, ignore=self.ignore_ipolicy) 
 360   
 361        # self.target_node is already populated, either directly or by the 
 362        # iallocator run 
 363        target_node_uuid = self.target_node_uuid 
 364        if self.target_node_uuid == self.instance.primary_node: 
 365          raise errors.OpPrereqError( 
 366            "Cannot migrate instance %s to its primary (%s)" % 
 367            (self.instance.name, 
 368             self.cfg.GetNodeName(self.instance.primary_node)), 
 369            errors.ECODE_STATE) 
 370   
 371        if len(self.lu.tasklets) == 1: 
 372          # It is safe to release locks only when we're the only tasklet 
 373          # in the LU 
 374          ReleaseLocks(self.lu, locking.LEVEL_NODE, 
 375                       keep=[self.instance.primary_node, self.target_node_uuid]) 
 376   
 377      elif utils.AllDiskOfType(disks, constants.DTS_INT_MIRROR): 
 378        templates = [d.dev_type for d in disks] 
 379        secondary_node_uuids = \ 
 380          self.cfg.GetInstanceSecondaryNodes(self.instance.uuid) 
 381        if not secondary_node_uuids: 
 382          raise errors.ConfigurationError("No secondary node but using" 
 383                                          " %s disk types" % 
 384                                          utils.CommaJoin(set(templates))) 
 385        self.target_node_uuid = target_node_uuid = secondary_node_uuids[0] 
 386        if self.lu.op.iallocator or \ 
 387          (self.lu.op.target_node_uuid and 
 388           self.lu.op.target_node_uuid != target_node_uuid): 
 389          if self.failover: 
 390            text = "failed over" 
 391          else: 
 392            text = "migrated" 
 393          raise errors.OpPrereqError("Instances with disk types %s cannot" 
 394                                     " be %s to arbitrary nodes" 
 395                                     " (neither an iallocator nor a target" 
 396                                     " node can be passed)" % 
 397                                     (utils.CommaJoin(set(templates)), text), 
 398                                     errors.ECODE_INVAL) 
 399        nodeinfo = self.cfg.GetNodeInfo(target_node_uuid) 
 400        group_info = self.cfg.GetNodeGroup(nodeinfo.group) 
 401        ipolicy = ganeti.masterd.instance.CalculateGroupIPolicy(cluster, 
 402                                                                group_info) 
 403        CheckTargetNodeIPolicy(self.lu, ipolicy, self.instance, nodeinfo, 
 404                               self.cfg, ignore=self.ignore_ipolicy) 
 405   
 406      else: 
 407        raise errors.OpPrereqError("Instance mixes internal and external " 
 408                                   "mirroring. This is not currently supported.") 
 409   
 410      i_be = cluster.FillBE(self.instance) 
 411   
 412      # check memory requirements on the secondary node 
 413      if (not self.cleanup and 
 414           (not self.failover or 
 415             self.instance.admin_state == constants.ADMINST_UP)): 
 416        self.tgt_free_mem = CheckNodeFreeMemory( 
 417            self.lu, target_node_uuid, 
 418            "migrating instance %s" % self.instance.name, 
 419            i_be[constants.BE_MINMEM], self.instance.hypervisor, 
 420            self.cfg.GetClusterInfo().hvparams[self.instance.hypervisor]) 
 421      else: 
 422        self.lu.LogInfo("Not checking memory on the secondary node as" 
 423                        " instance will not be started") 
 424   
 425      # check if failover must be forced instead of migration 
 426      if (not self.cleanup and not self.failover and 
 427          i_be[constants.BE_ALWAYS_FAILOVER]): 
 428        self.lu.LogInfo("Instance configured to always failover; fallback" 
 429                        " to failover") 
 430        self.failover = True 
 431   
 432      # check bridge existance 
 433      CheckInstanceBridgesExist(self.lu, self.instance, 
 434                                node_uuid=target_node_uuid) 
 435   
 436      if not self.cleanup: 
 437        CheckNodeNotDrained(self.lu, target_node_uuid) 
 438        if not self.failover: 
 439          result = self.rpc.call_instance_migratable(self.instance.primary_node, 
 440                                                     self.instance) 
 441          if result.fail_msg and self.fallback: 
 442            self.lu.LogInfo("Can't migrate, instance offline, fallback to" 
 443                            " failover") 
 444            self.failover = True 
 445          else: 
 446            result.Raise("Can't migrate, please use failover", 
 447                         prereq=True, ecode=errors.ECODE_STATE) 
 448   
 449      assert not (self.failover and self.cleanup) 
 450   
 451      if not self.failover: 
 452        if self.lu.op.live is not None and self.lu.op.mode is not None: 
 453          raise errors.OpPrereqError("Only one of the 'live' and 'mode'" 
 454                                     " parameters are accepted", 
 455                                     errors.ECODE_INVAL) 
 456        if self.lu.op.live is not None: 
 457          if self.lu.op.live: 
 458            self.lu.op.mode = constants.HT_MIGRATION_LIVE 
 459          else: 
 460            self.lu.op.mode = constants.HT_MIGRATION_NONLIVE 
 461          # reset the 'live' parameter to None so that repeated 
 462          # invocations of CheckPrereq do not raise an exception 
 463          self.lu.op.live = None 
 464        elif self.lu.op.mode is None: 
 465          # read the default value from the hypervisor 
 466          i_hv = cluster.FillHV(self.instance, skip_globals=False) 
 467          self.lu.op.mode = i_hv[constants.HV_MIGRATION_MODE] 
 468   
 469        self.live = self.lu.op.mode == constants.HT_MIGRATION_LIVE 
 470      else: 
 471        # Failover is never live 
 472        self.live = False 
 473   
 474      if not (self.failover or self.cleanup): 
 475        remote_info = self.rpc.call_instance_info( 
 476            self.instance.primary_node, self.instance.name, 
 477            self.instance.hypervisor, cluster.hvparams[self.instance.hypervisor]) 
 478        remote_info.Raise("Error checking instance on node %s" % 
 479                          self.cfg.GetNodeName(self.instance.primary_node), 
 480                          prereq=True) 
 481        instance_running = bool(remote_info.payload) 
 482        if instance_running: 
 483          self.current_mem = int(remote_info.payload["memory"]) 
 484   
 485 -  def _RunAllocator(self): 
 486      """Run the allocator based on input opcode. 
 487   
 488      """ 
 489      # FIXME: add a self.ignore_ipolicy option 
 490      req = iallocator.IAReqRelocate( 
 491            inst_uuid=self.instance_uuid, 
 492            relocate_from_node_uuids=[self.instance.primary_node]) 
 493      ial = iallocator.IAllocator(self.cfg, self.rpc, req) 
 494   
 495      ial.Run(self.lu.op.iallocator) 
 496   
 497      if not ial.success: 
 498        raise errors.OpPrereqError("Can't compute nodes using" 
 499                                   " iallocator '%s': %s" % 
 500                                   (self.lu.op.iallocator, ial.info), 
 501                                   errors.ECODE_NORES) 
 502      self.target_node_uuid = self.cfg.GetNodeInfoByName(ial.result[0]).uuid 
 503      self.lu.LogInfo("Selected nodes for instance %s via iallocator %s: %s", 
 504                      self.instance_name, self.lu.op.iallocator, 
 505                      utils.CommaJoin(ial.result)) 
 506   
 507 -  def _WaitUntilSync(self): 
 508      """Poll with custom rpc for disk sync. 
 509   
 510      This uses our own step-based rpc call. 
 511   
 512      """ 
 513      self.feedback_fn("* wait until resync is done") 
 514      all_done = False 
 515      disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
 516      while not all_done: 
 517        all_done = True 
 518        result = self.rpc.call_drbd_wait_sync(self.all_node_uuids, 
 519                                              (disks, self.instance)) 
 520        min_percent = 100 
 521        for node_uuid, nres in result.items(): 
 522          nres.Raise("Cannot resync disks on node %s" % 
 523                     self.cfg.GetNodeName(node_uuid)) 
 524          node_done, node_percent = nres.payload 
 525          all_done = all_done and node_done 
 526          if node_percent is not None: 
 527            min_percent = min(min_percent, node_percent) 
 528        if not all_done: 
 529          if min_percent < 100: 
 530            self.feedback_fn("   - progress: %.1f%%" % min_percent) 
 531          time.sleep(2) 
 532   
 533 -  def _OpenInstanceDisks(self, node_uuid, exclusive): 
 534      """Open instance disks. 
 535   
 536      """ 
 537      if exclusive: 
 538        mode = "in exclusive mode" 
 539      else: 
 540        mode = "in shared mode" 
 541   
 542      node_name = self.cfg.GetNodeName(node_uuid) 
 543   
 544      self.feedback_fn("* opening instance disks on node %s %s" % 
 545                       (node_name, mode)) 
 546   
 547      disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
 548      result = self.rpc.call_blockdev_open(node_uuid, self.instance.name, 
 549                                           (disks, self.instance), exclusive) 
 550      result.Raise("Cannot open instance disks on node %s" % node_name) 
 551   
 552 -  def _CloseInstanceDisks(self, node_uuid): 
 553      """Close instance disks. 
 554   
 555      """ 
 556      node_name = self.cfg.GetNodeName(node_uuid) 
 557   
 558      self.feedback_fn("* closing instance disks on node %s" % node_name) 
 559   
 560      disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
 561      result = self.rpc.call_blockdev_close(node_uuid, self.instance.name, 
 562                                            (disks, self.instance)) 
 563      msg = result.fail_msg 
 564      if msg: 
 565        if result.offline or self.ignore_consistency: 
 566          self.lu.LogWarning("Could not close instance disks on node %s," 
 567                             " proceeding anyway" % node_name) 
 568        else: 
 569          raise errors.OpExecError("Cannot close instance disks on node %s: %s" % 
 570                                   (node_name, msg)) 
 571   
 572 -  def _GoStandalone(self): 
 573      """Disconnect from the network. 
 574   
 575      """ 
 576      self.feedback_fn("* changing into standalone mode") 
 577      disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
 578      result = self.rpc.call_drbd_disconnect_net( 
 579                 self.all_node_uuids, (disks, self.instance)) 
 580      for node_uuid, nres in result.items(): 
 581        nres.Raise("Cannot disconnect disks node %s" % 
 582                   self.cfg.GetNodeName(node_uuid)) 
 583   
 584 -  def _GoReconnect(self, multimaster): 
 585      """Reconnect to the network. 
 586   
 587      """ 
 588      if multimaster: 
 589        msg = "dual-master" 
 590      else: 
 591        msg = "single-master" 
 592      self.feedback_fn("* changing disks into %s mode" % msg) 
 593      disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
 594      result = self.rpc.call_drbd_attach_net(self.all_node_uuids, 
 595                                             (disks, self.instance), 
 596                                             multimaster) 
 597      for node_uuid, nres in result.items(): 
 598        nres.Raise("Cannot change disks config on node %s" % 
 599                   self.cfg.GetNodeName(node_uuid)) 
 600   
 601 -  def _ExecCleanup(self): 
 602      """Try to cleanup after a failed migration. 
 603   
 604      The cleanup is done by: 
 605        - check that the instance is running only on one node 
 606          (and update the config if needed) 
 607        - change disks on its secondary node to secondary 
 608        - wait until disks are fully synchronized 
 609        - disconnect from the network 
 610        - change disks into single-master mode 
 611        - wait again until disks are fully synchronized 
 612   
 613      """ 
 614      # check running on only one node 
 615      self.feedback_fn("* checking where the instance actually runs" 
 616                       " (if this hangs, the hypervisor might be in" 
 617                       " a bad state)") 
 618      cluster_hvparams = self.cfg.GetClusterInfo().hvparams 
 619      ins_l = self.rpc.call_instance_list(self.all_node_uuids, 
 620                                          [self.instance.hypervisor], 
 621                                          cluster_hvparams) 
 622      for node_uuid, result in ins_l.items(): 
 623        result.Raise("Can't contact node %s" % node_uuid) 
 624   
 625      runningon_source = self.instance.name in \ 
 626                           ins_l[self.source_node_uuid].payload 
 627      runningon_target = self.instance.name in \ 
 628                           ins_l[self.target_node_uuid].payload 
 629   
 630      if runningon_source and runningon_target: 
 631        raise errors.OpExecError("Instance seems to be running on two nodes," 
 632                                 " or the hypervisor is confused; you will have" 
 633                                 " to ensure manually that it runs only on one" 
 634                                 " and restart this operation") 
 635   
 636      if not (runningon_source or runningon_target): 
 637        raise errors.OpExecError("Instance does not seem to be running at all;" 
 638                                 " in this case it's safer to repair by" 
 639                                 " running 'gnt-instance stop' to ensure disk" 
 640                                 " shutdown, and then restarting it") 
 641   
 642      if runningon_target: 
 643        # the migration has actually succeeded, we need to update the config 
 644        self.feedback_fn("* instance running on secondary node (%s)," 
 645                         " updating config" % 
 646                         self.cfg.GetNodeName(self.target_node_uuid)) 
 647        self.cfg.SetInstancePrimaryNode(self.instance.uuid, 
 648                                        self.target_node_uuid) 
 649        demoted_node_uuid = self.source_node_uuid 
 650      else: 
 651        self.feedback_fn("* instance confirmed to be running on its" 
 652                         " primary node (%s)" % 
 653                         self.cfg.GetNodeName(self.source_node_uuid)) 
 654        demoted_node_uuid = self.target_node_uuid 
 655   
 656      disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
 657   
 658      self._CloseInstanceDisks(demoted_node_uuid) 
 659   
 660      if utils.AnyDiskOfType(disks, constants.DTS_INT_MIRROR): 
 661        try: 
 662          self._WaitUntilSync() 
 663        except errors.OpExecError: 
 664          # we ignore here errors, since if the device is standalone, it 
 665          # won't be able to sync 
 666          pass 
 667        self._GoStandalone() 
 668        self._GoReconnect(False) 
 669        self._WaitUntilSync() 
 670      elif utils.AnyDiskOfType(disks, constants.DTS_EXT_MIRROR): 
 671        self._OpenInstanceDisks(self.instance.primary_node, True) 
 672   
 673      self.feedback_fn("* done") 
 674   
 675 -  def _RevertDiskStatus(self): 
 676      """Try to revert the disk status after a failed migration. 
 677   
 678      """ 
 679   
 680      disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
 681   
 682      self._CloseInstanceDisks(self.target_node_uuid) 
 683   
 684      if utils.AllDiskOfType(disks, constants.DTS_EXT_MIRROR): 
 685        self._OpenInstanceDisks(self.source_node_uuid, True) 
 686        return 
 687   
 688      try: 
 689        self._GoStandalone() 
 690        self._GoReconnect(False) 
 691        self._WaitUntilSync() 
 692      except errors.OpExecError, err: 
 693        self.lu.LogWarning("Migration failed and I can't reconnect the drives," 
 694                           " please try to recover the instance manually;" 
 695                           " error '%s'" % str(err)) 
 696   
 697 -  def _AbortMigration(self): 
 698      """Call the hypervisor code to abort a started migration. 
 699   
 700      """ 
 701      abort_result = self.rpc.call_instance_finalize_migration_dst( 
 702                       self.target_node_uuid, self.instance, self.migration_info, 
 703                       False) 
 704      abort_msg = abort_result.fail_msg 
 705      if abort_msg: 
 706        logging.error("Aborting migration failed on target node %s: %s", 
 707                      self.cfg.GetNodeName(self.target_node_uuid), abort_msg) 
 708        # Don't raise an exception here, as we stil have to try to revert the 
 709        # disk status, even if this step failed. 
 710   
 711      abort_result = self.rpc.call_instance_finalize_migration_src( 
 712        self.source_node_uuid, self.instance, False, self.live) 
 713      abort_msg = abort_result.fail_msg 
 714      if abort_msg: 
 715        logging.error("Aborting migration failed on source node %s: %s", 
 716                      self.cfg.GetNodeName(self.source_node_uuid), abort_msg) 
 717   
 718 -  def _ExecMigration(self): 
 719      """Migrate an instance. 
 720   
 721      The migrate is done by: 
 722        - change the disks into dual-master mode 
 723        - wait until disks are fully synchronized again 
 724        - migrate the instance 
 725        - change disks on the new secondary node (the old primary) to secondary 
 726        - wait until disks are fully synchronized 
 727        - change disks into single-master mode 
 728   
 729      """ 
 730      # Check for hypervisor version mismatch and warn the user. 
 731      hvspecs = [(self.instance.hypervisor, 
 732                  self.cfg.GetClusterInfo().hvparams[self.instance.hypervisor])] 
 733      nodeinfo = self.rpc.call_node_info( 
 734                   [self.source_node_uuid, self.target_node_uuid], None, hvspecs) 
 735      for ninfo in nodeinfo.values(): 
 736        ninfo.Raise("Unable to retrieve node information from node '%s'" % 
 737                    ninfo.node) 
 738      (_, _, (src_info, )) = nodeinfo[self.source_node_uuid].payload 
 739      (_, _, (dst_info, )) = nodeinfo[self.target_node_uuid].payload 
 740   
 741      if ((constants.HV_NODEINFO_KEY_VERSION in src_info) and 
 742          (constants.HV_NODEINFO_KEY_VERSION in dst_info)): 
 743        src_version = src_info[constants.HV_NODEINFO_KEY_VERSION] 
 744        dst_version = dst_info[constants.HV_NODEINFO_KEY_VERSION] 
 745        if src_version != dst_version: 
 746          self.feedback_fn("* warning: hypervisor version mismatch between" 
 747                           " source (%s) and target (%s) node" % 
 748                           (src_version, dst_version)) 
 749          hv = hypervisor.GetHypervisorClass(self.instance.hypervisor) 
 750          if hv.VersionsSafeForMigration(src_version, dst_version): 
 751            self.feedback_fn("  migrating from hypervisor version %s to %s should" 
 752                             " be safe" % (src_version, dst_version)) 
 753          else: 
 754            self.feedback_fn("  migrating from hypervisor version %s to %s is" 
 755                             " likely unsupported" % (src_version, dst_version)) 
 756            if self.ignore_hvversions: 
 757              self.feedback_fn("  continuing anyway (told to ignore version" 
 758                               " mismatch)") 
 759            else: 
 760              raise errors.OpExecError("Unsupported migration between hypervisor" 
 761                                       " versions (%s to %s)" % 
 762                                       (src_version, dst_version)) 
 763   
 764      self.feedback_fn("* checking disk consistency between source and target") 
 765      for (idx, dev) in enumerate(self.cfg.GetInstanceDisks(self.instance.uuid)): 
 766        if not CheckDiskConsistency(self.lu, self.instance, dev, 
 767                                    self.target_node_uuid, 
 768                                    False): 
 769          raise errors.OpExecError("Disk %s is degraded or not fully" 
 770                                   " synchronized on target node," 
 771                                   " aborting migration" % idx) 
 772   
 773      if self.current_mem > self.tgt_free_mem: 
 774        if not self.allow_runtime_changes: 
 775          raise errors.OpExecError("Memory ballooning not allowed and not enough" 
 776                                   " free memory to fit instance %s on target" 
 777                                   " node %s (have %dMB, need %dMB)" % 
 778                                   (self.instance.name, 
 779                                    self.cfg.GetNodeName(self.target_node_uuid), 
 780                                    self.tgt_free_mem, self.current_mem)) 
 781        self.feedback_fn("* setting instance memory to %s" % self.tgt_free_mem) 
 782        rpcres = self.rpc.call_instance_balloon_memory(self.instance.primary_node, 
 783                                                       self.instance, 
 784                                                       self.tgt_free_mem) 
 785        rpcres.Raise("Cannot modify instance runtime memory") 
 786   
 787      # First get the migration information from the remote node 
 788      result = self.rpc.call_migration_info(self.source_node_uuid, self.instance) 
 789      msg = result.fail_msg 
 790      if msg: 
 791        log_err = ("Failed fetching source migration information from %s: %s" % 
 792                   (self.cfg.GetNodeName(self.source_node_uuid), msg)) 
 793        logging.error(log_err) 
 794        raise errors.OpExecError(log_err) 
 795   
 796      self.migration_info = migration_info = result.payload 
 797   
 798      disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
 799   
 800      self._CloseInstanceDisks(self.target_node_uuid) 
 801   
 802      if utils.AnyDiskOfType(disks, constants.DTS_INT_MIRROR): 
 803        # Then switch the disks to master/master mode 
 804        self._GoStandalone() 
 805        self._GoReconnect(True) 
 806        self._WaitUntilSync() 
 807   
 808      self._OpenInstanceDisks(self.source_node_uuid, False) 
 809      self._OpenInstanceDisks(self.target_node_uuid, False) 
 810   
 811      self.feedback_fn("* preparing %s to accept the instance" % 
 812                       self.cfg.GetNodeName(self.target_node_uuid)) 
 813      result = self.rpc.call_accept_instance(self.target_node_uuid, 
 814                                             self.instance, 
 815                                             migration_info, 
 816                                             self.nodes_ip[self.target_node_uuid]) 
 817   
 818      msg = result.fail_msg 
 819      if msg: 
 820        logging.error("Instance pre-migration failed, trying to revert" 
 821                      " disk status: %s", msg) 
 822        self.feedback_fn("Pre-migration failed, aborting") 
 823        self._AbortMigration() 
 824        self._RevertDiskStatus() 
 825        raise errors.OpExecError("Could not pre-migrate instance %s: %s" % 
 826                                 (self.instance.name, msg)) 
 827   
 828      self.feedback_fn("* migrating instance to %s" % 
 829                       self.cfg.GetNodeName(self.target_node_uuid)) 
 830      cluster = self.cfg.GetClusterInfo() 
 831      result = self.rpc.call_instance_migrate( 
 832          self.source_node_uuid, cluster.cluster_name, self.instance, 
 833          self.nodes_ip[self.target_node_uuid], self.live) 
 834      msg = result.fail_msg 
 835      if msg: 
 836        logging.error("Instance migration failed, trying to revert" 
 837                      " disk status: %s", msg) 
 838        self.feedback_fn("Migration failed, aborting") 
 839        self._AbortMigration() 
 840        self._RevertDiskStatus() 
 841        raise errors.OpExecError("Could not migrate instance %s: %s" % 
 842                                 (self.instance.name, msg)) 
 843   
 844      self.feedback_fn("* starting memory transfer") 
 845      last_feedback = time.time() 
 846      while True: 
 847        result = self.rpc.call_instance_get_migration_status( 
 848                   self.source_node_uuid, self.instance) 
 849        msg = result.fail_msg 
 850        ms = result.payload   # MigrationStatus instance 
 851        if msg or (ms.status in constants.HV_MIGRATION_FAILED_STATUSES): 
 852          logging.error("Instance migration failed, trying to revert" 
 853                        " disk status: %s", msg) 
 854          self.feedback_fn("Migration failed, aborting") 
 855          self._AbortMigration() 
 856          self._RevertDiskStatus() 
 857          if not msg: 
 858            msg = "hypervisor returned failure" 
 859          raise errors.OpExecError("Could not migrate instance %s: %s" % 
 860                                   (self.instance.name, msg)) 
 861   
 862        if result.payload.status != constants.HV_MIGRATION_ACTIVE: 
 863          self.feedback_fn("* memory transfer complete") 
 864          break 
 865   
 866        if (utils.TimeoutExpired(last_feedback, 
 867                                 self._MIGRATION_FEEDBACK_INTERVAL) and 
 868            ms.transferred_ram is not None): 
 869          mem_progress = 100 * float(ms.transferred_ram) / float(ms.total_ram) 
 870          self.feedback_fn("* memory transfer progress: %.2f %%" % mem_progress) 
 871          last_feedback = time.time() 
 872   
 873        time.sleep(self._MIGRATION_POLL_INTERVAL) 
 874   
 875      result = self.rpc.call_instance_finalize_migration_src( 
 876                 self.source_node_uuid, self.instance, True, self.live) 
 877      msg = result.fail_msg 
 878      if msg: 
 879        logging.error("Instance migration succeeded, but finalization failed" 
 880                      " on the source node: %s", msg) 
 881        raise errors.OpExecError("Could not finalize instance migration: %s" % 
 882                                 msg) 
 883   
 884      self.cfg.SetInstancePrimaryNode(self.instance.uuid, self.target_node_uuid) 
 885      self.instance = self.cfg.GetInstanceInfo(self.instance_uuid) 
 886      disks = self.cfg.GetInstanceDisks(self.instance_uuid) 
 887   
 888      result = self.rpc.call_instance_finalize_migration_dst( 
 889                 self.target_node_uuid, self.instance, migration_info, True) 
 890      msg = result.fail_msg 
 891      if msg: 
 892        logging.error("Instance migration succeeded, but finalization failed" 
 893                      " on the target node: %s", msg) 
 894        raise errors.OpExecError("Could not finalize instance migration: %s" % 
 895                                 msg) 
 896   
 897      self._CloseInstanceDisks(self.source_node_uuid) 
 898   
 899      if utils.AnyDiskOfType(disks, constants.DTS_INT_MIRROR): 
 900        self._WaitUntilSync() 
 901        self._GoStandalone() 
 902        self._GoReconnect(False) 
 903        self._WaitUntilSync() 
 904      elif utils.AnyDiskOfType(disks, constants.DTS_EXT_MIRROR): 
 905        self._OpenInstanceDisks(self.target_node_uuid, True) 
 906   
 907      # If the instance's disk template is `rbd' or `ext' and there was a 
 908      # successful migration, unmap the device from the source node. 
 909      unmap_types = (constants.DT_RBD, constants.DT_EXT) 
 910   
 911      if utils.AnyDiskOfType(disks, unmap_types): 
 912        unmap_disks = [d for d in disks if d.dev_type in unmap_types] 
 913        disks = ExpandCheckDisks(unmap_disks, unmap_disks) 
 914        self.feedback_fn("* unmapping instance's disks %s from %s" % 
 915                         (utils.CommaJoin(d.name for d in unmap_disks), 
 916                          self.cfg.GetNodeName(self.source_node_uuid))) 
 917        for disk in disks: 
 918          result = self.rpc.call_blockdev_shutdown(self.source_node_uuid, 
 919                                                   (disk, self.instance)) 
 920          msg = result.fail_msg 
 921          if msg: 
 922            logging.error("Migration was successful, but couldn't unmap the" 
 923                          " block device %s on source node %s: %s", 
 924                          disk.iv_name, 
 925                          self.cfg.GetNodeName(self.source_node_uuid), msg) 
 926            logging.error("You need to unmap the device %s manually on %s", 
 927                          disk.iv_name, 
 928                          self.cfg.GetNodeName(self.source_node_uuid)) 
 929   
 930      self.feedback_fn("* done") 
 931   
 932 -  def _ExecFailover(self): 
 933      """Failover an instance. 
 934   
 935      The failover is done by shutting it down on its present node and 
 936      starting it on the secondary. 
 937   
 938      """ 
 939      if self.instance.forthcoming: 
 940        self.feedback_fn("Instance is forthcoming, just updating the" 
 941                         "  configuration") 
 942        self.cfg.SetInstancePrimaryNode(self.instance.uuid, 
 943                                        self.target_node_uuid) 
 944        return 
 945   
 946      primary_node = self.cfg.GetNodeInfo(self.instance.primary_node) 
 947   
 948      source_node_uuid = self.instance.primary_node 
 949   
 950      if self.instance.disks_active: 
 951        self.feedback_fn("* checking disk consistency between source and target") 
 952        inst_disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
 953        for (idx, dev) in enumerate(inst_disks): 
 954          # for drbd, these are drbd over lvm 
 955          if not CheckDiskConsistency(self.lu, self.instance, dev, 
 956                                      self.target_node_uuid, False): 
 957            if primary_node.offline: 
 958              self.feedback_fn("Node %s is offline, ignoring degraded disk %s on" 
 959                               " target node %s" % 
 960                               (primary_node.name, idx, 
 961                                self.cfg.GetNodeName(self.target_node_uuid))) 
 962            elif not self.ignore_consistency: 
 963              raise errors.OpExecError("Disk %s is degraded on target node," 
 964                                       " aborting failover" % idx) 
 965      else: 
 966        self.feedback_fn("* not checking disk consistency as instance is not" 
 967                         " running") 
 968   
 969      self.feedback_fn("* shutting down instance on source node") 
 970      logging.info("Shutting down instance %s on node %s", 
 971                   self.instance.name, self.cfg.GetNodeName(source_node_uuid)) 
 972   
 973      result = self.rpc.call_instance_shutdown(source_node_uuid, self.instance, 
 974                                               self.shutdown_timeout, 
 975                                               self.lu.op.reason) 
 976      msg = result.fail_msg 
 977      if msg: 
 978        if self.ignore_consistency or primary_node.offline: 
 979          self.lu.LogWarning("Could not shutdown instance %s on node %s," 
 980                             " proceeding anyway; please make sure node" 
 981                             " %s is down; error details: %s", 
 982                             self.instance.name, 
 983                             self.cfg.GetNodeName(source_node_uuid), 
 984                             self.cfg.GetNodeName(source_node_uuid), msg) 
 985        else: 
 986          raise errors.OpExecError("Could not shutdown instance %s on" 
 987                                   " node %s: %s" % 
 988                                   (self.instance.name, 
 989                                    self.cfg.GetNodeName(source_node_uuid), msg)) 
 990   
 991      disk_template = self.cfg.GetInstanceDiskTemplate(self.instance.uuid) 
 992      if disk_template in constants.DTS_EXT_MIRROR: 
 993        self._CloseInstanceDisks(source_node_uuid) 
 994   
 995      self.feedback_fn("* deactivating the instance's disks on source node") 
 996      if not ShutdownInstanceDisks(self.lu, self.instance, ignore_primary=True): 
 997        raise errors.OpExecError("Can't shut down the instance's disks") 
 998   
 999      self.cfg.SetInstancePrimaryNode(self.instance.uuid, self.target_node_uuid) 
1000      self.instance = self.cfg.GetInstanceInfo(self.instance_uuid) 
1001   
1002      # Only start the instance if it's marked as up 
1003      if self.instance.admin_state == constants.ADMINST_UP: 
1004        self.feedback_fn("* activating the instance's disks on target node %s" % 
1005                         self.cfg.GetNodeName(self.target_node_uuid)) 
1006        logging.info("Starting instance %s on node %s", self.instance.name, 
1007                     self.cfg.GetNodeName(self.target_node_uuid)) 
1008   
1009        disks_ok, _, _ = AssembleInstanceDisks(self.lu, self.instance, 
1010                                               ignore_secondaries=True) 
1011        if not disks_ok: 
1012          ShutdownInstanceDisks(self.lu, self.instance) 
1013          raise errors.OpExecError("Can't activate the instance's disks") 
1014   
1015        self.feedback_fn("* starting the instance on the target node %s" % 
1016                         self.cfg.GetNodeName(self.target_node_uuid)) 
1017        result = self.rpc.call_instance_start(self.target_node_uuid, 
1018                                              (self.instance, None, None), False, 
1019                                              self.lu.op.reason) 
1020        msg = result.fail_msg 
1021        if msg: 
1022          ShutdownInstanceDisks(self.lu, self.instance) 
1023          raise errors.OpExecError("Could not start instance %s on node %s: %s" % 
1024                                   (self.instance.name, 
1025                                    self.cfg.GetNodeName(self.target_node_uuid), 
1026                                    msg)) 
1027   
1028 -  def Exec(self, feedback_fn): 
1029      """Perform the migration. 
1030   
1031      """ 
1032      self.feedback_fn = feedback_fn 
1033      self.source_node_uuid = self.instance.primary_node 
1034   
1035      # FIXME: if we implement migrate-to-any in DRBD, this needs fixing 
1036      disks = self.cfg.GetInstanceDisks(self.instance.uuid) 
1037   
1038      # TODO allow mixed disks 
1039      if utils.AllDiskOfType(disks, constants.DTS_INT_MIRROR): 
1040        secondary_nodes = self.cfg.GetInstanceSecondaryNodes(self.instance.uuid) 
1041        self.target_node_uuid = secondary_nodes[0] 
1042        # Otherwise self.target_node has been populated either 
1043        # directly, or through an iallocator. 
1044   
1045      self.all_node_uuids = [self.source_node_uuid, self.target_node_uuid] 
1046      self.nodes_ip = dict((uuid, node.secondary_ip) for (uuid, node) 
1047                           in self.cfg.GetMultiNodeInfo(self.all_node_uuids)) 
1048   
1049      if self.failover: 
1050        feedback_fn("Failover instance %s" % self.instance.name) 
1051        self._ExecFailover() 
1052      else: 
1053        feedback_fn("Migrating instance %s" % self.instance.name) 
1054   
1055        if self.cleanup: 
1056          return self._ExecCleanup() 
1057        else: 
1058          return self._ExecMigration() 
1059