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

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