Synnefo » snf-ganeti

#

#

# Copyright (C) 2010, 2011, 2012 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.

"""Module implementing the parameter types code."""

import re

import operator

import ipaddr

from ganeti import compat

from ganeti import utils

from ganeti import constants

from ganeti import objects

_PAREN_RE = re.compile("^[a-zA-Z0-9_-]+$")

def Parens(text):

  """Enclose text in parens if necessary.

  @param text: Text

  """

  text = str(text)

  if _PAREN_RE.match(text):

    return text

  else:

    return "(%s)" % text

class _WrapperBase(object):

  __slots__ = [

    "_fn",

    "_text",

    ]

  def __init__(self, text, fn):

    """Initializes this class.

    @param text: Description

    @param fn: Wrapped function

    """

    assert text.strip()

    self._text = text

    self._fn = fn

  def __call__(self, *args):

    return self._fn(*args)

class _DescWrapper(_WrapperBase):

  """Wrapper class for description text.

  """

  def __str__(self):

    return self._text

class _CommentWrapper(_WrapperBase):

  """Wrapper class for comment.

  """

  def __str__(self):

    return "%s [%s]" % (self._fn, self._text)

def WithDesc(text):

  """Builds wrapper class with description text.

  @type text: string

  @param text: Description text

  @return: Callable class

  """

  assert text[0] == text[0].upper()

  return compat.partial(_DescWrapper, text)

def Comment(text):

  """Builds wrapper for adding comment to description text.

  @type text: string

  @param text: Comment text

  @return: Callable class

  """

  assert not frozenset(text).intersection("[]")

  return compat.partial(_CommentWrapper, text)

def CombinationDesc(op, args, fn):

  """Build description for combinating operator.

  @type op: string

  @param op: Operator as text (e.g. "and")

  @type args: list

  @param args: Operator arguments

  @type fn: callable

  @param fn: Wrapped function

  """

  # Some type descriptions are rather long. If "None" is listed at the

  # end or somewhere in between it is easily missed. Therefore it should

  # be at the beginning, e.g. "None or (long description)".

  if __debug__ and TNone in args and args.index(TNone) > 0:

    raise Exception("TNone must be listed first")

  if len(args) == 1:

    descr = str(args[0])

  else:

    descr = (" %s " % op).join(Parens(i) for i in args)

  return WithDesc(descr)(fn)

# Modifiable default values; need to define these here before the

# actual LUs

@WithDesc(str([]))

def EmptyList():

  """Returns an empty list.

  """

  return []

@WithDesc(str({}))

def EmptyDict():

  """Returns an empty dict.

  """

  return {}

#: The without-default default value

NoDefault = object()

#: The no-type (value too complex to check it in the type system)

NoType = object()

# Some basic types

@WithDesc("Anything")

def TAny(_):

  """Accepts any value.

  """

  return True

@WithDesc("NotNone")

def TNotNone(val):

  """Checks if the given value is not None.

  """

  return val is not None

@WithDesc("None")

def TNone(val):

  """Checks if the given value is None.

  """

  return val is None

@WithDesc("ValueNone")

def TValueNone(val):

  """Checks if the given value is L{constants.VALUE_NONE}.

  """

  return val == constants.VALUE_NONE

@WithDesc("Boolean")

def TBool(val):

  """Checks if the given value is a boolean.

  """

  return isinstance(val, bool)

@WithDesc("Integer")

def TInt(val):

  """Checks if the given value is an integer.

  """

  # For backwards compatibility with older Python versions, boolean values are

  # also integers and should be excluded in this test.

  #

  # >>> (isinstance(False, int), isinstance(True, int))

  # (True, True)

  return isinstance(val, (int, long)) and not isinstance(val, bool)

@WithDesc("Float")

def TFloat(val):

  """Checks if the given value is a float.

  """

  return isinstance(val, float)

@WithDesc("String")

def TString(val):

  """Checks if the given value is a string.

  """

  return isinstance(val, basestring)

@WithDesc("EvalToTrue")

def TTrue(val):

  """Checks if a given value evaluates to a boolean True value.

  """

  return bool(val)

def TElemOf(target_list):

  """Builds a function that checks if a given value is a member of a list.

  """

  def fn(val):

    return val in target_list

  return WithDesc("OneOf %s" % (utils.CommaJoin(target_list), ))(fn)

# Container types

@WithDesc("List")

def TList(val):

  """Checks if the given value is a list.

  """

  return isinstance(val, list)

@WithDesc("Tuple")

def TTuple(val):

  """Checks if the given value is a tuple.

  """

  return isinstance(val, tuple)

@WithDesc("Dictionary")

def TDict(val):

  """Checks if the given value is a dictionary.

  """

  return isinstance(val, dict)

def TIsLength(size):

  """Check is the given container is of the given size.

  """

  def fn(container):

    return len(container) == size

  return WithDesc("Length %s" % (size, ))(fn)

# Combinator types

def TAnd(*args):

  """Combine multiple functions using an AND operation.

  """

  def fn(val):

    return compat.all(t(val) for t in args)

  return CombinationDesc("and", args, fn)

def TOr(*args):

  """Combine multiple functions using an OR operation.

  """

  def fn(val):

    return compat.any(t(val) for t in args)

  return CombinationDesc("or", args, fn)

def TMap(fn, test):

  """Checks that a modified version of the argument passes the given test.

  """

  return WithDesc("Result of %s must be %s" %

                  (Parens(fn), Parens(test)))(lambda val: test(fn(val)))

def TRegex(pobj):

  """Checks whether a string matches a specific regular expression.

  @param pobj: Compiled regular expression as returned by C{re.compile}

  """

  desc = WithDesc("String matching regex \"%s\"" %

                  pobj.pattern.encode("string_escape"))

  return desc(TAnd(TString, pobj.match))

def TMaybe(test):

  """Wrap a test in a TOr(TNone, test).

  This makes it easier to define TMaybe* types.

  """

  return TOr(TNone, test)

def TMaybeValueNone(test):

  """Used for unsetting values.

  """

  return TMaybe(TOr(TValueNone, test))

# Type aliases

#: a non-empty string

TNonEmptyString = WithDesc("NonEmptyString")(TAnd(TString, TTrue))

#: a maybe non-empty string

TMaybeString = TMaybe(TNonEmptyString)

#: a maybe boolean (bool or none)

TMaybeBool = TMaybe(TBool)

#: Maybe a dictionary (dict or None)

TMaybeDict = TMaybe(TDict)

#: Maybe a list (list or None)

TMaybeList = TMaybe(TList)

#: a non-negative number (value > 0)

# val_type should be TInt, TDouble (== TFloat), or TNumber

def TNonNegative(val_type):

  return WithDesc("EqualOrGreaterThanZero")(TAnd(val_type, lambda v: v >= 0))

#: a positive number (value >= 0)

# val_type should be TInt, TDouble (== TFloat), or TNumber

def TPositive(val_type):

  return WithDesc("GreaterThanZero")(TAnd(val_type, lambda v: v > 0))

#: a non-negative integer (value >= 0)

TNonNegativeInt = TNonNegative(TInt)

#: a positive integer (value > 0)

TPositiveInt = TPositive(TInt)

#: a maybe positive integer (positive integer or None)

TMaybePositiveInt = TMaybe(TPositiveInt)

#: a negative integer (value < 0)

TNegativeInt = \

  TAnd(TInt, WithDesc("LessThanZero")(compat.partial(operator.gt, 0)))

#: a positive float

TNonNegativeFloat = \

  TAnd(TFloat, WithDesc("EqualOrGreaterThanZero")(lambda v: v >= 0.0))

#: Job ID

TJobId = WithDesc("JobId")(TOr(TNonNegativeInt,

                               TRegex(re.compile("^%s$" %

                                                 constants.JOB_ID_TEMPLATE))))

#: Double (== Float)

TDouble = TFloat

#: Number

TNumber = TOr(TInt, TFloat)

#: Relative job ID

TRelativeJobId = WithDesc("RelativeJobId")(TNegativeInt)

def TInstanceOf(cls):

  """Checks if a given value is an instance of C{cls}.

  @type cls: class

  @param cls: Class object

  """

  name = "%s.%s" % (cls.__module__, cls.__name__)

  desc = WithDesc("Instance of %s" % (Parens(name), ))

  return desc(lambda val: isinstance(val, cls))

def TListOf(my_type):

  """Checks if a given value is a list with all elements of the same type.

  """

  desc = WithDesc("List of %s" % (Parens(my_type), ))

  return desc(TAnd(TList, lambda lst: compat.all(my_type(v) for v in lst)))

TMaybeListOf = lambda item_type: TMaybe(TListOf(item_type))

def TTupleOf(*val_types):

  """Checks if a given value is a list with the proper size and its

     elements match the given types.

  """

  desc = WithDesc("Tuple of %s" % (Parens(val_types), ))

  return desc(TAnd(TIsLength(len(val_types)), TItems(val_types)))

def TSetOf(val_type):

  """Checks if a given value is a list with all elements of the same

     type and eliminates duplicated elements.

  """

  desc = WithDesc("Set of %s" % (Parens(val_type), ))

  return desc(lambda st: TListOf(val_type)(list(set(st))))

def TDictOf(key_type, val_type):

  """Checks a dict type for the type of its key/values.

  """

  desc = WithDesc("Dictionary with keys of %s and values of %s" %

                  (Parens(key_type), Parens(val_type)))

  def fn(container):

    return (compat.all(key_type(v) for v in container.keys()) and

            compat.all(val_type(v) for v in container.values()))

  return desc(TAnd(TDict, fn))

def _TStrictDictCheck(require_all, exclusive, items, val):

  """Helper function for L{TStrictDict}.

  """

  notfound_fn = lambda _: not exclusive

  if require_all and not frozenset(val.keys()).issuperset(items.keys()):

    # Requires items not found in value

    return False

  return compat.all(items.get(key, notfound_fn)(value)

                    for (key, value) in val.items())

def TStrictDict(require_all, exclusive, items):

  """Strict dictionary check with specific keys.

  @type require_all: boolean

  @param require_all: Whether all keys in L{items} are required

  @type exclusive: boolean

  @param exclusive: Whether only keys listed in L{items} should be accepted

  @type items: dictionary

  @param items: Mapping from key (string) to verification function

  """

  descparts = ["Dictionary containing"]

  if exclusive:

    descparts.append(" none but the")

  if require_all:

    descparts.append(" required")

  if len(items) == 1:

    descparts.append(" key ")

  else:

    descparts.append(" keys ")

  descparts.append(utils.CommaJoin("\"%s\" (value %s)" % (key, value)

                                   for (key, value) in items.items()))

  desc = WithDesc("".join(descparts))

  return desc(TAnd(TDict,

                   compat.partial(_TStrictDictCheck, require_all, exclusive,

                                  items)))

def TItems(items):

  """Checks individual items of a container.

  If the verified value and the list of expected items differ in length, this

  check considers only as many items as are contained in the shorter list. Use

  L{TIsLength} to enforce a certain length.

  @type items: list

  @param items: List of checks

  """

  assert items, "Need items"

  text = ["Item", "item"]

  desc = WithDesc(utils.CommaJoin("%s %s is %s" %

                                  (text[int(idx > 0)], idx, Parens(check))

                                  for (idx, check) in enumerate(items)))

  return desc(lambda value: compat.all(check(i)

                                       for (check, i) in zip(items, value)))

TAllocPolicy = TElemOf(constants.VALID_ALLOC_POLICIES)

TCVErrorCode = TElemOf(constants.CV_ALL_ECODES_STRINGS)

TQueryResultCode = TElemOf(constants.RS_ALL)

TExportTarget = TOr(TNonEmptyString, TList)

TExportMode = TElemOf(constants.EXPORT_MODES)

TDiskIndex = TAnd(TNonNegativeInt, lambda val: val < constants.MAX_DISKS)

TReplaceDisksMode = TElemOf(constants.REPLACE_MODES)

TDiskTemplate = TElemOf(constants.DISK_TEMPLATES)

TNodeEvacMode = TElemOf(constants.IALLOCATOR_NEVAC_MODES)

TIAllocatorTestDir = TElemOf(constants.VALID_IALLOCATOR_DIRECTIONS)

TIAllocatorMode = TElemOf(constants.VALID_IALLOCATOR_MODES)

def TSetParamsMods(fn):

  """Generates a check for modification lists.

  """

  # Old format

  # TODO: Remove in version 2.11 including support in LUInstanceSetParams

  old_mod_item_fn = \

    TAnd(TIsLength(2),

         TItems([TOr(TElemOf(constants.DDMS_VALUES), TNonNegativeInt), fn]))

  # New format, supporting adding/removing disks/NICs at arbitrary indices

  mod_item_fn = \

      TAnd(TIsLength(3), TItems([

        TElemOf(constants.DDMS_VALUES_WITH_MODIFY),

        Comment("Device index, can be negative, e.g. -1 for last disk")

                 (TOr(TInt, TString)),

        fn,

        ]))

  return TOr(Comment("Recommended")(TListOf(mod_item_fn)),

             Comment("Deprecated")(TListOf(old_mod_item_fn)))

TINicParams = \

    Comment("NIC parameters")(TDictOf(TElemOf(constants.INIC_PARAMS),

                                      TMaybeString))

TIDiskParams = \

    Comment("Disk parameters")(TDictOf(TElemOf(constants.IDISK_PARAMS),

                                       TOr(TNonEmptyString, TInt)))

THypervisor = TElemOf(constants.HYPER_TYPES)

TMigrationMode = TElemOf(constants.HT_MIGRATION_MODES)

TNICMode = TElemOf(constants.NIC_VALID_MODES)

TInstCreateMode = TElemOf(constants.INSTANCE_CREATE_MODES)

TRebootType = TElemOf(constants.REBOOT_TYPES)

TFileDriver = TElemOf(constants.FILE_DRIVER)

TOobCommand = TElemOf(constants.OOB_COMMANDS)

TQueryTypeOp = TElemOf(constants.QR_VIA_OP)

TDiskParams = \

    Comment("Disk parameters")(TDictOf(TNonEmptyString,

                                       TOr(TNonEmptyString, TInt)))

TDiskChanges = \

    TAnd(TIsLength(2),

         TItems([Comment("Disk index")(TNonNegativeInt),

                 Comment("Parameters")(TDiskParams)]))

TRecreateDisksInfo = TOr(TListOf(TNonNegativeInt), TListOf(TDiskChanges))

def TStorageType(val):

  """Builds a function that checks if a given value is a valid storage

  type.

  """

  return (val in constants.STORAGE_TYPES)

TTagKind = TElemOf(constants.VALID_TAG_TYPES)

TDdmSimple = TElemOf(constants.DDMS_VALUES)

TVerifyOptionalChecks = TElemOf(constants.VERIFY_OPTIONAL_CHECKS)

@WithDesc("IPv4 network")

def _CheckCIDRNetNotation(value):

  """Ensure a given CIDR notation type is valid.

  """

  try:

    ipaddr.IPv4Network(value)

  except ipaddr.AddressValueError:

    return False

  return True

@WithDesc("IPv4 address")

def _CheckCIDRAddrNotation(value):

  """Ensure a given CIDR notation type is valid.

  """

  try:

    ipaddr.IPv4Address(value)

  except ipaddr.AddressValueError:

    return False

  return True

@WithDesc("IPv6 address")

def _CheckCIDR6AddrNotation(value):

  """Ensure a given CIDR notation type is valid.

  """

  try:

    ipaddr.IPv6Address(value)

  except ipaddr.AddressValueError:

    return False

  return True

@WithDesc("IPv6 network")

def _CheckCIDR6NetNotation(value):

  """Ensure a given CIDR notation type is valid.

  """

  try:

    ipaddr.IPv6Network(value)

  except ipaddr.AddressValueError:

    return False

  return True

TIPv4Address = TAnd(TString, _CheckCIDRAddrNotation)

TIPv6Address = TAnd(TString, _CheckCIDR6AddrNotation)

TIPv4Network = TAnd(TString, _CheckCIDRNetNotation)

TIPv6Network = TAnd(TString, _CheckCIDR6NetNotation)

def TObject(val_type):

  return TDictOf(TAny, val_type)

def TObjectCheck(obj, fields_types):

  """Helper to generate type checks for objects.

  @param obj: The object to generate type checks

  @param fields_types: The fields and their types as a dict

  @return: A ht type check function

  """

  assert set(obj.GetAllSlots()) == set(fields_types.keys()), \

    "%s != %s" % (set(obj.GetAllSlots()), set(fields_types.keys()))

  return TStrictDict(True, True, fields_types)

TQueryFieldDef = \

    TObjectCheck(objects.QueryFieldDefinition, {

        "name": TNonEmptyString,

        "title": TNonEmptyString,

        "kind": TElemOf(constants.QFT_ALL),

        "doc": TNonEmptyString

    })

TQueryRow = \

    TListOf(TAnd(TIsLength(2),

                 TItems([TElemOf(constants.RS_ALL), TAny])))

TQueryResult = TListOf(TQueryRow)

TQueryResponse = \

    TObjectCheck(objects.QueryResponse, {

        "fields": TListOf(TQueryFieldDef),

        "data": TQueryResult

    })

TQueryFieldsResponse = \

    TObjectCheck(objects.QueryFieldsResponse, {

        "fields": TListOf(TQueryFieldDef)

    })

TJobIdListItem = \

    TAnd(TIsLength(2),

         TItems([Comment("success")(TBool),

                 Comment("Job ID if successful, error message"

                         " otherwise")(TOr(TString, TJobId))]))

TJobIdList = TListOf(TJobIdListItem)

TJobIdListOnly = TStrictDict(True, True, {

  constants.JOB_IDS_KEY: Comment("List of submitted jobs")(TJobIdList)

  })

TInstanceMultiAllocResponse = \

    TStrictDict(True, True, {

      constants.JOB_IDS_KEY: Comment("List of submitted jobs")(TJobIdList),

      constants.ALLOCATABLE_KEY: TListOf(TNonEmptyString),

      constants.FAILED_KEY: TListOf(TNonEmptyString)

    })