Synnefo » snf-ganeti » ganeti-local

#

#

# 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.

"""Logical units dealing with storage of instances."""

import itertools

import logging

import os

import time

from ganeti import compat

from ganeti import constants

from ganeti import errors

from ganeti import ht

from ganeti import locking

from ganeti.masterd import iallocator

from ganeti import objects

from ganeti import utils

from ganeti import opcodes

from ganeti import rpc

from ganeti.cmdlib.base import LogicalUnit, NoHooksLU, Tasklet

from ganeti.cmdlib.common import INSTANCE_DOWN, INSTANCE_NOT_RUNNING, \

  AnnotateDiskParams, CheckIAllocatorOrNode, ExpandNodeName, \

  CheckNodeOnline, CheckInstanceNodeGroups, CheckInstanceState, \

  IsExclusiveStorageEnabledNode, FindFaultyInstanceDisks

from ganeti.cmdlib.instance_utils import GetInstanceInfoText, \

  CopyLockList, ReleaseLocks, CheckNodeVmCapable, \

  BuildInstanceHookEnvByObject, CheckNodeNotDrained, CheckTargetNodeIPolicy

import ganeti.masterd.instance

_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 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 _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 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 _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 _UndoCreateDisks(lu, disks_created):

  """Undo the work performed by L{CreateDisks}.

  This function is called in case of an error to undo the work of

  L{CreateDisks}.

  @type lu: L{LogicalUnit}

  @param lu: the logical unit on whose behalf we execute

  @param disks_created: the result returned by L{CreateDisks}

  """

  for (node, disk) in disks_created:

    lu.cfg.SetDiskID(disk, node)

    result = lu.rpc.call_blockdev_remove(node, disk)

    if result.fail_msg:

      logging.warning("Failed to remove newly-created disk %s on node %s:"

                      " %s", disk, node, result.fail_msg)

def CreateDisks(lu, instance, to_skip=None, target_node=None, disks=None):

  """Create all disks for an instance.

  This abstracts away some work from AddInstance.

  @type lu: L{LogicalUnit}

  @param lu: the logical unit on whose behalf we execute

  @type instance: L{objects.Instance}

  @param instance: the instance whose disks we should create

  @type to_skip: list

  @param to_skip: list of indices to skip

  @type target_node: string

  @param target_node: if passed, overrides the target node for creation

  @type disks: list of {objects.Disk}

  @param disks: the disks to create; if not specified, all the disks of the

      instance are created

  @return: information about the created disks, to be used to call

      L{_UndoCreateDisks}

  @raise errors.OpPrereqError: in case of error

  """

  info = GetInstanceInfoText(instance)

  if target_node is None:

    pnode = instance.primary_node

    all_nodes = instance.all_nodes

  else:

    pnode = target_node

    all_nodes = [pnode]

  if disks is None:

    disks = instance.disks

  if instance.disk_template in constants.DTS_FILEBASED:

    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])

    result = lu.rpc.call_file_storage_dir_create(pnode, file_storage_dir)

    result.Raise("Failed to create directory '%s' on"

                 " node %s" % (file_storage_dir, pnode))

  disks_created = []

  for idx, device in enumerate(disks):

    if to_skip and idx in to_skip:

      continue

    logging.info("Creating disk %s for instance '%s'", idx, instance.name)

    for node in all_nodes:

      f_create = node == pnode

      try:

        _CreateBlockDev(lu, node, instance, device, f_create, info, f_create)

        disks_created.append((node, device))

      except errors.DeviceCreationError, e:

        logging.warning("Creating disk %s for instance '%s' failed",

                        idx, instance.name)

        disks_created.extend(e.created_devices)

        _UndoCreateDisks(lu, disks_created)

        raise errors.OpExecError(e.message)

  return disks_created

def ComputeDiskSizePerVG(disk_template, disks):

  """Compute disk size requirements in the volume group

  """

  def _compute(disks, payload):

    """Universal algorithm.

    """

    vgs = {}

    for disk in disks:

      vgs[disk[constants.IDISK_VG]] = \

        vgs.get(constants.IDISK_VG, 0) + disk[constants.IDISK_SIZE] + payload

    return vgs

  # Required free disk space as a function of disk and swap space

  req_size_dict = {

    constants.DT_DISKLESS: {},

    constants.DT_PLAIN: _compute(disks, 0),

    # 128 MB are added for drbd metadata for each disk

    constants.DT_DRBD8: _compute(disks, constants.DRBD_META_SIZE),

    constants.DT_FILE: {},

    constants.DT_SHARED_FILE: {},

    }

  if disk_template not in req_size_dict:

    raise errors.ProgrammerError("Disk template '%s' size requirement"

                                 " is unknown" % disk_template)

  return req_size_dict[disk_template]

def ComputeDisks(op, default_vg):

  """Computes the instance disks.

  @param op: The instance opcode

  @param default_vg: The default_vg to assume

  @return: The computed disks

  """

  disks = []

  for disk in op.disks:

    mode = disk.get(constants.IDISK_MODE, constants.DISK_RDWR)

    if mode not in constants.DISK_ACCESS_SET:

      raise errors.OpPrereqError("Invalid disk access mode '%s'" %

                                 mode, errors.ECODE_INVAL)

    size = disk.get(constants.IDISK_SIZE, None)

    if size is None:

      raise errors.OpPrereqError("Missing disk size", errors.ECODE_INVAL)

    try:

      size = int(size)

    except (TypeError, ValueError):

      raise errors.OpPrereqError("Invalid disk size '%s'" % size,

                                 errors.ECODE_INVAL)

    ext_provider = disk.get(constants.IDISK_PROVIDER, None)

    if ext_provider and op.disk_template != constants.DT_EXT:

      raise errors.OpPrereqError("The '%s' option is only valid for the %s"

                                 " disk template, not %s" %

                                 (constants.IDISK_PROVIDER, constants.DT_EXT,

                                  op.disk_template), errors.ECODE_INVAL)

    data_vg = disk.get(constants.IDISK_VG, default_vg)

    name = disk.get(constants.IDISK_NAME, None)

    if name is not None and name.lower() == constants.VALUE_NONE:

      name = None

    new_disk = {

      constants.IDISK_SIZE: size,

      constants.IDISK_MODE: mode,

      constants.IDISK_VG: data_vg,

      constants.IDISK_NAME: name,

      }

    for key in [

      constants.IDISK_METAVG,

      constants.IDISK_ADOPT,

      constants.IDISK_SPINDLES,

      ]:

      if key in disk:

        new_disk[key] = disk[key]

    # For extstorage, demand the `provider' option and add any

    # additional parameters (ext-params) to the dict

    if op.disk_template == constants.DT_EXT:

      if ext_provider:

        new_disk[constants.IDISK_PROVIDER] = ext_provider

        for key in disk:

          if key not in constants.IDISK_PARAMS:

            new_disk[key] = disk[key]

      else:

        raise errors.OpPrereqError("Missing provider for template '%s'" %

                                   constants.DT_EXT, errors.ECODE_INVAL)

    disks.append(new_disk)

  return disks

def CheckRADOSFreeSpace():

  """Compute disk size requirements inside the RADOS cluster.

  """

  # For the RADOS cluster we assume there is always enough space.

  pass

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

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,

                              spindles=disk.get(constants.IDISK_SPINDLES))

      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 CheckSpindlesExclusiveStorage(diskdict, es_flag, required):

  """Check the presence of the spindle options with exclusive_storage.

  @type diskdict: dict

  @param diskdict: disk parameters

  @type es_flag: bool

  @param es_flag: the effective value of the exlusive_storage flag

  @type required: bool

  @param required: whether spindles are required or just optional

  @raise errors.OpPrereqError when spindles are given and they should not

  """

  if (not es_flag and constants.IDISK_SPINDLES in diskdict and

      diskdict[constants.IDISK_SPINDLES] is not None):

    raise errors.OpPrereqError("Spindles in instance disks cannot be specified"

                               " when exclusive storage is not active",

                               errors.ECODE_INVAL)

  if (es_flag and required and (constants.IDISK_SPINDLES not in diskdict or

                                diskdict[constants.IDISK_SPINDLES] is None)):

    raise errors.OpPrereqError("You must specify spindles in instance disks"

                               " when exclusive storage is active",

                               errors.ECODE_INVAL)

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,

    constants.IDISK_SPINDLES,

    ])

  # 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]

    disks = [{

      constants.IDISK_SIZE: d.size,

      constants.IDISK_MODE: d.mode,

      constants.IDISK_SPINDLES: d.spindles,

      } for d in self.instance.disks]

    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=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)

    if self.op.nodes:

      nodes = self.op.nodes

    else:

      nodes = instance.all_nodes

    excl_stor = compat.any(

      rpc.GetExclusiveStorageForNodeNames(self.cfg, nodes).values()

      )

    for new_params in self.disks.values():

      CheckSpindlesExclusiveStorage(new_params, excl_stor, False)

  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),

                    spindles=changes.get(constants.IDISK_SPINDLES, 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))

    new_disks = CreateDisks(self, instance, to_skip=to_skip)

    # TODO: Release node locks before wiping, or explain why it's not possible

    if self.cfg.GetClusterInfo().prealloc_wipe_disks:

      wipedisks = [(idx, disk, 0)

                   for (idx, disk) in enumerate(instance.disks)

                   if idx not in to_skip]

      WipeOrCleanupDisks(self, instance, disks=wipedisks, cleanup=new_disks)

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)

  # FIXME: This maps everything to storage type 'lvm-vg' to maintain

  # the current functionality. Refactor to make it more flexible.

  nodeinfo = lu.rpc.call_node_info(nodenames, [(constants.ST_LVM_VG, 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 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 _DiskSizeInBytesToMebibytes(lu, size):

  """Converts a disk size in bytes to mebibytes.

  Warns and rounds up if the size isn't an even multiple of 1 MiB.

  """

  (mib, remainder) = divmod(size, 1024 * 1024)

  if remainder != 0:

    lu.LogWarning("Disk size is not an even multiple of 1 MiB; rounding up"

                  " to not overwrite existing data (%s bytes will not be"

                  " wiped)", (1024 * 1024) - remainder)

    mib += 1

  return mib

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

  """

  avg_time = time_taken / float(written)

  return (total_size - written) * avg_time

def WipeDisks(lu, instance, disks=None):

  """Wipes instance disks.

  @type lu: L{LogicalUnit}

  @param lu: the logical unit on whose behalf we execute

  @type instance: L{objects.Instance}

  @param instance: the instance whose disks we should create

  @type disks: None or list of tuple of (number, L{objects.Disk}, number)

  @param disks: Disk details; tuple contains disk index, disk object and the

    start offset

  """

  node = instance.primary_node

  if disks is None:

    disks = [(idx, disk, 0)

             for (idx, disk) in enumerate(instance.disks)]

  for (_, device, _) in disks:

    lu.cfg.SetDiskID(device, node)

  logging.info("Pausing synchronization of disks of instance '%s'",

               instance.name)

  result = lu.rpc.call_blockdev_pause_resume_sync(node,

                                                  (map(compat.snd, disks),

                                                   instance),

                                                  True)

  result.Raise("Failed to pause disk synchronization on node '%s'" % node)

  for idx, success in enumerate(result.payload):

    if not success:

      logging.warn("Pausing synchronization of disk %s of instance '%s'"

                   " failed", idx, instance.name)

  try:

    for (idx, device, offset) in disks:

      # The wipe size is MIN_WIPE_CHUNK_PERCENT % of the instance disk but

      # MAX_WIPE_CHUNK at max. Truncating to integer to avoid rounding errors.

      wipe_chunk_size = \

        int(min(constants.MAX_WIPE_CHUNK,

                device.size / 100.0 * constants.MIN_WIPE_CHUNK_PERCENT))

      size = device.size

      last_output = 0

      start_time = time.time()

      if offset == 0:

        info_text = ""

      else:

        info_text = (" (from %s to %s)" %

                     (utils.FormatUnit(offset, "h"),

                      utils.FormatUnit(size, "h")))

      lu.LogInfo("* Wiping disk %s%s", idx, info_text)

      logging.info("Wiping disk %d for instance %s on node %s using"

                   " chunk size %s", idx, instance.name, node, wipe_chunk_size)

      while offset < size:

        wipe_size = min(wipe_chunk_size, size - offset)

        logging.debug("Wiping disk %d, offset %s, chunk %s",

                      idx, offset, wipe_size)

        result = lu.rpc.call_blockdev_wipe(node, (device, instance), offset,

                                           wipe_size)

        result.Raise("Could not wipe disk %d at offset %d for size %d" %

                     (idx, offset, wipe_size))

        now = time.time()

        offset += wipe_size

        if now - last_output >= 60:

          eta = _CalcEta(now - start_time, offset, size)

          lu.LogInfo(" - done: %.1f%% ETA: %s",

                     offset / float(size) * 100, utils.FormatSeconds(eta))

          last_output = now

  finally:

    logging.info("Resuming synchronization of disks for instance '%s'",

                 instance.name)

    result = lu.rpc.call_blockdev_pause_resume_sync(node,

                                                    (map(compat.snd, disks),

                                                     instance),

                                                    False)

    if result.fail_msg:

      lu.LogWarning("Failed to resume disk synchronization on node '%s': %s",

                    node, result.fail_msg)

    else:

      for idx, success in enumerate(result.payload):

        if not success:

          lu.LogWarning("Resuming synchronization of disk %s of instance '%s'"

                        " failed", idx, instance.name)

def WipeOrCleanupDisks(lu, instance, disks=None, cleanup=None):

  """Wrapper for L{WipeDisks} that handles errors.

  @type lu: L{LogicalUnit}

  @param lu: the logical unit on whose behalf we execute

  @type instance: L{objects.Instance}

  @param instance: the instance whose disks we should wipe

  @param disks: see L{WipeDisks}

  @param cleanup: the result returned by L{CreateDisks}, used for cleanup in

      case of error

  @raise errors.OpPrereqError: in case of failure

  """

  try:

    WipeDisks(lu, instance, disks=disks)

  except errors.OpExecError:

    logging.warning("Wiping disks for instance '%s' failed",

                    instance.name)

    _UndoCreateDisks(lu, cleanup)

    raise

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: expected a subset of %r,"

                                   " got %r" % (instance.disks, disks))

    return disks

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)