Synnefo » snf-ganeti » ganeti-local

#

#

# Copyright (C) 2006, 2007, 2008, 2009, 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.

"""OpCodes module

This module implements the data structures which define the cluster

operations - the so-called opcodes.

Every operation which modifies the cluster state is expressed via

opcodes.

"""

# this are practically structures, so disable the message about too

# few public methods:

# pylint: disable=R0903

import logging

import re

import ipaddr

from ganeti import constants

from ganeti import errors

from ganeti import ht

from ganeti import objects

from ganeti import objectutils

# Common opcode attributes

#: output fields for a query operation

_POutputFields = ("output_fields", ht.NoDefault, ht.TListOf(ht.TNonEmptyString),

                  "Selected output fields")

#: the shutdown timeout

_PShutdownTimeout = \

  ("shutdown_timeout", constants.DEFAULT_SHUTDOWN_TIMEOUT, ht.TNonNegativeInt,

   "How long to wait for instance to shut down")

#: the force parameter

_PForce = ("force", False, ht.TBool, "Whether to force the operation")

#: a required instance name (for single-instance LUs)

_PInstanceName = ("instance_name", ht.NoDefault, ht.TNonEmptyString,

                  "Instance name")

#: Whether to ignore offline nodes

_PIgnoreOfflineNodes = ("ignore_offline_nodes", False, ht.TBool,

                        "Whether to ignore offline nodes")

#: a required node name (for single-node LUs)

_PNodeName = ("node_name", ht.NoDefault, ht.TNonEmptyString, "Node name")

#: a required node group name (for single-group LUs)

_PGroupName = ("group_name", ht.NoDefault, ht.TNonEmptyString, "Group name")

#: Migration type (live/non-live)

_PMigrationMode = ("mode", None,

                   ht.TMaybe(ht.TElemOf(constants.HT_MIGRATION_MODES)),

                   "Migration mode")

#: Obsolete 'live' migration mode (boolean)

_PMigrationLive = ("live", None, ht.TMaybeBool,

                   "Legacy setting for live migration, do not use")

#: Tag type

_PTagKind = ("kind", ht.NoDefault, ht.TElemOf(constants.VALID_TAG_TYPES),

             "Tag kind")

#: List of tag strings

_PTags = ("tags", ht.NoDefault, ht.TListOf(ht.TNonEmptyString),

          "List of tag names")

_PForceVariant = ("force_variant", False, ht.TBool,

                  "Whether to force an unknown OS variant")

_PWaitForSync = ("wait_for_sync", True, ht.TBool,

                 "Whether to wait for the disk to synchronize")

_PWaitForSyncFalse = ("wait_for_sync", False, ht.TBool,

                      "Whether to wait for the disk to synchronize"

                      " (defaults to false)")

_PIgnoreConsistency = ("ignore_consistency", False, ht.TBool,

                       "Whether to ignore disk consistency")

_PStorageName = ("name", ht.NoDefault, ht.TMaybeString, "Storage name")

_PUseLocking = ("use_locking", False, ht.TBool,

                "Whether to use synchronization")

_PNameCheck = ("name_check", True, ht.TBool, "Whether to check name")

_PNodeGroupAllocPolicy = \

  ("alloc_policy", None,

   ht.TMaybe(ht.TElemOf(constants.VALID_ALLOC_POLICIES)),

   "Instance allocation policy")

_PGroupNodeParams = ("ndparams", None, ht.TMaybeDict,

                     "Default node parameters for group")

_PQueryWhat = ("what", ht.NoDefault, ht.TElemOf(constants.QR_VIA_OP),

               "Resource(s) to query for")

_PEarlyRelease = ("early_release", False, ht.TBool,

                  "Whether to release locks as soon as possible")

_PIpCheckDoc = "Whether to ensure instance's IP address is inactive"

#: Do not remember instance state changes

_PNoRemember = ("no_remember", False, ht.TBool,

                "Do not remember the state change")

#: Target node for instance migration/failover

_PMigrationTargetNode = ("target_node", None, ht.TMaybeString,

                         "Target node for shared-storage instances")

_PStartupPaused = ("startup_paused", False, ht.TBool,

                   "Pause instance at startup")

_PVerbose = ("verbose", False, ht.TBool, "Verbose mode")

# Parameters for cluster verification

_PDebugSimulateErrors = ("debug_simulate_errors", False, ht.TBool,

                         "Whether to simulate errors (useful for debugging)")

_PErrorCodes = ("error_codes", False, ht.TBool, "Error codes")

_PSkipChecks = ("skip_checks", ht.EmptyList,

                ht.TListOf(ht.TElemOf(constants.VERIFY_OPTIONAL_CHECKS)),

                "Which checks to skip")

_PIgnoreErrors = ("ignore_errors", ht.EmptyList,

                  ht.TListOf(ht.TElemOf(constants.CV_ALL_ECODES_STRINGS)),

                  "List of error codes that should be treated as warnings")

# Disk parameters

_PDiskParams = \

  ("diskparams", None,

   ht.TMaybe(ht.TDictOf(ht.TElemOf(constants.DISK_TEMPLATES), ht.TDict)),

   "Disk templates' parameter defaults")

# Parameters for node resource model

_PHvState = ("hv_state", None, ht.TMaybeDict, "Set hypervisor states")

_PDiskState = ("disk_state", None, ht.TMaybeDict, "Set disk states")

#: Opportunistic locking

_POpportunisticLocking = \

  ("opportunistic_locking", False, ht.TBool,

   ("Whether to employ opportunistic locking for nodes, meaning nodes"

    " already locked by another opcode won't be considered for instance"

    " allocation (only when an iallocator is used)"))

_PIgnoreIpolicy = ("ignore_ipolicy", False, ht.TBool,

                   "Whether to ignore ipolicy violations")

# Allow runtime changes while migrating

_PAllowRuntimeChgs = ("allow_runtime_changes", True, ht.TBool,

                      "Allow runtime changes (eg. memory ballooning)")

#: IAllocator field builder

_PIAllocFromDesc = lambda desc: ("iallocator", None, ht.TMaybeString, desc)

#: a required network name

_PNetworkName = ("network_name", ht.NoDefault, ht.TNonEmptyString,

                 "Set network name")

#: OP_ID conversion regular expression

_OPID_RE = re.compile("([a-z])([A-Z])")

#: Utility function for L{OpClusterSetParams}

_TestClusterOsListItem = \

  ht.TAnd(ht.TIsLength(2), ht.TItems([

    ht.TElemOf(constants.DDMS_VALUES),

    ht.TNonEmptyString,

    ]))

_TestClusterOsList = ht.TMaybeListOf(_TestClusterOsListItem)

# TODO: Generate check from constants.INIC_PARAMS_TYPES

#: Utility function for testing NIC definitions

_TestNicDef = \

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

                                          ht.TMaybeString))

_TSetParamsResultItemItems = [

  ht.Comment("name of changed parameter")(ht.TNonEmptyString),

  ht.Comment("new value")(ht.TAny),

  ]

_TSetParamsResult = \

  ht.TListOf(ht.TAnd(ht.TIsLength(len(_TSetParamsResultItemItems)),

                     ht.TItems(_TSetParamsResultItemItems)))

# In the disks option we can provide arbitrary parameters too, which

# we may not be able to validate at this level, so we just check the

# format of the dict here and the checks concerning IDISK_PARAMS will

# happen at the LU level

_TDiskParams = \

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

                                           ht.TOr(ht.TNonEmptyString, ht.TInt)))

_TQueryRow = \

  ht.TListOf(ht.TAnd(ht.TIsLength(2),

                     ht.TItems([ht.TElemOf(constants.RS_ALL),

                                ht.TAny])))

_TQueryResult = ht.TListOf(_TQueryRow)

_TOldQueryRow = ht.TListOf(ht.TAny)

_TOldQueryResult = ht.TListOf(_TOldQueryRow)

_SUMMARY_PREFIX = {

  "CLUSTER_": "C_",

  "GROUP_": "G_",

  "NODE_": "N_",

  "INSTANCE_": "I_",

  }

#: Attribute name for dependencies

DEPEND_ATTR = "depends"

#: Attribute name for comment

COMMENT_ATTR = "comment"

def _NameToId(name):

  """Convert an opcode class name to an OP_ID.

  @type name: string

  @param name: the class name, as OpXxxYyy

  @rtype: string

  @return: the name in the OP_XXXX_YYYY format

  """

  if not name.startswith("Op"):

    return None

  # Note: (?<=[a-z])(?=[A-Z]) would be ideal, since it wouldn't

  # consume any input, and hence we would just have all the elements

  # in the list, one by one; but it seems that split doesn't work on

  # non-consuming input, hence we have to process the input string a

  # bit

  name = _OPID_RE.sub(r"\1,\2", name)

  elems = name.split(",")

  return "_".join(n.upper() for n in elems)

def _GenerateObjectTypeCheck(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 ht.TStrictDict(True, True, fields_types)

_TQueryFieldDef = \

  _GenerateObjectTypeCheck(objects.QueryFieldDefinition, {

    "name": ht.TNonEmptyString,

    "title": ht.TNonEmptyString,

    "kind": ht.TElemOf(constants.QFT_ALL),

    "doc": ht.TNonEmptyString,

    })

def RequireFileStorage():

  """Checks that file storage is enabled.

  While it doesn't really fit into this module, L{utils} was deemed too large

  of a dependency to be imported for just one or two functions.

  @raise errors.OpPrereqError: when file storage is disabled

  """

  if not constants.ENABLE_FILE_STORAGE:

    raise errors.OpPrereqError("File storage disabled at configure time",

                               errors.ECODE_INVAL)

def RequireSharedFileStorage():

  """Checks that shared file storage is enabled.

  While it doesn't really fit into this module, L{utils} was deemed too large

  of a dependency to be imported for just one or two functions.

  @raise errors.OpPrereqError: when shared file storage is disabled

  """

  if not constants.ENABLE_SHARED_FILE_STORAGE:

    raise errors.OpPrereqError("Shared file storage disabled at"

                               " configure time", errors.ECODE_INVAL)

@ht.WithDesc("CheckFileStorage")

def _CheckFileStorage(value):

  """Ensures file storage is enabled if used.

  """

  if value == constants.DT_FILE:

    RequireFileStorage()

  elif value == constants.DT_SHARED_FILE:

    RequireSharedFileStorage()

  return True

def _BuildDiskTemplateCheck(accept_none):

  """Builds check for disk template.

  @type accept_none: bool

  @param accept_none: whether to accept None as a correct value

  @rtype: callable

  """

  template_check = ht.TElemOf(constants.DISK_TEMPLATES)

  if accept_none:

    template_check = ht.TMaybe(template_check)

  return ht.TAnd(template_check, _CheckFileStorage)

def _CheckStorageType(storage_type):

  """Ensure a given storage type is valid.

  """

  if storage_type not in constants.VALID_STORAGE_TYPES:

    raise errors.OpPrereqError("Unknown storage type: %s" % storage_type,

                               errors.ECODE_INVAL)

  if storage_type == constants.ST_FILE:

    # TODO: What about shared file storage?

    RequireFileStorage()

  return True

#: Storage type parameter

_PStorageType = ("storage_type", ht.NoDefault, _CheckStorageType,

                 "Storage type")

_CheckNetworkType = ht.TElemOf(constants.NETWORK_VALID_TYPES)

@ht.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

@ht.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

@ht.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

@ht.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

_TIpAddress4 = ht.TAnd(ht.TString, _CheckCIDRAddrNotation)

_TIpAddress6 = ht.TAnd(ht.TString, _CheckCIDR6AddrNotation)

_TIpNetwork4 = ht.TAnd(ht.TString, _CheckCIDRNetNotation)

_TIpNetwork6 = ht.TAnd(ht.TString, _CheckCIDR6NetNotation)

_TMaybeAddr4List = ht.TMaybe(ht.TListOf(_TIpAddress4))

class _AutoOpParamSlots(objectutils.AutoSlots):

  """Meta class for opcode definitions.

  """

  def __new__(mcs, name, bases, attrs):

    """Called when a class should be created.

    @param mcs: The meta class

    @param name: Name of created class

    @param bases: Base classes

    @type attrs: dict

    @param attrs: Class attributes

    """

    assert "OP_ID" not in attrs, "Class '%s' defining OP_ID" % name

    slots = mcs._GetSlots(attrs)

    assert "OP_DSC_FIELD" not in attrs or attrs["OP_DSC_FIELD"] in slots, \

      "Class '%s' uses unknown field in OP_DSC_FIELD" % name

    assert ("OP_DSC_FORMATTER" not in attrs or

            callable(attrs["OP_DSC_FORMATTER"])), \

      ("Class '%s' uses non-callable in OP_DSC_FORMATTER (%s)" %

       (name, type(attrs["OP_DSC_FORMATTER"])))

    attrs["OP_ID"] = _NameToId(name)

    return objectutils.AutoSlots.__new__(mcs, name, bases, attrs)

  @classmethod

  def _GetSlots(mcs, attrs):

    """Build the slots out of OP_PARAMS.

    """

    # Always set OP_PARAMS to avoid duplicates in BaseOpCode.GetAllParams

    params = attrs.setdefault("OP_PARAMS", [])

    # Use parameter names as slots

    return [pname for (pname, _, _, _) in params]

class BaseOpCode(objectutils.ValidatedSlots):

  """A simple serializable object.

  This object serves as a parent class for OpCode without any custom

  field handling.

  """

  # pylint: disable=E1101

  # as OP_ID is dynamically defined

  __metaclass__ = _AutoOpParamSlots

  def __getstate__(self):

    """Generic serializer.

    This method just returns the contents of the instance as a

    dictionary.

    @rtype:  C{dict}

    @return: the instance attributes and their values

    """

    state = {}

    for name in self.GetAllSlots():

      if hasattr(self, name):

        state[name] = getattr(self, name)

    return state

  def __setstate__(self, state):

    """Generic unserializer.

    This method just restores from the serialized state the attributes

    of the current instance.

    @param state: the serialized opcode data

    @type state:  C{dict}

    """

    if not isinstance(state, dict):

      raise ValueError("Invalid data to __setstate__: expected dict, got %s" %

                       type(state))

    for name in self.GetAllSlots():

      if name not in state and hasattr(self, name):

        delattr(self, name)

    for name in state:

      setattr(self, name, state[name])

  @classmethod

  def GetAllParams(cls):

    """Compute list of all parameters for an opcode.

    """

    slots = []

    for parent in cls.__mro__:

      slots.extend(getattr(parent, "OP_PARAMS", []))

    return slots

  def Validate(self, set_defaults): # pylint: disable=W0221

    """Validate opcode parameters, optionally setting default values.

    @type set_defaults: bool

    @param set_defaults: Whether to set default values

    @raise errors.OpPrereqError: When a parameter value doesn't match

                                 requirements

    """

    for (attr_name, default, test, _) in self.GetAllParams():

      assert test == ht.NoType or callable(test)

      if not hasattr(self, attr_name):

        if default == ht.NoDefault:

          raise errors.OpPrereqError("Required parameter '%s.%s' missing" %

                                     (self.OP_ID, attr_name),

                                     errors.ECODE_INVAL)

        elif set_defaults:

          if callable(default):

            dval = default()

          else:

            dval = default

          setattr(self, attr_name, dval)

      if test == ht.NoType:

        # no tests here

        continue

      if set_defaults or hasattr(self, attr_name):

        attr_val = getattr(self, attr_name)

        if not test(attr_val):

          logging.error("OpCode %s, parameter %s, has invalid type %s/value"

                        " '%s' expecting type %s",

                        self.OP_ID, attr_name, type(attr_val), attr_val, test)

          raise errors.OpPrereqError("Parameter '%s.%s' fails validation" %

                                     (self.OP_ID, attr_name),

                                     errors.ECODE_INVAL)

def _BuildJobDepCheck(relative):

  """Builds check for job dependencies (L{DEPEND_ATTR}).

  @type relative: bool

  @param relative: Whether to accept relative job IDs (negative)

  @rtype: callable

  """

  if relative:

    job_id = ht.TOr(ht.TJobId, ht.TRelativeJobId)

  else:

    job_id = ht.TJobId

  job_dep = \

    ht.TAnd(ht.TOr(ht.TList, ht.TTuple),

            ht.TIsLength(2),

            ht.TItems([job_id,

                       ht.TListOf(ht.TElemOf(constants.JOBS_FINALIZED))]))

  return ht.TMaybeListOf(job_dep)

TNoRelativeJobDependencies = _BuildJobDepCheck(False)

#: List of submission status and job ID as returned by C{SubmitManyJobs}

_TJobIdListItem = \

  ht.TAnd(ht.TIsLength(2),

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

                     ht.Comment("Job ID if successful, error message"

                                " otherwise")(ht.TOr(ht.TString,

                                                     ht.TJobId))]))

TJobIdList = ht.TListOf(_TJobIdListItem)

#: Result containing only list of submitted jobs

TJobIdListOnly = ht.TStrictDict(True, True, {

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

  })

class OpCode(BaseOpCode):

  """Abstract OpCode.

  This is the root of the actual OpCode hierarchy. All clases derived

  from this class should override OP_ID.

  @cvar OP_ID: The ID of this opcode. This should be unique amongst all

               children of this class.

  @cvar OP_DSC_FIELD: The name of a field whose value will be included in the

                      string returned by Summary(); see the docstring of that

                      method for details).

  @cvar OP_DSC_FORMATTER: A callable that should format the OP_DSC_FIELD; if

                          not present, then the field will be simply converted

                          to string

  @cvar OP_PARAMS: List of opcode attributes, the default values they should

                   get if not already defined, and types they must match.

  @cvar OP_RESULT: Callable to verify opcode result

  @cvar WITH_LU: Boolean that specifies whether this should be included in

      mcpu's dispatch table

  @ivar dry_run: Whether the LU should be run in dry-run mode, i.e. just

                 the check steps

  @ivar priority: Opcode priority for queue

  """

  # pylint: disable=E1101

  # as OP_ID is dynamically defined

  WITH_LU = True

  OP_PARAMS = [

    ("dry_run", None, ht.TMaybeBool, "Run checks only, don't execute"),

    ("debug_level", None, ht.TMaybe(ht.TNonNegativeInt), "Debug level"),

    ("priority", constants.OP_PRIO_DEFAULT,

     ht.TElemOf(constants.OP_PRIO_SUBMIT_VALID), "Opcode priority"),

    (DEPEND_ATTR, None, _BuildJobDepCheck(True),

     "Job dependencies; if used through ``SubmitManyJobs`` relative (negative)"

     " job IDs can be used; see :doc:`design document <design-chained-jobs>`"

     " for details"),

    (COMMENT_ATTR, None, ht.TMaybeString,

     "Comment describing the purpose of the opcode"),

    ]

  OP_RESULT = None

  def __getstate__(self):

    """Specialized getstate for opcodes.

    This method adds to the state dictionary the OP_ID of the class,

    so that on unload we can identify the correct class for

    instantiating the opcode.

    @rtype:   C{dict}

    @return:  the state as a dictionary

    """

    data = BaseOpCode.__getstate__(self)

    data["OP_ID"] = self.OP_ID

    return data

  @classmethod

  def LoadOpCode(cls, data):

    """Generic load opcode method.

    The method identifies the correct opcode class from the dict-form

    by looking for a OP_ID key, if this is not found, or its value is

    not available in this module as a child of this class, we fail.

    @type data:  C{dict}

    @param data: the serialized opcode

    """

    if not isinstance(data, dict):

      raise ValueError("Invalid data to LoadOpCode (%s)" % type(data))

    if "OP_ID" not in data:

      raise ValueError("Invalid data to LoadOpcode, missing OP_ID")

    op_id = data["OP_ID"]

    op_class = None

    if op_id in OP_MAPPING:

      op_class = OP_MAPPING[op_id]

    else:

      raise ValueError("Invalid data to LoadOpCode: OP_ID %s unsupported" %

                       op_id)

    op = op_class()

    new_data = data.copy()

    del new_data["OP_ID"]

    op.__setstate__(new_data)

    return op

  def Summary(self):

    """Generates a summary description of this opcode.

    The summary is the value of the OP_ID attribute (without the "OP_"

    prefix), plus the value of the OP_DSC_FIELD attribute, if one was

    defined; this field should allow to easily identify the operation

    (for an instance creation job, e.g., it would be the instance

    name).

    """

    assert self.OP_ID is not None and len(self.OP_ID) > 3

    # all OP_ID start with OP_, we remove that

    txt = self.OP_ID[3:]

    field_name = getattr(self, "OP_DSC_FIELD", None)

    if field_name:

      field_value = getattr(self, field_name, None)

      field_formatter = getattr(self, "OP_DSC_FORMATTER", None)

      if callable(field_formatter):

        field_value = field_formatter(field_value)

      elif isinstance(field_value, (list, tuple)):

        field_value = ",".join(str(i) for i in field_value)

      txt = "%s(%s)" % (txt, field_value)

    return txt

  def TinySummary(self):

    """Generates a compact summary description of the opcode.

    """

    assert self.OP_ID.startswith("OP_")

    text = self.OP_ID[3:]

    for (prefix, supplement) in _SUMMARY_PREFIX.items():

      if text.startswith(prefix):

        return supplement + text[len(prefix):]

    return text

# cluster opcodes

class OpClusterPostInit(OpCode):

  """Post cluster initialization.

  This opcode does not touch the cluster at all. Its purpose is to run hooks

  after the cluster has been initialized.

  """

  OP_RESULT = ht.TBool

class OpClusterDestroy(OpCode):

  """Destroy the cluster.

  This opcode has no other parameters. All the state is irreversibly

  lost after the execution of this opcode.

  """

  OP_RESULT = ht.TNonEmptyString

class OpClusterQuery(OpCode):

  """Query cluster information."""

  OP_RESULT = ht.TDictOf(ht.TNonEmptyString, ht.TAny)

class OpClusterVerify(OpCode):

  """Submits all jobs necessary to verify the cluster.

  """

  OP_PARAMS = [