commit 7352d33beb716e17793874dacd1468eef6baeda7
Author: Thomas Thrainer <thomasth@google.com>
Date:   Mon May 13 13:49:33 2013 +0200

    Extract cluster related logical units from cmdlib
    
    All LUCluster* classes are extracted to cluster.py. Shared functions are
    extracted to common.py, helper functions only used by LUCluster* are
    extracted to cluster.py.
    
    Signed-off-by: Thomas Thrainer <thomasth@google.com>
    Reviewed-by: Bernardo Dal Seno <bdalseno@google.com>

diff --git a/lib/cmdlib/common.py b/lib/cmdlib/common.py
index 91403ad..4ea6941 100644
--- a/lib/cmdlib/common.py
+++ b/lib/cmdlib/common.py
@@ -20,9 +20,17 @@
 
 
 """Common functions used by multiple logical units."""
+import copy
+import os
 
+from ganeti import constants
 from ganeti import errors
+from ganeti import hypervisor
 from ganeti import locking
+from ganeti import objects
+from ganeti import pathutils
+from ganeti import rpc
+from ganeti import ssconf
 from ganeti import utils
 
 
@@ -120,3 +128,597 @@ def _GetWantedInstances(lu, instances):
   else:
     wanted = utils.NiceSort(lu.cfg.GetInstanceList())
   return wanted
+
+
+def _RunPostHook(lu, node_name):
+  """Runs the post-hook for an opcode on a single node.
+
+  """
+  hm = lu.proc.BuildHooksManager(lu)
+  try:
+    hm.RunPhase(constants.HOOKS_PHASE_POST, nodes=[node_name])
+  except Exception, err: # pylint: disable=W0703
+    lu.LogWarning("Errors occurred running hooks on %s: %s",
+                  node_name, err)
+
+
+def _RedistributeAncillaryFiles(lu, additional_nodes=None, additional_vm=True):
+  """Distribute additional files which are part of the cluster configuration.
+
+  ConfigWriter takes care of distributing the config and ssconf files, but
+  there are more files which should be distributed to all nodes. This function
+  makes sure those are copied.
+
+  @param lu: calling logical unit
+  @param additional_nodes: list of nodes not in the config to distribute to
+  @type additional_vm: boolean
+  @param additional_vm: whether the additional nodes are vm-capable or not
+
+  """
+  # Gather target nodes
+  cluster = lu.cfg.GetClusterInfo()
+  master_info = lu.cfg.GetNodeInfo(lu.cfg.GetMasterNode())
+
+  online_nodes = lu.cfg.GetOnlineNodeList()
+  online_set = frozenset(online_nodes)
+  vm_nodes = list(online_set.intersection(lu.cfg.GetVmCapableNodeList()))
+
+  if additional_nodes is not None:
+    online_nodes.extend(additional_nodes)
+    if additional_vm:
+      vm_nodes.extend(additional_nodes)
+
+  # Never distribute to master node
+  for nodelist in [online_nodes, vm_nodes]:
+    if master_info.name in nodelist:
+      nodelist.remove(master_info.name)
+
+  # Gather file lists
+  (files_all, _, files_mc, files_vm) = \
+    _ComputeAncillaryFiles(cluster, True)
+
+  # Never re-distribute configuration file from here
+  assert not (pathutils.CLUSTER_CONF_FILE in files_all or
+              pathutils.CLUSTER_CONF_FILE in files_vm)
+  assert not files_mc, "Master candidates not handled in this function"
+
+  filemap = [
+    (online_nodes, files_all),
+    (vm_nodes, files_vm),
+    ]
+
+  # Upload the files
+  for (node_list, files) in filemap:
+    for fname in files:
+      _UploadHelper(lu, node_list, fname)
+
+
+def _ComputeAncillaryFiles(cluster, redist):
+  """Compute files external to Ganeti which need to be consistent.
+
+  @type redist: boolean
+  @param redist: Whether to include files which need to be redistributed
+
+  """
+  # Compute files for all nodes
+  files_all = set([
+    pathutils.SSH_KNOWN_HOSTS_FILE,
+    pathutils.CONFD_HMAC_KEY,
+    pathutils.CLUSTER_DOMAIN_SECRET_FILE,
+    pathutils.SPICE_CERT_FILE,
+    pathutils.SPICE_CACERT_FILE,
+    pathutils.RAPI_USERS_FILE,
+    ])
+
+  if redist:
+    # we need to ship at least the RAPI certificate
+    files_all.add(pathutils.RAPI_CERT_FILE)
+  else:
+    files_all.update(pathutils.ALL_CERT_FILES)
+    files_all.update(ssconf.SimpleStore().GetFileList())
+
+  if cluster.modify_etc_hosts:
+    files_all.add(pathutils.ETC_HOSTS)
+
+  if cluster.use_external_mip_script:
+    files_all.add(pathutils.EXTERNAL_MASTER_SETUP_SCRIPT)
+
+  # Files which are optional, these must:
+  # - be present in one other category as well
+  # - either exist or not exist on all nodes of that category (mc, vm all)
+  files_opt = set([
+    pathutils.RAPI_USERS_FILE,
+    ])
+
+  # Files which should only be on master candidates
+  files_mc = set()
+
+  if not redist:
+    files_mc.add(pathutils.CLUSTER_CONF_FILE)
+
+  # File storage
+  if (not redist and (constants.ENABLE_FILE_STORAGE or
+                        constants.ENABLE_SHARED_FILE_STORAGE)):
+    files_all.add(pathutils.FILE_STORAGE_PATHS_FILE)
+    files_opt.add(pathutils.FILE_STORAGE_PATHS_FILE)
+
+  # Files which should only be on VM-capable nodes
+  files_vm = set(
+    filename
+    for hv_name in cluster.enabled_hypervisors
+    for filename in
+    hypervisor.GetHypervisorClass(hv_name).GetAncillaryFiles()[0])
+
+  files_opt |= set(
+    filename
+    for hv_name in cluster.enabled_hypervisors
+    for filename in
+    hypervisor.GetHypervisorClass(hv_name).GetAncillaryFiles()[1])
+
+  # Filenames in each category must be unique
+  all_files_set = files_all | files_mc | files_vm
+  assert (len(all_files_set) ==
+          sum(map(len, [files_all, files_mc, files_vm]))), \
+    "Found file listed in more than one file list"
+
+  # Optional files must be present in one other category
+  assert all_files_set.issuperset(files_opt), \
+    "Optional file not in a different required list"
+
+  # This one file should never ever be re-distributed via RPC
+  assert not (redist and
+              pathutils.FILE_STORAGE_PATHS_FILE in all_files_set)
+
+  return (files_all, files_opt, files_mc, files_vm)
+
+
+def _UploadHelper(lu, nodes, fname):
+  """Helper for uploading a file and showing warnings.
+
+  """
+  if os.path.exists(fname):
+    result = lu.rpc.call_upload_file(nodes, fname)
+    for to_node, to_result in result.items():
+      msg = to_result.fail_msg
+      if msg:
+        msg = ("Copy of file %s to node %s failed: %s" %
+               (fname, to_node, msg))
+        lu.LogWarning(msg)
+
+
+def _MergeAndVerifyHvState(op_input, obj_input):
+  """Combines the hv state from an opcode with the one of the object
+
+  @param op_input: The input dict from the opcode
+  @param obj_input: The input dict from the objects
+  @return: The verified and updated dict
+
+  """
+  if op_input:
+    invalid_hvs = set(op_input) - constants.HYPER_TYPES
+    if invalid_hvs:
+      raise errors.OpPrereqError("Invalid hypervisor(s) in hypervisor state:"
+                                 " %s" % utils.CommaJoin(invalid_hvs),
+                                 errors.ECODE_INVAL)
+    if obj_input is None:
+      obj_input = {}
+    type_check = constants.HVSTS_PARAMETER_TYPES
+    return _UpdateAndVerifySubDict(obj_input, op_input, type_check)
+
+  return None
+
+
+def _MergeAndVerifyDiskState(op_input, obj_input):
+  """Combines the disk state from an opcode with the one of the object
+
+  @param op_input: The input dict from the opcode
+  @param obj_input: The input dict from the objects
+  @return: The verified and updated dict
+  """
+  if op_input:
+    invalid_dst = set(op_input) - constants.DS_VALID_TYPES
+    if invalid_dst:
+      raise errors.OpPrereqError("Invalid storage type(s) in disk state: %s" %
+                                 utils.CommaJoin(invalid_dst),
+                                 errors.ECODE_INVAL)
+    type_check = constants.DSS_PARAMETER_TYPES
+    if obj_input is None:
+      obj_input = {}
+    return dict((key, _UpdateAndVerifySubDict(obj_input.get(key, {}), value,
+                                              type_check))
+                for key, value in op_input.items())
+
+  return None
+
+
+def _CheckOSParams(lu, required, nodenames, osname, osparams):
+  """OS parameters validation.
+
+  @type lu: L{LogicalUnit}
+  @param lu: the logical unit for which we check
+  @type required: boolean
+  @param required: whether the validation should fail if the OS is not
+      found
+  @type nodenames: list
+  @param nodenames: the list of nodes on which we should check
+  @type osname: string
+  @param osname: the name of the hypervisor we should use
+  @type osparams: dict
+  @param osparams: the parameters which we need to check
+  @raise errors.OpPrereqError: if the parameters are not valid
+
+  """
+  nodenames = _FilterVmNodes(lu, nodenames)
+  result = lu.rpc.call_os_validate(nodenames, required, osname,
+                                   [constants.OS_VALIDATE_PARAMETERS],
+                                   osparams)
+  for node, nres in result.items():
+    # we don't check for offline cases since this should be run only
+    # against the master node and/or an instance's nodes
+    nres.Raise("OS Parameters validation failed on node %s" % node)
+    if not nres.payload:
+      lu.LogInfo("OS %s not found on node %s, validation skipped",
+                 osname, node)
+
+
+def _CheckHVParams(lu, nodenames, hvname, hvparams):
+  """Hypervisor parameter validation.
+
+  This function abstract the hypervisor parameter validation to be
+  used in both instance create and instance modify.
+
+  @type lu: L{LogicalUnit}
+  @param lu: the logical unit for which we check
+  @type nodenames: list
+  @param nodenames: the list of nodes on which we should check
+  @type hvname: string
+  @param hvname: the name of the hypervisor we should use
+  @type hvparams: dict
+  @param hvparams: the parameters which we need to check
+  @raise errors.OpPrereqError: if the parameters are not valid
+
+  """
+  nodenames = _FilterVmNodes(lu, nodenames)
+
+  cluster = lu.cfg.GetClusterInfo()
+  hvfull = objects.FillDict(cluster.hvparams.get(hvname, {}), hvparams)
+
+  hvinfo = lu.rpc.call_hypervisor_validate_params(nodenames, hvname, hvfull)
+  for node in nodenames:
+    info = hvinfo[node]
+    if info.offline:
+      continue
+    info.Raise("Hypervisor parameter validation failed on node %s" % node)
+
+
+def _AdjustCandidatePool(lu, exceptions):
+  """Adjust the candidate pool after node operations.
+
+  """
+  mod_list = lu.cfg.MaintainCandidatePool(exceptions)
+  if mod_list:
+    lu.LogInfo("Promoted nodes to master candidate role: %s",
+               utils.CommaJoin(node.name for node in mod_list))
+    for name in mod_list:
+      lu.context.ReaddNode(name)
+  mc_now, mc_max, _ = lu.cfg.GetMasterCandidateStats(exceptions)
+  if mc_now > mc_max:
+    lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
+               (mc_now, mc_max))
+
+
+def _CheckNodePVs(nresult, exclusive_storage):
+  """Check node PVs.
+
+  """
+  pvlist_dict = nresult.get(constants.NV_PVLIST, None)
+  if pvlist_dict is None:
+    return (["Can't get PV list from node"], None)
+  pvlist = map(objects.LvmPvInfo.FromDict, pvlist_dict)
+  errlist = []
+  # check that ':' is not present in PV names, since it's a
+  # special character for lvcreate (denotes the range of PEs to
+  # use on the PV)
+  for pv in pvlist:
+    if ":" in pv.name:
+      errlist.append("Invalid character ':' in PV '%s' of VG '%s'" %
+                     (pv.name, pv.vg_name))
+  es_pvinfo = None
+  if exclusive_storage:
+    (errmsgs, es_pvinfo) = utils.LvmExclusiveCheckNodePvs(pvlist)
+    errlist.extend(errmsgs)
+    shared_pvs = nresult.get(constants.NV_EXCLUSIVEPVS, None)
+    if shared_pvs:
+      for (pvname, lvlist) in shared_pvs:
+        # TODO: Check that LVs are really unrelated (snapshots, DRBD meta...)
+        errlist.append("PV %s is shared among unrelated LVs (%s)" %
+                       (pvname, utils.CommaJoin(lvlist)))
+  return (errlist, es_pvinfo)
+
+
+def _ComputeMinMaxSpec(name, qualifier, ispecs, value):
+  """Computes if value is in the desired range.
+
+  @param name: name of the parameter for which we perform the check
+  @param qualifier: a qualifier used in the error message (e.g. 'disk/1',
+      not just 'disk')
+  @param ispecs: dictionary containing min and max values
+  @param value: actual value that we want to use
+  @return: None or an error string
+
+  """
+  if value in [None, constants.VALUE_AUTO]:
+    return None
+  max_v = ispecs[constants.ISPECS_MAX].get(name, value)
+  min_v = ispecs[constants.ISPECS_MIN].get(name, value)
+  if value > max_v or min_v > value:
+    if qualifier:
+      fqn = "%s/%s" % (name, qualifier)
+    else:
+      fqn = name
+    return ("%s value %s is not in range [%s, %s]" %
+            (fqn, value, min_v, max_v))
+  return None
+
+
+def _ComputeIPolicySpecViolation(ipolicy, mem_size, cpu_count, disk_count,
+                                 nic_count, disk_sizes, spindle_use,
+                                 disk_template,
+                                 _compute_fn=_ComputeMinMaxSpec):
+  """Verifies ipolicy against provided specs.
+
+  @type ipolicy: dict
+  @param ipolicy: The ipolicy
+  @type mem_size: int
+  @param mem_size: The memory size
+  @type cpu_count: int
+  @param cpu_count: Used cpu cores
+  @type disk_count: int
+  @param disk_count: Number of disks used
+  @type nic_count: int
+  @param nic_count: Number of nics used
+  @type disk_sizes: list of ints
+  @param disk_sizes: Disk sizes of used disk (len must match C{disk_count})
+  @type spindle_use: int
+  @param spindle_use: The number of spindles this instance uses
+  @type disk_template: string
+  @param disk_template: The disk template of the instance
+  @param _compute_fn: The compute function (unittest only)
+  @return: A list of violations, or an empty list of no violations are found
+
+  """
+  assert disk_count == len(disk_sizes)
+
+  test_settings = [
+    (constants.ISPEC_MEM_SIZE, "", mem_size),
+    (constants.ISPEC_CPU_COUNT, "", cpu_count),
+    (constants.ISPEC_NIC_COUNT, "", nic_count),
+    (constants.ISPEC_SPINDLE_USE, "", spindle_use),
+    ] + [(constants.ISPEC_DISK_SIZE, str(idx), d)
+         for idx, d in enumerate(disk_sizes)]
+  if disk_template != constants.DT_DISKLESS:
+    # This check doesn't make sense for diskless instances
+    test_settings.append((constants.ISPEC_DISK_COUNT, "", disk_count))
+  ret = []
+  allowed_dts = ipolicy[constants.IPOLICY_DTS]
+  if disk_template not in allowed_dts:
+    ret.append("Disk template %s is not allowed (allowed templates: %s)" %
+               (disk_template, utils.CommaJoin(allowed_dts)))
+
+  min_errs = None
+  for minmax in ipolicy[constants.ISPECS_MINMAX]:
+    errs = filter(None,
+                  (_compute_fn(name, qualifier, minmax, value)
+                   for (name, qualifier, value) in test_settings))
+    if min_errs is None or len(errs) < len(min_errs):
+      min_errs = errs
+  assert min_errs is not None
+  return ret + min_errs
+
+
+def _ComputeIPolicyInstanceViolation(ipolicy, instance, cfg,
+                                     _compute_fn=_ComputeIPolicySpecViolation):
+  """Compute if instance meets the specs of ipolicy.
+
+  @type ipolicy: dict
+  @param ipolicy: The ipolicy to verify against
+  @type instance: L{objects.Instance}
+  @param instance: The instance to verify
+  @type cfg: L{config.ConfigWriter}
+  @param cfg: Cluster configuration
+  @param _compute_fn: The function to verify ipolicy (unittest only)
+  @see: L{_ComputeIPolicySpecViolation}
+
+  """
+  be_full = cfg.GetClusterInfo().FillBE(instance)
+  mem_size = be_full[constants.BE_MAXMEM]
+  cpu_count = be_full[constants.BE_VCPUS]
+  spindle_use = be_full[constants.BE_SPINDLE_USE]
+  disk_count = len(instance.disks)
+  disk_sizes = [disk.size for disk in instance.disks]
+  nic_count = len(instance.nics)
+  disk_template = instance.disk_template
+
+  return _compute_fn(ipolicy, mem_size, cpu_count, disk_count, nic_count,
+                     disk_sizes, spindle_use, disk_template)
+
+
+def _ComputeViolatingInstances(ipolicy, instances, cfg):
+  """Computes a set of instances who violates given ipolicy.
+
+  @param ipolicy: The ipolicy to verify
+  @type instances: L{objects.Instance}
+  @param instances: List of instances to verify
+  @type cfg: L{config.ConfigWriter}
+  @param cfg: Cluster configuration
+  @return: A frozenset of instance names violating the ipolicy
+
+  """
+  return frozenset([inst.name for inst in instances
+                    if _ComputeIPolicyInstanceViolation(ipolicy, inst, cfg)])
+
+
+def _ComputeNewInstanceViolations(old_ipolicy, new_ipolicy, instances, cfg):
+  """Computes a set of any instances that would violate the new ipolicy.
+
+  @param old_ipolicy: The current (still in-place) ipolicy
+  @param new_ipolicy: The new (to become) ipolicy
+  @param instances: List of instances to verify
+  @type cfg: L{config.ConfigWriter}
+  @param cfg: Cluster configuration
+  @return: A list of instances which violates the new ipolicy but
+      did not before
+
+  """
+  return (_ComputeViolatingInstances(new_ipolicy, instances, cfg) -
+          _ComputeViolatingInstances(old_ipolicy, instances, cfg))
+
+
+def _GetUpdatedParams(old_params, update_dict,
+                      use_default=True, use_none=False):
+  """Return the new version of a parameter dictionary.
+
+  @type old_params: dict
+  @param old_params: old parameters
+  @type update_dict: dict
+  @param update_dict: dict containing new parameter values, or
+      constants.VALUE_DEFAULT to reset the parameter to its default
+      value
+  @param use_default: boolean
+  @type use_default: whether to recognise L{constants.VALUE_DEFAULT}
+      values as 'to be deleted' values
+  @param use_none: boolean
+  @type use_none: whether to recognise C{None} values as 'to be
+      deleted' values
+  @rtype: dict
+  @return: the new parameter dictionary
+
+  """
+  params_copy = copy.deepcopy(old_params)
+  for key, val in update_dict.iteritems():
+    if ((use_default and val == constants.VALUE_DEFAULT) or
+          (use_none and val is None)):
+      try:
+        del params_copy[key]
+      except KeyError:
+        pass
+    else:
+      params_copy[key] = val
+  return params_copy
+
+
+def _GetUpdatedIPolicy(old_ipolicy, new_ipolicy, group_policy=False):
+  """Return the new version of an instance policy.
+
+  @param group_policy: whether this policy applies to a group and thus
+    we should support removal of policy entries
+
+  """
+  ipolicy = copy.deepcopy(old_ipolicy)
+  for key, value in new_ipolicy.items():
+    if key not in constants.IPOLICY_ALL_KEYS:
+      raise errors.OpPrereqError("Invalid key in new ipolicy: %s" % key,
+                                 errors.ECODE_INVAL)
+    if (not value or value == [constants.VALUE_DEFAULT] or
+            value == constants.VALUE_DEFAULT):
+      if group_policy:
+        if key in ipolicy:
+          del ipolicy[key]
+      else:
+        raise errors.OpPrereqError("Can't unset ipolicy attribute '%s'"
+                                   " on the cluster'" % key,
+                                   errors.ECODE_INVAL)
+    else:
+      if key in constants.IPOLICY_PARAMETERS:
+        # FIXME: we assume all such values are float
+        try:
+          ipolicy[key] = float(value)
+        except (TypeError, ValueError), err:
+          raise errors.OpPrereqError("Invalid value for attribute"
+                                     " '%s': '%s', error: %s" %
+                                     (key, value, err), errors.ECODE_INVAL)
+      elif key == constants.ISPECS_MINMAX:
+        for minmax in value:
+          for k in minmax.keys():
+            utils.ForceDictType(minmax[k], constants.ISPECS_PARAMETER_TYPES)
+        ipolicy[key] = value
+      elif key == constants.ISPECS_STD:
+        if group_policy:
+          msg = "%s cannot appear in group instance specs" % key
+          raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
+        ipolicy[key] = _GetUpdatedParams(old_ipolicy.get(key, {}), value,
+                                         use_none=False, use_default=False)
+        utils.ForceDictType(ipolicy[key], constants.ISPECS_PARAMETER_TYPES)
+      else:
+        # FIXME: we assume all others are lists; this should be redone
+        # in a nicer way
+        ipolicy[key] = list(value)
+  try:
+    objects.InstancePolicy.CheckParameterSyntax(ipolicy, not group_policy)
+  except errors.ConfigurationError, err:
+    raise errors.OpPrereqError("Invalid instance policy: %s" % err,
+                               errors.ECODE_INVAL)
+  return ipolicy
+
+
+def _AnnotateDiskParams(instance, devs, cfg):
+  """Little helper wrapper to the rpc annotation method.
+
+  @param instance: The instance object
+  @type devs: List of L{objects.Disk}
+  @param devs: The root devices (not any of its children!)
+  @param cfg: The config object
+  @returns The annotated disk copies
+  @see L{rpc.AnnotateDiskParams}
+
+  """
+  return rpc.AnnotateDiskParams(instance.disk_template, devs,
+                                cfg.GetInstanceDiskParams(instance))
+
+
+def _SupportsOob(cfg, node):
+  """Tells if node supports OOB.
+
+  @type cfg: L{config.ConfigWriter}
+  @param cfg: The cluster configuration
+  @type node: L{objects.Node}
+  @param node: The node
+  @return: The OOB script if supported or an empty string otherwise
+
+  """
+  return cfg.GetNdParams(node)[constants.ND_OOB_PROGRAM]
+
+
+def _UpdateAndVerifySubDict(base, updates, type_check):
+  """Updates and verifies a dict with sub dicts of the same type.
+
+  @param base: The dict with the old data
+  @param updates: The dict with the new data
+  @param type_check: Dict suitable to ForceDictType to verify correct types
+  @returns: A new dict with updated and verified values
+
+  """
+  def fn(old, value):
+    new = _GetUpdatedParams(old, value)
+    utils.ForceDictType(new, type_check)
+    return new
+
+  ret = copy.deepcopy(base)
+  ret.update(dict((key, fn(base.get(key, {}), value))
+                  for key, value in updates.items()))
+  return ret
+
+
+def _FilterVmNodes(lu, nodenames):
+  """Filters out non-vm_capable nodes from a list.
+
+  @type lu: L{LogicalUnit}
+  @param lu: the logical unit for which we check
+  @type nodenames: list
+  @param nodenames: the list of nodes on which we should check
+  @rtype: list
+  @return: the list of vm-capable nodes
+
+  """
+  vm_nodes = frozenset(lu.cfg.GetNonVmCapableNodeList())
+  return [name for name in nodenames if name not in vm_nodes]
