Synnefo » snf-ganeti

#

#

# 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 the cluster."""

import OpenSSL

import copy

import itertools

import logging

import operator

import os

import re

import time

from ganeti import compat

from ganeti import constants

from ganeti import errors

from ganeti import hypervisor

from ganeti import locking

from ganeti import masterd

from ganeti import netutils

from ganeti import objects

from ganeti import opcodes

from ganeti import pathutils

from ganeti import query

from ganeti import rpc

from ganeti import runtime

from ganeti import ssh

from ganeti import uidpool

from ganeti import utils

from ganeti import vcluster

from ganeti.cmdlib.base import NoHooksLU, QueryBase, LogicalUnit, \

  ResultWithJobs

from ganeti.cmdlib.common import ShareAll, RunPostHook, \

  ComputeAncillaryFiles, RedistributeAncillaryFiles, UploadHelper, \

  GetWantedInstances, MergeAndVerifyHvState, MergeAndVerifyDiskState, \

  GetUpdatedIPolicy, ComputeNewInstanceViolations, GetUpdatedParams, \

  CheckOSParams, CheckHVParams, AdjustCandidatePool, CheckNodePVs, \

  ComputeIPolicyInstanceViolation, AnnotateDiskParams, SupportsOob, \

  CheckIpolicyVsDiskTemplates, CheckDiskAccessModeValidity, \

  CheckDiskAccessModeConsistency

import ganeti.masterd.instance

class LUClusterActivateMasterIp(NoHooksLU):

  """Activate the master IP on the master node.

  """

  def Exec(self, feedback_fn):

    """Activate the master IP.

    """

    master_params = self.cfg.GetMasterNetworkParameters()

    ems = self.cfg.GetUseExternalMipScript()

    result = self.rpc.call_node_activate_master_ip(master_params.uuid,

                                                   master_params, ems)

    result.Raise("Could not activate the master IP")

class LUClusterDeactivateMasterIp(NoHooksLU):

  """Deactivate the master IP on the master node.

  """

  def Exec(self, feedback_fn):

    """Deactivate the master IP.

    """

    master_params = self.cfg.GetMasterNetworkParameters()

    ems = self.cfg.GetUseExternalMipScript()

    result = self.rpc.call_node_deactivate_master_ip(master_params.uuid,

                                                     master_params, ems)

    result.Raise("Could not deactivate the master IP")

class LUClusterConfigQuery(NoHooksLU):

  """Return configuration values.

  """

  REQ_BGL = False

  def CheckArguments(self):

    self.cq = ClusterQuery(None, self.op.output_fields, False)

  def ExpandNames(self):

    self.cq.ExpandNames(self)

  def DeclareLocks(self, level):

    self.cq.DeclareLocks(self, level)

  def Exec(self, feedback_fn):

    result = self.cq.OldStyleQuery(self)

    assert len(result) == 1

    return result[0]

class LUClusterDestroy(LogicalUnit):

  """Logical unit for destroying the cluster.

  """

  HPATH = "cluster-destroy"

  HTYPE = constants.HTYPE_CLUSTER

  def BuildHooksEnv(self):

    """Build hooks env.

    """

    return {

      "OP_TARGET": self.cfg.GetClusterName(),

      }

  def BuildHooksNodes(self):

    """Build hooks nodes.

    """

    return ([], [])

  def CheckPrereq(self):

    """Check prerequisites.

    This checks whether the cluster is empty.

    Any errors are signaled by raising errors.OpPrereqError.

    """

    master = self.cfg.GetMasterNode()

    nodelist = self.cfg.GetNodeList()

    if len(nodelist) != 1 or nodelist[0] != master:

      raise errors.OpPrereqError("There are still %d node(s) in"

                                 " this cluster." % (len(nodelist) - 1),

                                 errors.ECODE_INVAL)

    instancelist = self.cfg.GetInstanceList()

    if instancelist:

      raise errors.OpPrereqError("There are still %d instance(s) in"

                                 " this cluster." % len(instancelist),

                                 errors.ECODE_INVAL)

  def Exec(self, feedback_fn):

    """Destroys the cluster.

    """

    master_params = self.cfg.GetMasterNetworkParameters()

    # Run post hooks on master node before it's removed

    RunPostHook(self, self.cfg.GetNodeName(master_params.uuid))

    ems = self.cfg.GetUseExternalMipScript()

    result = self.rpc.call_node_deactivate_master_ip(master_params.uuid,

                                                     master_params, ems)

    result.Warn("Error disabling the master IP address", self.LogWarning)

    return master_params.uuid

class LUClusterPostInit(LogicalUnit):

  """Logical unit for running hooks after cluster initialization.

  """

  HPATH = "cluster-init"

  HTYPE = constants.HTYPE_CLUSTER

  def BuildHooksEnv(self):

    """Build hooks env.

    """

    return {

      "OP_TARGET": self.cfg.GetClusterName(),

      }

  def BuildHooksNodes(self):

    """Build hooks nodes.

    """

    return ([], [self.cfg.GetMasterNode()])

  def Exec(self, feedback_fn):

    """Nothing to do.

    """

    return True

class ClusterQuery(QueryBase):

  FIELDS = query.CLUSTER_FIELDS

  #: Do not sort (there is only one item)

  SORT_FIELD = None

  def ExpandNames(self, lu):

    lu.needed_locks = {}

    # The following variables interact with _QueryBase._GetNames

    self.wanted = locking.ALL_SET

    self.do_locking = self.use_locking

    if self.do_locking:

      raise errors.OpPrereqError("Can not use locking for cluster queries",

                                 errors.ECODE_INVAL)

  def DeclareLocks(self, lu, level):

    pass

  def _GetQueryData(self, lu):

    """Computes the list of nodes and their attributes.

    """

    # Locking is not used

    assert not (compat.any(lu.glm.is_owned(level)

                           for level in locking.LEVELS

                           if level != locking.LEVEL_CLUSTER) or

                self.do_locking or self.use_locking)

    if query.CQ_CONFIG in self.requested_data:

      cluster = lu.cfg.GetClusterInfo()

      nodes = lu.cfg.GetAllNodesInfo()

    else:

      cluster = NotImplemented

      nodes = NotImplemented

    if query.CQ_QUEUE_DRAINED in self.requested_data:

      drain_flag = os.path.exists(pathutils.JOB_QUEUE_DRAIN_FILE)

    else:

      drain_flag = NotImplemented

    if query.CQ_WATCHER_PAUSE in self.requested_data:

      master_node_uuid = lu.cfg.GetMasterNode()

      result = lu.rpc.call_get_watcher_pause(master_node_uuid)

      result.Raise("Can't retrieve watcher pause from master node '%s'" %

                   lu.cfg.GetMasterNodeName())

      watcher_pause = result.payload

    else:

      watcher_pause = NotImplemented

    return query.ClusterQueryData(cluster, nodes, drain_flag, watcher_pause)

class LUClusterQuery(NoHooksLU):

  """Query cluster configuration.

  """

  REQ_BGL = False

  def ExpandNames(self):

    self.needed_locks = {}

  def Exec(self, feedback_fn):

    """Return cluster config.

    """

    cluster = self.cfg.GetClusterInfo()

    os_hvp = {}

    # Filter just for enabled hypervisors

    for os_name, hv_dict in cluster.os_hvp.items():

      os_hvp[os_name] = {}

      for hv_name, hv_params in hv_dict.items():

        if hv_name in cluster.enabled_hypervisors:

          os_hvp[os_name][hv_name] = hv_params

    # Convert ip_family to ip_version

    primary_ip_version = constants.IP4_VERSION

    if cluster.primary_ip_family == netutils.IP6Address.family:

      primary_ip_version = constants.IP6_VERSION

    result = {

      "software_version": constants.RELEASE_VERSION,

      "protocol_version": constants.PROTOCOL_VERSION,

      "config_version": constants.CONFIG_VERSION,

      "os_api_version": max(constants.OS_API_VERSIONS),

      "export_version": constants.EXPORT_VERSION,

      "vcs_version": constants.VCS_VERSION,

      "architecture": runtime.GetArchInfo(),

      "name": cluster.cluster_name,

      "master": self.cfg.GetMasterNodeName(),

      "default_hypervisor": cluster.primary_hypervisor,

      "enabled_hypervisors": cluster.enabled_hypervisors,

      "hvparams": dict([(hypervisor_name, cluster.hvparams[hypervisor_name])

                        for hypervisor_name in cluster.enabled_hypervisors]),

      "os_hvp": os_hvp,

      "beparams": cluster.beparams,

      "osparams": cluster.osparams,

      "ipolicy": cluster.ipolicy,

      "nicparams": cluster.nicparams,

      "ndparams": cluster.ndparams,

      "diskparams": cluster.diskparams,

      "candidate_pool_size": cluster.candidate_pool_size,

      "master_netdev": cluster.master_netdev,

      "master_netmask": cluster.master_netmask,

      "use_external_mip_script": cluster.use_external_mip_script,

      "volume_group_name": cluster.volume_group_name,

      "drbd_usermode_helper": cluster.drbd_usermode_helper,

      "file_storage_dir": cluster.file_storage_dir,

      "shared_file_storage_dir": cluster.shared_file_storage_dir,

      "maintain_node_health": cluster.maintain_node_health,

      "ctime": cluster.ctime,

      "mtime": cluster.mtime,

      "uuid": cluster.uuid,

      "tags": list(cluster.GetTags()),

      "uid_pool": cluster.uid_pool,

      "default_iallocator": cluster.default_iallocator,

      "reserved_lvs": cluster.reserved_lvs,

      "primary_ip_version": primary_ip_version,

      "prealloc_wipe_disks": cluster.prealloc_wipe_disks,

      "hidden_os": cluster.hidden_os,

      "blacklisted_os": cluster.blacklisted_os,

      "enabled_disk_templates": cluster.enabled_disk_templates,

      }

    return result

class LUClusterRedistConf(NoHooksLU):

  """Force the redistribution of cluster configuration.

  This is a very simple LU.

  """

  REQ_BGL = False

  def ExpandNames(self):

    self.needed_locks = {

      locking.LEVEL_NODE: locking.ALL_SET,

      locking.LEVEL_NODE_ALLOC: locking.ALL_SET,

    }

    self.share_locks = ShareAll()

  def Exec(self, feedback_fn):

    """Redistribute the configuration.

    """

    self.cfg.Update(self.cfg.GetClusterInfo(), feedback_fn)

    RedistributeAncillaryFiles(self)

class LUClusterRename(LogicalUnit):

  """Rename the cluster.

  """

  HPATH = "cluster-rename"

  HTYPE = constants.HTYPE_CLUSTER

  def BuildHooksEnv(self):

    """Build hooks env.

    """

    return {

      "OP_TARGET": self.cfg.GetClusterName(),

      "NEW_NAME": self.op.name,

      }

  def BuildHooksNodes(self):

    """Build hooks nodes.

    """

    return ([self.cfg.GetMasterNode()], self.cfg.GetNodeList())

  def CheckPrereq(self):

    """Verify that the passed name is a valid one.

    """

    hostname = netutils.GetHostname(name=self.op.name,

                                    family=self.cfg.GetPrimaryIPFamily())

    new_name = hostname.name

    self.ip = new_ip = hostname.ip

    old_name = self.cfg.GetClusterName()

    old_ip = self.cfg.GetMasterIP()

    if new_name == old_name and new_ip == old_ip:

      raise errors.OpPrereqError("Neither the name nor the IP address of the"

                                 " cluster has changed",

                                 errors.ECODE_INVAL)

    if new_ip != old_ip:

      if netutils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):

        raise errors.OpPrereqError("The given cluster IP address (%s) is"

                                   " reachable on the network" %

                                   new_ip, errors.ECODE_NOTUNIQUE)

    self.op.name = new_name

  def Exec(self, feedback_fn):

    """Rename the cluster.

    """

    clustername = self.op.name

    new_ip = self.ip

    # shutdown the master IP

    master_params = self.cfg.GetMasterNetworkParameters()

    ems = self.cfg.GetUseExternalMipScript()

    result = self.rpc.call_node_deactivate_master_ip(master_params.uuid,

                                                     master_params, ems)

    result.Raise("Could not disable the master role")

    try:

      cluster = self.cfg.GetClusterInfo()

      cluster.cluster_name = clustername

      cluster.master_ip = new_ip

      self.cfg.Update(cluster, feedback_fn)

      # update the known hosts file

      ssh.WriteKnownHostsFile(self.cfg, pathutils.SSH_KNOWN_HOSTS_FILE)

      node_list = self.cfg.GetOnlineNodeList()

      try:

        node_list.remove(master_params.uuid)

      except ValueError:

        pass

      UploadHelper(self, node_list, pathutils.SSH_KNOWN_HOSTS_FILE)

    finally:

      master_params.ip = new_ip

      result = self.rpc.call_node_activate_master_ip(master_params.uuid,

                                                     master_params, ems)

      result.Warn("Could not re-enable the master role on the master,"

                  " please restart manually", self.LogWarning)

    return clustername

class LUClusterRepairDiskSizes(NoHooksLU):

  """Verifies the cluster disks sizes.

  """

  REQ_BGL = False

  def ExpandNames(self):

    if self.op.instances:

      (_, self.wanted_names) = GetWantedInstances(self, self.op.instances)

      # Not getting the node allocation lock as only a specific set of

      # instances (and their nodes) is going to be acquired

      self.needed_locks = {

        locking.LEVEL_NODE_RES: [],

        locking.LEVEL_INSTANCE: self.wanted_names,

        }

      self.recalculate_locks[locking.LEVEL_NODE_RES] = constants.LOCKS_REPLACE

    else:

      self.wanted_names = None

      self.needed_locks = {

        locking.LEVEL_NODE_RES: locking.ALL_SET,

        locking.LEVEL_INSTANCE: locking.ALL_SET,

        # This opcode is acquires the node locks for all instances

        locking.LEVEL_NODE_ALLOC: locking.ALL_SET,

        }

    self.share_locks = {

      locking.LEVEL_NODE_RES: 1,

      locking.LEVEL_INSTANCE: 0,

      locking.LEVEL_NODE_ALLOC: 1,

      }

  def DeclareLocks(self, level):

    if level == locking.LEVEL_NODE_RES and self.wanted_names is not None:

      self._LockInstancesNodes(primary_only=True, level=level)

  def CheckPrereq(self):

    """Check prerequisites.

    This only checks the optional instance list against the existing names.

    """

    if self.wanted_names is None:

      self.wanted_names = self.owned_locks(locking.LEVEL_INSTANCE)

    self.wanted_instances = \

        map(compat.snd, self.cfg.GetMultiInstanceInfoByName(self.wanted_names))

  def _EnsureChildSizes(self, disk):

    """Ensure children of the disk have the needed disk size.

    This is valid mainly for DRBD8 and fixes an issue where the

    children have smaller disk size.

    @param disk: an L{ganeti.objects.Disk} object

    """

    if disk.dev_type == constants.DT_DRBD8:

      assert disk.children, "Empty children for DRBD8?"

      fchild = disk.children[0]

      mismatch = fchild.size < disk.size

      if mismatch:

        self.LogInfo("Child disk has size %d, parent %d, fixing",

                     fchild.size, disk.size)

        fchild.size = disk.size

      # and we recurse on this child only, not on the metadev

      return self._EnsureChildSizes(fchild) or mismatch

    else:

      return False

  def Exec(self, feedback_fn):

    """Verify the size of cluster disks.

    """

    # TODO: check child disks too

    # TODO: check differences in size between primary/secondary nodes

    per_node_disks = {}

    for instance in self.wanted_instances:

      pnode = instance.primary_node

      if pnode not in per_node_disks:

        per_node_disks[pnode] = []

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

        per_node_disks[pnode].append((instance, idx, disk))

    assert not (frozenset(per_node_disks.keys()) -

                self.owned_locks(locking.LEVEL_NODE_RES)), \

      "Not owning correct locks"

    assert not self.owned_locks(locking.LEVEL_NODE)

    es_flags = rpc.GetExclusiveStorageForNodes(self.cfg,

                                               per_node_disks.keys())

    changed = []

    for node_uuid, dskl in per_node_disks.items():

      if not dskl:

        # no disks on the node

        continue

      newl = [([v[2].Copy()], v[0]) for v in dskl]

      node_name = self.cfg.GetNodeName(node_uuid)

      result = self.rpc.call_blockdev_getdimensions(node_uuid, newl)

      if result.fail_msg:

        self.LogWarning("Failure in blockdev_getdimensions call to node"

                        " %s, ignoring", node_name)

        continue

      if len(result.payload) != len(dskl):

        logging.warning("Invalid result from node %s: len(dksl)=%d,"

                        " result.payload=%s", node_name, len(dskl),

                        result.payload)

        self.LogWarning("Invalid result from node %s, ignoring node results",

                        node_name)

        continue

      for ((instance, idx, disk), dimensions) in zip(dskl, result.payload):

        if dimensions is None:

          self.LogWarning("Disk %d of instance %s did not return size"

                          " information, ignoring", idx, instance.name)

          continue

        if not isinstance(dimensions, (tuple, list)):

          self.LogWarning("Disk %d of instance %s did not return valid"

                          " dimension information, ignoring", idx,

                          instance.name)

          continue

        (size, spindles) = dimensions

        if not isinstance(size, (int, long)):

          self.LogWarning("Disk %d of instance %s did not return valid"

                          " size information, ignoring", idx, instance.name)

          continue

        size = size >> 20

        if size != disk.size:

          self.LogInfo("Disk %d of instance %s has mismatched size,"

                       " correcting: recorded %d, actual %d", idx,

                       instance.name, disk.size, size)

          disk.size = size

          self.cfg.Update(instance, feedback_fn)

          changed.append((instance.name, idx, "size", size))

        if es_flags[node_uuid]:

          if spindles is None:

            self.LogWarning("Disk %d of instance %s did not return valid"

                            " spindles information, ignoring", idx,

                            instance.name)

          elif disk.spindles is None or disk.spindles != spindles:

            self.LogInfo("Disk %d of instance %s has mismatched spindles,"

                         " correcting: recorded %s, actual %s",

                         idx, instance.name, disk.spindles, spindles)

            disk.spindles = spindles

            self.cfg.Update(instance, feedback_fn)

            changed.append((instance.name, idx, "spindles", disk.spindles))

        if self._EnsureChildSizes(disk):

          self.cfg.Update(instance, feedback_fn)

          changed.append((instance.name, idx, "size", disk.size))

    return changed

def _ValidateNetmask(cfg, netmask):

  """Checks if a netmask is valid.

  @type cfg: L{config.ConfigWriter}

  @param cfg: The cluster configuration

  @type netmask: int

  @param netmask: the netmask to be verified

  @raise errors.OpPrereqError: if the validation fails

  """

  ip_family = cfg.GetPrimaryIPFamily()

  try:

    ipcls = netutils.IPAddress.GetClassFromIpFamily(ip_family)

  except errors.ProgrammerError:

    raise errors.OpPrereqError("Invalid primary ip family: %s." %

                               ip_family, errors.ECODE_INVAL)

  if not ipcls.ValidateNetmask(netmask):

    raise errors.OpPrereqError("CIDR netmask (%s) not valid" %

                               (netmask), errors.ECODE_INVAL)

def CheckFileBasedStoragePathVsEnabledDiskTemplates(

    logging_warn_fn, file_storage_dir, enabled_disk_templates,

    file_disk_template):

  """Checks whether the given file-based storage directory is acceptable.

  Note: This function is public, because it is also used in bootstrap.py.

  @type logging_warn_fn: function

  @param logging_warn_fn: function which accepts a string and logs it

  @type file_storage_dir: string

  @param file_storage_dir: the directory to be used for file-based instances

  @type enabled_disk_templates: list of string

  @param enabled_disk_templates: the list of enabled disk templates

  @type file_disk_template: string

  @param file_disk_template: the file-based disk template for which the

      path should be checked

  """

  assert (file_disk_template in

          utils.storage.GetDiskTemplatesOfStorageType(constants.ST_FILE))

  file_storage_enabled = file_disk_template in enabled_disk_templates

  if file_storage_dir is not None:

    if file_storage_dir == "":

      if file_storage_enabled:

        raise errors.OpPrereqError(

            "Unsetting the '%s' storage directory while having '%s' storage"

            " enabled is not permitted." %

            (file_disk_template, file_disk_template))

    else:

      if not file_storage_enabled:

        logging_warn_fn(

            "Specified a %s storage directory, although %s storage is not"

            " enabled." % (file_disk_template, file_disk_template))

  else:

    raise errors.ProgrammerError("Received %s storage dir with value"

                                 " 'None'." % file_disk_template)

def CheckFileStoragePathVsEnabledDiskTemplates(

    logging_warn_fn, file_storage_dir, enabled_disk_templates):

  """Checks whether the given file storage directory is acceptable.

  @see: C{CheckFileBasedStoragePathVsEnabledDiskTemplates}

  """

  CheckFileBasedStoragePathVsEnabledDiskTemplates(

      logging_warn_fn, file_storage_dir, enabled_disk_templates,

      constants.DT_FILE)

def CheckSharedFileStoragePathVsEnabledDiskTemplates(

    logging_warn_fn, file_storage_dir, enabled_disk_templates):

  """Checks whether the given shared file storage directory is acceptable.

  @see: C{CheckFileBasedStoragePathVsEnabledDiskTemplates}

  """

  CheckFileBasedStoragePathVsEnabledDiskTemplates(

      logging_warn_fn, file_storage_dir, enabled_disk_templates,

      constants.DT_SHARED_FILE)

class LUClusterSetParams(LogicalUnit):

  """Change the parameters of the cluster.

  """

  HPATH = "cluster-modify"

  HTYPE = constants.HTYPE_CLUSTER

  REQ_BGL = False

  def CheckArguments(self):

    """Check parameters

    """

    if self.op.uid_pool:

      uidpool.CheckUidPool(self.op.uid_pool)

    if self.op.add_uids:

      uidpool.CheckUidPool(self.op.add_uids)

    if self.op.remove_uids:

      uidpool.CheckUidPool(self.op.remove_uids)

    if self.op.master_netmask is not None:

      _ValidateNetmask(self.cfg, self.op.master_netmask)

    if self.op.diskparams:

      for dt_params in self.op.diskparams.values():

        utils.ForceDictType(dt_params, constants.DISK_DT_TYPES)

      try:

        utils.VerifyDictOptions(self.op.diskparams, constants.DISK_DT_DEFAULTS)

        CheckDiskAccessModeValidity(self.op.diskparams)

      except errors.OpPrereqError, err:

        raise errors.OpPrereqError("While verify diskparams options: %s" % err,

                                   errors.ECODE_INVAL)

  def ExpandNames(self):

    # FIXME: in the future maybe other cluster params won't require checking on

    # all nodes to be modified.

    # FIXME: This opcode changes cluster-wide settings. Is acquiring all

    # resource locks the right thing, shouldn't it be the BGL instead?

    self.needed_locks = {

      locking.LEVEL_NODE: locking.ALL_SET,

      locking.LEVEL_INSTANCE: locking.ALL_SET,

      locking.LEVEL_NODEGROUP: locking.ALL_SET,

      locking.LEVEL_NODE_ALLOC: locking.ALL_SET,

    }

    self.share_locks = ShareAll()

  def BuildHooksEnv(self):

    """Build hooks env.

    """

    return {

      "OP_TARGET": self.cfg.GetClusterName(),

      "NEW_VG_NAME": self.op.vg_name,

      }

  def BuildHooksNodes(self):

    """Build hooks nodes.

    """

    mn = self.cfg.GetMasterNode()

    return ([mn], [mn])

  def _CheckVgName(self, node_uuids, enabled_disk_templates,

                   new_enabled_disk_templates):

    """Check the consistency of the vg name on all nodes and in case it gets

       unset whether there are instances still using it.

    """

    lvm_is_enabled = utils.IsLvmEnabled(enabled_disk_templates)

    lvm_gets_enabled = utils.LvmGetsEnabled(enabled_disk_templates,

                                            new_enabled_disk_templates)

    current_vg_name = self.cfg.GetVGName()

    if self.op.vg_name == '':

      if lvm_is_enabled:

        raise errors.OpPrereqError("Cannot unset volume group if lvm-based"

                                   " disk templates are or get enabled.")

    if self.op.vg_name is None:

      if current_vg_name is None and lvm_is_enabled:

        raise errors.OpPrereqError("Please specify a volume group when"

                                   " enabling lvm-based disk-templates.")

    if self.op.vg_name is not None and not self.op.vg_name:

      if self.cfg.HasAnyDiskOfType(constants.DT_PLAIN):

        raise errors.OpPrereqError("Cannot disable lvm storage while lvm-based"

                                   " instances exist", errors.ECODE_INVAL)

    if (self.op.vg_name is not None and lvm_is_enabled) or \

        (self.cfg.GetVGName() is not None and lvm_gets_enabled):

      self._CheckVgNameOnNodes(node_uuids)

  def _CheckVgNameOnNodes(self, node_uuids):

    """Check the status of the volume group on each node.

    """

    vglist = self.rpc.call_vg_list(node_uuids)

    for node_uuid in node_uuids:

      msg = vglist[node_uuid].fail_msg

      if msg:

        # ignoring down node

        self.LogWarning("Error while gathering data on node %s"

                        " (ignoring node): %s",

                        self.cfg.GetNodeName(node_uuid), msg)

        continue

      vgstatus = utils.CheckVolumeGroupSize(vglist[node_uuid].payload,

                                            self.op.vg_name,

                                            constants.MIN_VG_SIZE)

      if vgstatus:

        raise errors.OpPrereqError("Error on node '%s': %s" %

                                   (self.cfg.GetNodeName(node_uuid), vgstatus),

                                   errors.ECODE_ENVIRON)

  @staticmethod

  def _GetEnabledDiskTemplatesInner(op_enabled_disk_templates,

                                    old_enabled_disk_templates):

    """Determines the enabled disk templates and the subset of disk templates

       that are newly enabled by this operation.

    """

    enabled_disk_templates = None

    new_enabled_disk_templates = []

    if op_enabled_disk_templates:

      enabled_disk_templates = op_enabled_disk_templates

      new_enabled_disk_templates = \

        list(set(enabled_disk_templates)

             - set(old_enabled_disk_templates))

    else:

      enabled_disk_templates = old_enabled_disk_templates

    return (enabled_disk_templates, new_enabled_disk_templates)

  def _GetEnabledDiskTemplates(self, cluster):

    """Determines the enabled disk templates and the subset of disk templates

       that are newly enabled by this operation.

    """

    return self._GetEnabledDiskTemplatesInner(self.op.enabled_disk_templates,

                                              cluster.enabled_disk_templates)

  def _CheckIpolicy(self, cluster, enabled_disk_templates):

    """Checks the ipolicy.

    @type cluster: C{objects.Cluster}

    @param cluster: the cluster's configuration

    @type enabled_disk_templates: list of string

    @param enabled_disk_templates: list of (possibly newly) enabled disk

      templates

    """

    # FIXME: write unit tests for this

    if self.op.ipolicy:

      self.new_ipolicy = GetUpdatedIPolicy(cluster.ipolicy, self.op.ipolicy,

                                           group_policy=False)

      CheckIpolicyVsDiskTemplates(self.new_ipolicy,

                                  enabled_disk_templates)

      all_instances = self.cfg.GetAllInstancesInfo().values()

      violations = set()

      for group in self.cfg.GetAllNodeGroupsInfo().values():

        instances = frozenset([inst for inst in all_instances

                               if compat.any(nuuid in group.members

                                             for nuuid in inst.all_nodes)])

        new_ipolicy = objects.FillIPolicy(self.new_ipolicy, group.ipolicy)

        ipol = masterd.instance.CalculateGroupIPolicy(cluster, group)

        new = ComputeNewInstanceViolations(ipol, new_ipolicy, instances,

                                           self.cfg)

        if new:

          violations.update(new)

      if violations:

        self.LogWarning("After the ipolicy change the following instances"

                        " violate them: %s",

                        utils.CommaJoin(utils.NiceSort(violations)))

    else:

      CheckIpolicyVsDiskTemplates(cluster.ipolicy,

                                  enabled_disk_templates)

  def _CheckDrbdHelperOnNodes(self, drbd_helper, node_uuids):

    """Checks whether the set DRBD helper actually exists on the nodes.

    @type drbd_helper: string

    @param drbd_helper: path of the drbd usermode helper binary

    @type node_uuids: list of strings

    @param node_uuids: list of node UUIDs to check for the helper

    """

    # checks given drbd helper on all nodes

    helpers = self.rpc.call_drbd_helper(node_uuids)

    for (_, ninfo) in self.cfg.GetMultiNodeInfo(node_uuids):

      if ninfo.offline:

        self.LogInfo("Not checking drbd helper on offline node %s",

                     ninfo.name)

        continue

      msg = helpers[ninfo.uuid].fail_msg

      if msg:

        raise errors.OpPrereqError("Error checking drbd helper on node"

                                   " '%s': %s" % (ninfo.name, msg),

                                   errors.ECODE_ENVIRON)

      node_helper = helpers[ninfo.uuid].payload

      if node_helper != drbd_helper:

        raise errors.OpPrereqError("Error on node '%s': drbd helper is %s" %

                                   (ninfo.name, node_helper),

                                   errors.ECODE_ENVIRON)

  def _CheckDrbdHelper(self, node_uuids, drbd_enabled, drbd_gets_enabled):

    """Check the DRBD usermode helper.

    @type node_uuids: list of strings

    @param node_uuids: a list of nodes' UUIDs

    @type drbd_enabled: boolean

    @param drbd_enabled: whether DRBD will be enabled after this operation

      (no matter if it was disabled before or not)

    @type drbd_gets_enabled: boolen

    @param drbd_gets_enabled: true if DRBD was disabled before this

      operation, but will be enabled afterwards

    """

    if self.op.drbd_helper == '':

      if drbd_enabled:

        raise errors.OpPrereqError("Cannot disable drbd helper while"

                                   " DRBD is enabled.")

      if self.cfg.HasAnyDiskOfType(constants.DT_DRBD8):

        raise errors.OpPrereqError("Cannot disable drbd helper while"

                                   " drbd-based instances exist",

                                   errors.ECODE_INVAL)

    else:

      if self.op.drbd_helper is not None and drbd_enabled:

        self._CheckDrbdHelperOnNodes(self.op.drbd_helper, node_uuids)

      else:

        if drbd_gets_enabled:

          current_drbd_helper = self.cfg.GetClusterInfo().drbd_usermode_helper

          if current_drbd_helper is not None:

            self._CheckDrbdHelperOnNodes(current_drbd_helper, node_uuids)

          else:

            raise errors.OpPrereqError("Cannot enable DRBD without a"

                                       " DRBD usermode helper set.")

  def CheckPrereq(self):

    """Check prerequisites.

    This checks whether the given params don't conflict and

    if the given volume group is valid.

    """

    node_uuids = self.owned_locks(locking.LEVEL_NODE)

    self.cluster = cluster = self.cfg.GetClusterInfo()

    vm_capable_node_uuids = [node.uuid

                             for node in self.cfg.GetAllNodesInfo().values()

                             if node.uuid in node_uuids and node.vm_capable]

    (enabled_disk_templates, new_enabled_disk_templates) = \

      self._GetEnabledDiskTemplates(cluster)

    self._CheckVgName(vm_capable_node_uuids, enabled_disk_templates,

                      new_enabled_disk_templates)

    if self.op.file_storage_dir is not None:

      CheckFileStoragePathVsEnabledDiskTemplates(

          self.LogWarning, self.op.file_storage_dir, enabled_disk_templates)

    if self.op.shared_file_storage_dir is not None:

      CheckSharedFileStoragePathVsEnabledDiskTemplates(

          self.LogWarning, self.op.shared_file_storage_dir,

          enabled_disk_templates)

    drbd_enabled = constants.DT_DRBD8 in enabled_disk_templates

    drbd_gets_enabled = constants.DT_DRBD8 in new_enabled_disk_templates

    self._CheckDrbdHelper(node_uuids, drbd_enabled, drbd_gets_enabled)

    # validate params changes

    if self.op.beparams:

      objects.UpgradeBeParams(self.op.beparams)

      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)

      self.new_beparams = cluster.SimpleFillBE(self.op.beparams)

    if self.op.ndparams:

      utils.ForceDictType(self.op.ndparams, constants.NDS_PARAMETER_TYPES)

      self.new_ndparams = cluster.SimpleFillND(self.op.ndparams)

      # TODO: we need a more general way to handle resetting

      # cluster-level parameters to default values

      if self.new_ndparams["oob_program"] == "":

        self.new_ndparams["oob_program"] = \

            constants.NDC_DEFAULTS[constants.ND_OOB_PROGRAM]

    if self.op.hv_state:

      new_hv_state = MergeAndVerifyHvState(self.op.hv_state,

                                           self.cluster.hv_state_static)

      self.new_hv_state = dict((hv, cluster.SimpleFillHvState(values))

                               for hv, values in new_hv_state.items())

    if self.op.disk_state:

      new_disk_state = MergeAndVerifyDiskState(self.op.disk_state,

                                               self.cluster.disk_state_static)

      self.new_disk_state = \

        dict((storage, dict((name, cluster.SimpleFillDiskState(values))

                            for name, values in svalues.items()))

             for storage, svalues in new_disk_state.items())

    self._CheckIpolicy(cluster, enabled_disk_templates)

    if self.op.nicparams:

      utils.ForceDictType(self.op.nicparams, constants.NICS_PARAMETER_TYPES)

      self.new_nicparams = cluster.SimpleFillNIC(self.op.nicparams)

      objects.NIC.CheckParameterSyntax(self.new_nicparams)

      nic_errors = []

      # check all instances for consistency

      for instance in self.cfg.GetAllInstancesInfo().values():

        for nic_idx, nic in enumerate(instance.nics):

          params_copy = copy.deepcopy(nic.nicparams)

          params_filled = objects.FillDict(self.new_nicparams, params_copy)

          # check parameter syntax

          try:

            objects.NIC.CheckParameterSyntax(params_filled)

          except errors.ConfigurationError, err:

            nic_errors.append("Instance %s, nic/%d: %s" %

                              (instance.name, nic_idx, err))

          # if we're moving instances to routed, check that they have an ip

          target_mode = params_filled[constants.NIC_MODE]

          if target_mode == constants.NIC_MODE_ROUTED and not nic.ip:

            nic_errors.append("Instance %s, nic/%d: routed NIC with no ip"

                              " address" % (instance.name, nic_idx))

      if nic_errors:

        raise errors.OpPrereqError("Cannot apply the change, errors:\n%s" %

                                   "\n".join(nic_errors), errors.ECODE_INVAL)

    # hypervisor list/parameters

    self.new_hvparams = new_hvp = objects.FillDict(cluster.hvparams, {})

    if self.op.hvparams:

      for hv_name, hv_dict in self.op.hvparams.items():

        if hv_name not in self.new_hvparams:

          self.new_hvparams[hv_name] = hv_dict

        else:

          self.new_hvparams[hv_name].update(hv_dict)

    # disk template parameters

    self.new_diskparams = objects.FillDict(cluster.diskparams, {})

    if self.op.diskparams:

      for dt_name, dt_params in self.op.diskparams.items():

        if dt_name not in self.new_diskparams:

          self.new_diskparams[dt_name] = dt_params

        else:

          self.new_diskparams[dt_name].update(dt_params)

      CheckDiskAccessModeConsistency(self.op.diskparams, self.cfg)

    # os hypervisor parameters

    self.new_os_hvp = objects.FillDict(cluster.os_hvp, {})

    if self.op.os_hvp:

      for os_name, hvs in self.op.os_hvp.items():

        if os_name not in self.new_os_hvp:

          self.new_os_hvp[os_name] = hvs

        else:

          for hv_name, hv_dict in hvs.items():

            if hv_dict is None:

              # Delete if it exists

              self.new_os_hvp[os_name].pop(hv_name, None)

            elif hv_name not in self.new_os_hvp[os_name]:

              self.new_os_hvp[os_name][hv_name] = hv_dict

            else:

              self.new_os_hvp[os_name][hv_name].update(hv_dict)

    # os parameters

    self.new_osp = objects.FillDict(cluster.osparams, {})

    if self.op.osparams:

      for os_name, osp in self.op.osparams.items():

        if os_name not in self.new_osp:

          self.new_osp[os_name] = {}

        self.new_osp[os_name] = GetUpdatedParams(self.new_osp[os_name], osp,

                                                 use_none=True)

        if not self.new_osp[os_name]:

          # we removed all parameters

          del self.new_osp[os_name]

        else:

          # check the parameter validity (remote check)

          CheckOSParams(self, False, [self.cfg.GetMasterNode()],

                        os_name, self.new_osp[os_name])

    # changes to the hypervisor list

    if self.op.enabled_hypervisors is not None:

      self.hv_list = self.op.enabled_hypervisors

      for hv in self.hv_list:

        # if the hypervisor doesn't already exist in the cluster

        # hvparams, we initialize it to empty, and then (in both

        # cases) we make sure to fill the defaults, as we might not

        # have a complete defaults list if the hypervisor wasn't

        # enabled before

        if hv not in new_hvp:

          new_hvp[hv] = {}

        new_hvp[hv] = objects.FillDict(constants.HVC_DEFAULTS[hv], new_hvp[hv])

        utils.ForceDictType(new_hvp[hv], constants.HVS_PARAMETER_TYPES)

    else:

      self.hv_list = cluster.enabled_hypervisors

    if self.op.hvparams or self.op.enabled_hypervisors is not None:

      # either the enabled list has changed, or the parameters have, validate

      for hv_name, hv_params in self.new_hvparams.items():

        if ((self.op.hvparams and hv_name in self.op.hvparams) or

            (self.op.enabled_hypervisors and

             hv_name in self.op.enabled_hypervisors)):

          # either this is a new hypervisor, or its parameters have changed

          hv_class = hypervisor.GetHypervisorClass(hv_name)

          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)

          hv_class.CheckParameterSyntax(hv_params)

          CheckHVParams(self, node_uuids, hv_name, hv_params)

    self._CheckDiskTemplateConsistency()

    if self.op.os_hvp:

      # no need to check any newly-enabled hypervisors, since the

      # defaults have already been checked in the above code-block

      for os_name, os_hvp in self.new_os_hvp.items():