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

import ganeti.rpc.node as 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, CreateNewClientCert

import ganeti.masterd.instance

def _UpdateMasterClientCert(

    lu, master_uuid, cluster, feedback_fn,

    client_cert=pathutils.NODED_CLIENT_CERT_FILE,

    client_cert_tmp=pathutils.NODED_CLIENT_CERT_FILE_TMP):

  """Renews the master's client certificate and propagates the config.

  @type lu: C{LogicalUnit}

  @param lu: the logical unit holding the config

  @type master_uuid: string

  @param master_uuid: the master node's UUID

  @type cluster: C{objects.Cluster}

  @param cluster: the cluster's configuration

  @type feedback_fn: function

  @param feedback_fn: feedback functions for config updates

  @type client_cert: string

  @param client_cert: the path of the client certificate

  @type client_cert_tmp: string

  @param client_cert_tmp: the temporary path of the client certificate

  @rtype: string

  @return: the digest of the newly created client certificate

  """

  client_digest = CreateNewClientCert(lu, master_uuid, filename=client_cert_tmp)

  utils.AddNodeToCandidateCerts(master_uuid, client_digest,

                                cluster.candidate_certs)

  # This triggers an update of the config and distribution of it with the old

  # SSL certificate

  lu.cfg.Update(cluster, feedback_fn)

  utils.RemoveFile(client_cert)

  utils.RenameFile(client_cert_tmp, client_cert)

  return client_digest

class LUClusterRenewCrypto(NoHooksLU):

  """Renew the cluster's crypto tokens.

  Note that most of this operation is done in gnt_cluster.py, this LU only

  takes care of the renewal of the client SSL certificates.

  """

  def Exec(self, feedback_fn):

    master_uuid = self.cfg.GetMasterNode()

    cluster = self.cfg.GetClusterInfo()

    server_digest = utils.GetCertificateDigest(

      cert_filename=pathutils.NODED_CERT_FILE)

    utils.AddNodeToCandidateCerts("%s-SERVER" % master_uuid,

                                  server_digest,

                                  cluster.candidate_certs)

    try:

      old_master_digest = utils.GetCertificateDigest(

        cert_filename=pathutils.NODED_CLIENT_CERT_FILE)

      utils.AddNodeToCandidateCerts("%s-OLDMASTER" % master_uuid,

                                    old_master_digest,

                                    cluster.candidate_certs)

    except IOError:

      logging.info("No old certificate available.")

    new_master_digest = _UpdateMasterClientCert(self, master_uuid, cluster,

                                                feedback_fn)

    utils.AddNodeToCandidateCerts(master_uuid,

                                  new_master_digest,

                                  cluster.candidate_certs)

    nodes = self.cfg.GetAllNodesInfo()

    for (node_uuid, node_info) in nodes.items():

      if node_uuid != master_uuid:

        new_digest = CreateNewClientCert(self, node_uuid)

        if node_info.master_candidate:

          utils.AddNodeToCandidateCerts(node_uuid,

                                        new_digest,

                                        cluster.candidate_certs)

    utils.RemoveNodeFromCandidateCerts("%s-SERVER" % master_uuid,

                                       cluster.candidate_certs)

    utils.RemoveNodeFromCandidateCerts("%s-OLDMASTER" % master_uuid,

                                       cluster.candidate_certs)

    # Trigger another update of the config now with the new master cert

    self.cfg.Update(cluster, feedback_fn)

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 CheckArguments(self):

    self.master_uuid = self.cfg.GetMasterNode()

    self.master_ndparams = self.cfg.GetNdParams(self.cfg.GetMasterNodeInfo())

    # TODO: When Issue 584 is solved, and None is properly parsed when used

    # as a default value, ndparams.get(.., None) can be changed to

    # ndparams[..] to access the values directly

    # OpenvSwitch: Warn user if link is missing

    if (self.master_ndparams[constants.ND_OVS] and not

        self.master_ndparams.get(constants.ND_OVS_LINK, None)):

      self.LogInfo("No physical interface for OpenvSwitch was given."

                   " OpenvSwitch will not have an outside connection. This"

                   " might not be what you want.")

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

    """Create and configure Open vSwitch

    """

    if self.master_ndparams[constants.ND_OVS]:

      result = self.rpc.call_node_configure_ovs(

                 self.master_uuid,

                 self.master_ndparams[constants.ND_OVS_NAME],

                 self.master_ndparams.get(constants.ND_OVS_LINK, None))

      result.Raise("Could not successully configure Open vSwitch")

    cluster = self.cfg.GetClusterInfo()

    _UpdateMasterClientCert(self, self.master_uuid, cluster, feedback_fn)

    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,

      "max_running_jobs": cluster.max_running_jobs,

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

      "default_iallocator_params": cluster.default_iallocator_params,

      "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.GetDiskTemplatesOfStorageTypes(

            constants.ST_FILE, constants.ST_SHARED_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}