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

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,

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

      }

    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.GetMultiInstanceInfo(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.LD_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():

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

      for dsk in newl:

        self.cfg.SetDiskID(dsk, node_uuid)

      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)

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)

      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.

    """

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

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

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

                                   " instances exist", errors.ECODE_INVAL)

    if (self.op.vg_name is not None and

        utils.IsLvmEnabled(enabled_disk_templates)) or \

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

            utils.LvmGetsEnabled(enabled_disk_templates,

                                 new_enabled_disk_templates)):

      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)

  def _GetEnabledDiskTemplates(self, cluster):

    """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 self.op.enabled_disk_templates:

      enabled_disk_templates = self.op.enabled_disk_templates

      new_enabled_disk_templates = \

        list(set(enabled_disk_templates)

             - set(cluster.enabled_disk_templates))

    else:

      enabled_disk_templates = cluster.enabled_disk_templates

    return (enabled_disk_templates, new_enabled_disk_templates)

  def CheckPrereq(self):

    """Check prerequisites.

    This checks whether the given params don't conflict and

    if the given volume group is valid.

    """

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

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

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

                                   " drbd-based instances exist",

                                   errors.ECODE_INVAL)

    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.drbd_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 != self.op.drbd_helper:

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

                                     (ninfo.name, node_helper),

                                     errors.ECODE_ENVIRON)

    # 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())

    if self.op.ipolicy:

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

                                           group_policy=False)

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

    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.op.diskparams:

          self.new_diskparams[dt_name] = dt_params

        else:

          self.new_diskparams[dt_name].update(dt_params)

    # 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():

        for hv_name, hv_params in os_hvp.items():

          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)

          # we need to fill in the new os_hvp on top of the actual hv_p

          cluster_defaults = self.new_hvparams.get(hv_name, {})

          new_osp = objects.FillDict(cluster_defaults, hv_params)

          hv_class = hypervisor.GetHypervisorClass(hv_name)

          hv_class.CheckParameterSyntax(new_osp)

          CheckHVParams(self, node_uuids, hv_name, new_osp)

    if self.op.default_iallocator:

      alloc_script = utils.FindFile(self.op.default_iallocator,

                                    constants.IALLOCATOR_SEARCH_PATH,

                                    os.path.isfile)

      if alloc_script is None:

        raise errors.OpPrereqError("Invalid default iallocator script '%s'"

                                   " specified" % self.op.default_iallocator,

                                   errors.ECODE_INVAL)

  def _CheckDiskTemplateConsistency(self):

    """Check whether the disk templates that are going to be disabled

       are still in use by some instances.

    """

    if self.op.enabled_disk_templates:

      cluster = self.cfg.GetClusterInfo()

      instances = self.cfg.GetAllInstancesInfo()

      disk_templates_to_remove = set(cluster.enabled_disk_templates) \

        - set(self.op.enabled_disk_templates)

      for instance in instances.itervalues():

        if instance.disk_template in disk_templates_to_remove:

          raise errors.OpPrereqError("Cannot disable disk template '%s',"

                                     " because instance '%s' is using it." %

                                     (instance.disk_template, instance.name))

  def _SetVgName(self, feedback_fn):

    """Determines and sets the new volume group name.

    """

    if self.op.vg_name is not None:

      if self.op.vg_name and not \

           utils.IsLvmEnabled(self.cluster.enabled_disk_templates):

        feedback_fn("Note that you specified a volume group, but did not"

                    " enable any lvm disk template.")

      new_volume = self.op.vg_name

      if not new_volume:

        if utils.IsLvmEnabled(self.cluster.enabled_disk_templates):

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

                                     " disk templates are enabled.")

        new_volume = None

      if new_volume != self.cfg.GetVGName():

        self.cfg.SetVGName(new_volume)

      else:

        feedback_fn("Cluster LVM configuration already in desired"

                    " state, not changing")

    else:

      if utils.IsLvmEnabled(self.cluster.enabled_disk_templates) and \

          not self.cfg.GetVGName():

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

                                   " enabling lvm-based disk-templates.")

  def Exec(self, feedback_fn):

    """Change the parameters of the cluster.

    """

    if self.op.enabled_disk_templates:

      self.cluster.enabled_disk_templates = \

        list(set(self.op.enabled_disk_templates))

    self._SetVgName(feedback_fn)

    if self.op.drbd_helper is not None:

      if not constants.DT_DRBD8 in self.cluster.enabled_disk_templates:

        feedback_fn("Note that you specified a drbd user helper, but did"

                    " enabled the drbd disk template.")

      new_helper = self.op.drbd_helper

      if not new_helper:

        new_helper = None

      if new_helper != self.cfg.GetDRBDHelper():

        self.cfg.SetDRBDHelper(new_helper)

      else:

        feedback_fn("Cluster DRBD helper already in desired state,"

                    " not changing")

    if self.op.hvparams:

      self.cluster.hvparams = self.new_hvparams

    if self.op.os_hvp:

      self.cluster.os_hvp = self.new_os_hvp

    if self.op.enabled_hypervisors is not None:

      self.cluster.hvparams = self.new_hvparams

      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors

    if self.op.beparams:

      self.cluster.beparams[constants.PP_DEFAULT] = self.new_beparams

    if self.op.nicparams:

      self.cluster.nicparams[constants.PP_DEFAULT] = self.new_nicparams

    if self.op.ipolicy:

      self.cluster.ipolicy = self.new_ipolicy

    if self.op.osparams:

      self.cluster.osparams = self.new_osp

    if self.op.ndparams:

      self.cluster.ndparams = self.new_ndparams

    if self.op.diskparams:

      self.cluster.diskparams = self.new_diskparams

    if self.op.hv_state:

      self.cluster.hv_state_static = self.new_hv_state

    if self.op.disk_state:

      self.cluster.disk_state_static = self.new_disk_state

    if self.op.candidate_pool_size is not None:

      self.cluster.candidate_pool_size = self.op.candidate_pool_size

      # we need to update the pool size here, otherwise the save will fail

      AdjustCandidatePool(self, [])

    if self.op.maintain_node_health is not None:

      if self.op.maintain_node_health and not constants.ENABLE_CONFD:

        feedback_fn("Note: CONFD was disabled at build time, node health"

                    " maintenance is not useful (still enabling it)")

      self.cluster.maintain_node_health = self.op.maintain_node_health

    if self.op.prealloc_wipe_disks is not None:

      self.cluster.prealloc_wipe_disks = self.op.prealloc_wipe_disks

    if self.op.add_uids is not None:

      uidpool.AddToUidPool(self.cluster.uid_pool, self.op.add_uids)

    if self.op.remove_uids is not None:

      uidpool.RemoveFromUidPool(self.cluster.uid_pool, self.op.remove_uids)

    if self.op.uid_pool is not None:

      self.cluster.uid_pool = self.op.uid_pool

    if self.op.default_iallocator is not None:

      self.cluster.default_iallocator = self.op.default_iallocator

    if self.op.reserved_lvs is not None:

      self.cluster.reserved_lvs = self.op.reserved_lvs

    if self.op.use_external_mip_script is not None:

      self.cluster.use_external_mip_script = self.op.use_external_mip_script

    def helper_os(aname, mods, desc):

      desc += " OS list"

      lst = getattr(self.cluster, aname)

      for key, val in mods:

        if key == constants.DDM_ADD:

          if val in lst:

            feedback_fn("OS %s already in %s, ignoring" % (val, desc))

          else:

            lst.append(val)

        elif key == constants.DDM_REMOVE:

          if val in lst:

            lst.remove(val)

          else:

            feedback_fn("OS %s not found in %s, ignoring" % (val, desc))

        else:

          raise errors.ProgrammerError("Invalid modification '%s'" % key)

    if self.op.hidden_os:

      helper_os("hidden_os", self.op.hidden_os, "hidden")

    if self.op.blacklisted_os:

      helper_os("blacklisted_os", self.op.blacklisted_os, "blacklisted")

    if self.op.master_netdev: