Synnefo » snf-ganeti

from ganeti.cmdlib.common import _ExpandInstanceName, _ExpandItemName, \

  _CheckInstanceNodeGroups

# States of instance

INSTANCE_DOWN = [constants.ADMINST_DOWN]

INSTANCE_ONLINE = [constants.ADMINST_DOWN, constants.ADMINST_UP]

INSTANCE_NOT_RUNNING = [constants.ADMINST_DOWN, constants.ADMINST_OFFLINE]

#: Instance status in which an instance can be marked as offline/online

CAN_CHANGE_INSTANCE_OFFLINE = (frozenset(INSTANCE_DOWN) | frozenset([

  constants.ADMINST_OFFLINE,

  ]))

def _IsExclusiveStorageEnabledNode(cfg, node):

  """Whether exclusive_storage is in effect for the given node.

  @type cfg: L{config.ConfigWriter}

  @param cfg: The cluster configuration

  @type node: L{objects.Node}

  @param node: The node

  @rtype: bool

  @return: The effective value of exclusive_storage

  """

  return cfg.GetNdParams(node)[constants.ND_EXCLUSIVE_STORAGE]

def _CheckOutputFields(static, dynamic, selected):

  """Checks whether all selected fields are valid.

  @type static: L{utils.FieldSet}

  @param static: static fields set

  @type dynamic: L{utils.FieldSet}

  @param dynamic: dynamic fields set

  """

  f = utils.FieldSet()

  f.Extend(static)

  f.Extend(dynamic)

  delta = f.NonMatching(selected)

  if delta:

    raise errors.OpPrereqError("Unknown output fields selected: %s"

                               % ",".join(delta), errors.ECODE_INVAL)

def _CheckParamsNotGlobal(params, glob_pars, kind, bad_levels, good_levels):

  """Make sure that none of the given paramters is global.

  If a global parameter is found, an L{errors.OpPrereqError} exception is

  raised. This is used to avoid setting global parameters for individual nodes.

  @type params: dictionary

  @param params: Parameters to check

  @type glob_pars: dictionary

  @param glob_pars: Forbidden parameters

  @type kind: string

  @param kind: Kind of parameters (e.g. "node")

  @type bad_levels: string

  @param bad_levels: Level(s) at which the parameters are forbidden (e.g.

      "instance")

  @type good_levels: strings

  @param good_levels: Level(s) at which the parameters are allowed (e.g.

      "cluster or group")

  """

  used_globals = glob_pars.intersection(params)

  if used_globals:

    msg = ("The following %s parameters are global and cannot"

           " be customized at %s level, please modify them at"

           " %s level: %s" %

           (kind, bad_levels, good_levels, utils.CommaJoin(used_globals)))

    raise errors.OpPrereqError(msg, errors.ECODE_INVAL)

def _CheckNodeHasSecondaryIP(lu, node, secondary_ip, prereq):

  """Ensure that a node has the given secondary ip.

  @type lu: L{LogicalUnit}

  @param lu: the LU on behalf of which we make the check

  @type node: string

  @param node: the node to check

  @type secondary_ip: string

  @param secondary_ip: the ip to check

  @type prereq: boolean

  @param prereq: whether to throw a prerequisite or an execute error

  @raise errors.OpPrereqError: if the node doesn't have the ip, and prereq=True

  @raise errors.OpExecError: if the node doesn't have the ip, and prereq=False

  """

  result = lu.rpc.call_node_has_ip_address(node, secondary_ip)

  result.Raise("Failure checking secondary ip on node %s" % node,

               prereq=prereq, ecode=errors.ECODE_ENVIRON)

  if not result.payload:

    msg = ("Node claims it doesn't have the secondary ip you gave (%s),"

           " please fix and re-run this command" % secondary_ip)

    if prereq:

      raise errors.OpPrereqError(msg, errors.ECODE_ENVIRON)

    else:

      raise errors.OpExecError(msg)

def _CheckInstanceState(lu, instance, req_states, msg=None):

  """Ensure that an instance is in one of the required states.

  @param lu: the LU on behalf of which we make the check

  @param instance: the instance to check

  @param msg: if passed, should be a message to replace the default one

  @raise errors.OpPrereqError: if the instance is not in the required state

  """

  if msg is None:

    msg = ("can't use instance from outside %s states" %

           utils.CommaJoin(req_states))

  if instance.admin_state not in req_states:

    raise errors.OpPrereqError("Instance '%s' is marked to be %s, %s" %

                               (instance.name, instance.admin_state, msg),

                               errors.ECODE_STATE)

  if constants.ADMINST_UP not in req_states:

    pnode = instance.primary_node

    if not lu.cfg.GetNodeInfo(pnode).offline:

      ins_l = lu.rpc.call_instance_list([pnode], [instance.hypervisor])[pnode]

      ins_l.Raise("Can't contact node %s for instance information" % pnode,

                  prereq=True, ecode=errors.ECODE_ENVIRON)

      if instance.name in ins_l.payload:

        raise errors.OpPrereqError("Instance %s is running, %s" %

                                   (instance.name, msg), errors.ECODE_STATE)

    else:

      lu.LogWarning("Primary node offline, ignoring check that instance"

                     " is down")

def _DecideSelfPromotion(lu, exceptions=None):

  """Decide whether I should promote myself as a master candidate.

  """

  cp_size = lu.cfg.GetClusterInfo().candidate_pool_size

  mc_now, mc_should, _ = lu.cfg.GetMasterCandidateStats(exceptions)

  # the new node will increase mc_max with one, so:

  mc_should = min(mc_should + 1, cp_size)

  return mc_now < mc_should

def _GetNodeInstancesInner(cfg, fn):

  return [i for i in cfg.GetAllInstancesInfo().values() if fn(i)]

def _GetNodeInstances(cfg, node_name):

  """Returns a list of all primary and secondary instances on a node.

  """

  return _GetNodeInstancesInner(cfg, lambda inst: node_name in inst.all_nodes)

def _GetNodePrimaryInstances(cfg, node_name):

  """Returns primary instances on a node.

  """

  return _GetNodeInstancesInner(cfg,

                                lambda inst: node_name == inst.primary_node)

def _GetNodeSecondaryInstances(cfg, node_name):

  """Returns secondary instances on a node.

  """

  return _GetNodeInstancesInner(cfg,

                                lambda inst: node_name in inst.secondary_nodes)

def _GetStorageTypeArgs(cfg, storage_type):

  """Returns the arguments for a storage type.

  """

  # Special case for file storage

  if storage_type == constants.ST_FILE:

    # storage.FileStorage wants a list of storage directories

    return [[cfg.GetFileStorageDir(), cfg.GetSharedFileStorageDir()]]

  return []

def _FindFaultyInstanceDisks(cfg, rpc_runner, instance, node_name, prereq):

  faulty = []

  for dev in instance.disks:

    cfg.SetDiskID(dev, node_name)

  result = rpc_runner.call_blockdev_getmirrorstatus(node_name, (instance.disks,

                                                                instance))

  result.Raise("Failed to get disk status from node %s" % node_name,

               prereq=prereq, ecode=errors.ECODE_ENVIRON)

  for idx, bdev_status in enumerate(result.payload):

    if bdev_status and bdev_status.ldisk_status == constants.LDS_FAULTY:

      faulty.append(idx)

  return faulty

def _CheckIAllocatorOrNode(lu, iallocator_slot, node_slot):

  """Check the sanity of iallocator and node arguments and use the

  cluster-wide iallocator if appropriate.

  Check that at most one of (iallocator, node) is specified. If none is

  specified, or the iallocator is L{constants.DEFAULT_IALLOCATOR_SHORTCUT},

  then the LU's opcode's iallocator slot is filled with the cluster-wide

  default iallocator.

  @type iallocator_slot: string

  @param iallocator_slot: the name of the opcode iallocator slot

  @type node_slot: string

  @param node_slot: the name of the opcode target node slot

  """

  node = getattr(lu.op, node_slot, None)

  ialloc = getattr(lu.op, iallocator_slot, None)

  if node == []:

    node = None

  if node is not None and ialloc is not None:

    raise errors.OpPrereqError("Do not specify both, iallocator and node",

                               errors.ECODE_INVAL)

  elif ((node is None and ialloc is None) or

        ialloc == constants.DEFAULT_IALLOCATOR_SHORTCUT):

    default_iallocator = lu.cfg.GetDefaultIAllocator()

    if default_iallocator:

      setattr(lu.op, iallocator_slot, default_iallocator)

    else:

      raise errors.OpPrereqError("No iallocator or node given and no"

                                 " cluster-wide default iallocator found;"

                                 " please specify either an iallocator or a"

                                 " node, or set a cluster-wide default"

                                 " iallocator", errors.ECODE_INVAL)

class LUNodeRemove(LogicalUnit):

  """Logical unit for removing a node.

  """

  HPATH = "node-remove"

  HTYPE = constants.HTYPE_NODE

  def BuildHooksEnv(self):

    """Build hooks env.

    """

    return {

      "OP_TARGET": self.op.node_name,

      "NODE_NAME": self.op.node_name,

      }

  def BuildHooksNodes(self):

    """Build hooks nodes.

    This doesn't run on the target node in the pre phase as a failed

    node would then be impossible to remove.

    """

    all_nodes = self.cfg.GetNodeList()

    try:

      all_nodes.remove(self.op.node_name)

    except ValueError:

      pass

    return (all_nodes, all_nodes)

  def CheckPrereq(self):

    """Check prerequisites.

    This checks:

     - the node exists in the configuration

     - it does not have primary or secondary instances

     - it's not the master

    Any errors are signaled by raising errors.OpPrereqError.

    """

    self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)

    node = self.cfg.GetNodeInfo(self.op.node_name)

    assert node is not None

    masternode = self.cfg.GetMasterNode()

    if node.name == masternode:

      raise errors.OpPrereqError("Node is the master node, failover to another"

                                 " node is required", errors.ECODE_INVAL)

    for instance_name, instance in self.cfg.GetAllInstancesInfo().items():

      if node.name in instance.all_nodes:

        raise errors.OpPrereqError("Instance %s is still running on the node,"

                                   " please remove first" % instance_name,

                                   errors.ECODE_INVAL)

    self.op.node_name = node.name

    self.node = node

  def Exec(self, feedback_fn):

    """Removes the node from the cluster.

    """

    node = self.node

    logging.info("Stopping the node daemon and removing configs from node %s",

                 node.name)

    modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup

    assert locking.BGL in self.owned_locks(locking.LEVEL_CLUSTER), \

      "Not owning BGL"

    # Promote nodes to master candidate as needed

    _AdjustCandidatePool(self, exceptions=[node.name])

    self.context.RemoveNode(node.name)

    # Run post hooks on the node before it's removed

    _RunPostHook(self, node.name)

    result = self.rpc.call_node_leave_cluster(node.name, modify_ssh_setup)

    msg = result.fail_msg

    if msg:

      self.LogWarning("Errors encountered on the remote node while leaving"

                      " the cluster: %s", msg)

    # Remove node from our /etc/hosts

    if self.cfg.GetClusterInfo().modify_etc_hosts:

      master_node = self.cfg.GetMasterNode()

      result = self.rpc.call_etc_hosts_modify(master_node,

                                              constants.ETC_HOSTS_REMOVE,

                                              node.name, None)

      result.Raise("Can't update hosts file with new host data")

      _RedistributeAncillaryFiles(self)

class _NodeQuery(_QueryBase):

  FIELDS = query.NODE_FIELDS

  def ExpandNames(self, lu):

    lu.needed_locks = {}

    lu.share_locks = _ShareAll()

    if self.names:

      self.wanted = _GetWantedNodes(lu, self.names)

    else:

      self.wanted = locking.ALL_SET

    self.do_locking = (self.use_locking and

                       query.NQ_LIVE in self.requested_data)

    if self.do_locking:

      # If any non-static field is requested we need to lock the nodes

      lu.needed_locks[locking.LEVEL_NODE] = self.wanted

      lu.needed_locks[locking.LEVEL_NODE_ALLOC] = locking.ALL_SET

  def DeclareLocks(self, lu, level):

    pass

  def _GetQueryData(self, lu):

    """Computes the list of nodes and their attributes.

    """

    all_info = lu.cfg.GetAllNodesInfo()

    nodenames = self._GetNames(lu, all_info.keys(), locking.LEVEL_NODE)

    # Gather data as requested

    if query.NQ_LIVE in self.requested_data:

      # filter out non-vm_capable nodes

      toquery_nodes = [name for name in nodenames if all_info[name].vm_capable]

      es_flags = rpc.GetExclusiveStorageForNodeNames(lu.cfg, toquery_nodes)

      node_data = lu.rpc.call_node_info(toquery_nodes, [lu.cfg.GetVGName()],

                                        [lu.cfg.GetHypervisorType()], es_flags)

      live_data = dict((name, rpc.MakeLegacyNodeInfo(nresult.payload))

                       for (name, nresult) in node_data.items()

                       if not nresult.fail_msg and nresult.payload)

    else:

      live_data = None

    if query.NQ_INST in self.requested_data:

      node_to_primary = dict([(name, set()) for name in nodenames])

      node_to_secondary = dict([(name, set()) for name in nodenames])

      inst_data = lu.cfg.GetAllInstancesInfo()

      for inst in inst_data.values():

        if inst.primary_node in node_to_primary:

          node_to_primary[inst.primary_node].add(inst.name)

        for secnode in inst.secondary_nodes:

          if secnode in node_to_secondary:

            node_to_secondary[secnode].add(inst.name)

    else:

      node_to_primary = None

      node_to_secondary = None

    if query.NQ_OOB in self.requested_data:

      oob_support = dict((name, bool(_SupportsOob(lu.cfg, node)))

                         for name, node in all_info.iteritems())

    else:

      oob_support = None

    if query.NQ_GROUP in self.requested_data:

      groups = lu.cfg.GetAllNodeGroupsInfo()

    else:

      groups = {}

    return query.NodeQueryData([all_info[name] for name in nodenames],

                               live_data, lu.cfg.GetMasterNode(),

                               node_to_primary, node_to_secondary, groups,

                               oob_support, lu.cfg.GetClusterInfo())

class LUNodeQuery(NoHooksLU):

  """Logical unit for querying nodes.

  """

  # pylint: disable=W0142

  REQ_BGL = False

  def CheckArguments(self):

    self.nq = _NodeQuery(qlang.MakeSimpleFilter("name", self.op.names),

                         self.op.output_fields, self.op.use_locking)

  def ExpandNames(self):

    self.nq.ExpandNames(self)

  def DeclareLocks(self, level):

    self.nq.DeclareLocks(self, level)

  def Exec(self, feedback_fn):

    return self.nq.OldStyleQuery(self)

class LUNodeQueryvols(NoHooksLU):

  """Logical unit for getting volumes on node(s).

  """

  REQ_BGL = False

  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")

  _FIELDS_STATIC = utils.FieldSet("node")

  def CheckArguments(self):

    _CheckOutputFields(static=self._FIELDS_STATIC,

                       dynamic=self._FIELDS_DYNAMIC,

                       selected=self.op.output_fields)

  def ExpandNames(self):

    self.share_locks = _ShareAll()

    if self.op.nodes:

      self.needed_locks = {

        locking.LEVEL_NODE: _GetWantedNodes(self, self.op.nodes),

        }

    else:

      self.needed_locks = {

        locking.LEVEL_NODE: locking.ALL_SET,

        locking.LEVEL_NODE_ALLOC: locking.ALL_SET,

        }

  def Exec(self, feedback_fn):

    """Computes the list of nodes and their attributes.

    """

    nodenames = self.owned_locks(locking.LEVEL_NODE)

    volumes = self.rpc.call_node_volumes(nodenames)

    ilist = self.cfg.GetAllInstancesInfo()

    vol2inst = _MapInstanceDisksToNodes(ilist.values())

    output = []

    for node in nodenames:

      nresult = volumes[node]

      if nresult.offline:

        continue

      msg = nresult.fail_msg

      if msg:

        self.LogWarning("Can't compute volume data on node %s: %s", node, msg)

        continue

      node_vols = sorted(nresult.payload,

                         key=operator.itemgetter("dev"))

      for vol in node_vols:

        node_output = []

        for field in self.op.output_fields:

          if field == "node":

            val = node

          elif field == "phys":

            val = vol["dev"]

          elif field == "vg":

            val = vol["vg"]

          elif field == "name":

            val = vol["name"]

          elif field == "size":

            val = int(float(vol["size"]))

          elif field == "instance":

            val = vol2inst.get((node, vol["vg"] + "/" + vol["name"]), "-")

          else:

            raise errors.ParameterError(field)

          node_output.append(str(val))

        output.append(node_output)

    return output

class LUNodeQueryStorage(NoHooksLU):

  """Logical unit for getting information on storage units on node(s).

  """

  _FIELDS_STATIC = utils.FieldSet(constants.SF_NODE)

  REQ_BGL = False

  def CheckArguments(self):

    _CheckOutputFields(static=self._FIELDS_STATIC,

                       dynamic=utils.FieldSet(*constants.VALID_STORAGE_FIELDS),

                       selected=self.op.output_fields)

  def ExpandNames(self):

    self.share_locks = _ShareAll()

    if self.op.nodes:

      self.needed_locks = {

        locking.LEVEL_NODE: _GetWantedNodes(self, self.op.nodes),

        }

    else:

      self.needed_locks = {

        locking.LEVEL_NODE: locking.ALL_SET,

        locking.LEVEL_NODE_ALLOC: locking.ALL_SET,

        }

  def Exec(self, feedback_fn):

    """Computes the list of nodes and their attributes.

    """

    self.nodes = self.owned_locks(locking.LEVEL_NODE)

    # Always get name to sort by

    if constants.SF_NAME in self.op.output_fields:

      fields = self.op.output_fields[:]

    else:

      fields = [constants.SF_NAME] + self.op.output_fields

    # Never ask for node or type as it's only known to the LU

    for extra in [constants.SF_NODE, constants.SF_TYPE]:

      while extra in fields:

        fields.remove(extra)

    field_idx = dict([(name, idx) for (idx, name) in enumerate(fields)])

    name_idx = field_idx[constants.SF_NAME]

    st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)

    data = self.rpc.call_storage_list(self.nodes,

                                      self.op.storage_type, st_args,

                                      self.op.name, fields)

    result = []

    for node in utils.NiceSort(self.nodes):

      nresult = data[node]

      if nresult.offline:

        continue

      msg = nresult.fail_msg

      if msg:

        self.LogWarning("Can't get storage data from node %s: %s", node, msg)

        continue

      rows = dict([(row[name_idx], row) for row in nresult.payload])

      for name in utils.NiceSort(rows.keys()):

        row = rows[name]

        out = []

        for field in self.op.output_fields:

          if field == constants.SF_NODE:

            val = node

          elif field == constants.SF_TYPE:

            val = self.op.storage_type

          elif field in field_idx:

            val = row[field_idx[field]]

          else:

            raise errors.ParameterError(field)

          out.append(val)

        result.append(out)

    return result

    for instance_name in owned_instances:

      _CheckInstanceNodeGroups(lu.cfg, instance_name, owned_groups)

  def _GetQueryData(self, lu):

    """Computes the list of instances and their attributes.

    """

    if self.do_grouplocks:

      self._CheckGroupLocks(lu)

    cluster = lu.cfg.GetClusterInfo()

    all_info = lu.cfg.GetAllInstancesInfo()

    instance_names = self._GetNames(lu, all_info.keys(), locking.LEVEL_INSTANCE)

    instance_list = [all_info[name] for name in instance_names]

    nodes = frozenset(itertools.chain(*(inst.all_nodes

                                        for inst in instance_list)))

    hv_list = list(set([inst.hypervisor for inst in instance_list]))

    bad_nodes = []

    offline_nodes = []

    wrongnode_inst = set()

    # Gather data as requested

    if self.requested_data & set([query.IQ_LIVE, query.IQ_CONSOLE]):

      live_data = {}

      node_data = lu.rpc.call_all_instances_info(nodes, hv_list)

      for name in nodes:

        result = node_data[name]

        if result.offline:

          # offline nodes will be in both lists

          assert result.fail_msg

          offline_nodes.append(name)

        if result.fail_msg:

          bad_nodes.append(name)

        elif result.payload:

          for inst in result.payload:

            if inst in all_info:

              if all_info[inst].primary_node == name:

                live_data.update(result.payload)

              else:

                wrongnode_inst.add(inst)

            else:

              # orphan instance; we don't list it here as we don't

              # handle this case yet in the output of instance listing

              logging.warning("Orphan instance '%s' found on node %s",

                              inst, name)

        # else no instance is alive

    else:

      live_data = {}

    if query.IQ_DISKUSAGE in self.requested_data:

      gmi = ganeti.masterd.instance

      disk_usage = dict((inst.name,

                         gmi.ComputeDiskSize(inst.disk_template,

                                             [{constants.IDISK_SIZE: disk.size}

                                              for disk in inst.disks]))

                        for inst in instance_list)

    else:

      disk_usage = None

    if query.IQ_CONSOLE in self.requested_data:

      consinfo = {}

      for inst in instance_list:

        if inst.name in live_data:

          # Instance is running

          consinfo[inst.name] = _GetInstanceConsole(cluster, inst)

        else:

          consinfo[inst.name] = None

      assert set(consinfo.keys()) == set(instance_names)

    else:

      consinfo = None

    if query.IQ_NODES in self.requested_data:

      node_names = set(itertools.chain(*map(operator.attrgetter("all_nodes"),

                                            instance_list)))

      nodes = dict(lu.cfg.GetMultiNodeInfo(node_names))

      groups = dict((uuid, lu.cfg.GetNodeGroup(uuid))

                    for uuid in set(map(operator.attrgetter("group"),

                                        nodes.values())))

    else:

      nodes = None

      groups = None

    if query.IQ_NETWORKS in self.requested_data:

      net_uuids = itertools.chain(*(lu.cfg.GetInstanceNetworks(i.name)

                                    for i in instance_list))

      networks = dict((uuid, lu.cfg.GetNetwork(uuid)) for uuid in net_uuids)

    else:

      networks = None

    return query.InstanceQueryData(instance_list, lu.cfg.GetClusterInfo(),

                                   disk_usage, offline_nodes, bad_nodes,

                                   live_data, wrongnode_inst, consinfo,

                                   nodes, groups, networks)

class LUQuery(NoHooksLU):

  """Query for resources/items of a certain kind.

  """

  # pylint: disable=W0142

  REQ_BGL = False

  def CheckArguments(self):

    qcls = _GetQueryImplementation(self.op.what)

    self.impl = qcls(self.op.qfilter, self.op.fields, self.op.use_locking)

  def ExpandNames(self):

    self.impl.ExpandNames(self)

  def DeclareLocks(self, level):

    self.impl.DeclareLocks(self, level)

  def Exec(self, feedback_fn):

    return self.impl.NewStyleQuery(self)

class LUQueryFields(NoHooksLU):

  """Query for resources/items of a certain kind.

  """

  # pylint: disable=W0142

  REQ_BGL = False

  def CheckArguments(self):

    self.qcls = _GetQueryImplementation(self.op.what)

  def ExpandNames(self):

    self.needed_locks = {}

  def Exec(self, feedback_fn):

    return query.QueryFields(self.qcls.FIELDS, self.op.fields)

class LUNodeModifyStorage(NoHooksLU):

  """Logical unit for modifying a storage volume on a node.

  """

  REQ_BGL = False

  def CheckArguments(self):

    self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)

    storage_type = self.op.storage_type

    try:

      modifiable = constants.MODIFIABLE_STORAGE_FIELDS[storage_type]

    except KeyError:

      raise errors.OpPrereqError("Storage units of type '%s' can not be"

                                 " modified" % storage_type,

                                 errors.ECODE_INVAL)

    diff = set(self.op.changes.keys()) - modifiable

    if diff:

      raise errors.OpPrereqError("The following fields can not be modified for"

                                 " storage units of type '%s': %r" %

                                 (storage_type, list(diff)),

                                 errors.ECODE_INVAL)

  def ExpandNames(self):

    self.needed_locks = {

      locking.LEVEL_NODE: self.op.node_name,

      }

  def Exec(self, feedback_fn):

    """Computes the list of nodes and their attributes.

    """

    st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)

    result = self.rpc.call_storage_modify(self.op.node_name,

                                          self.op.storage_type, st_args,

                                          self.op.name, self.op.changes)

    result.Raise("Failed to modify storage unit '%s' on %s" %

                 (self.op.name, self.op.node_name))

class LUNodeAdd(LogicalUnit):

  """Logical unit for adding node to the cluster.

  """

  HPATH = "node-add"

  HTYPE = constants.HTYPE_NODE

  _NFLAGS = ["master_capable", "vm_capable"]

  def CheckArguments(self):

    self.primary_ip_family = self.cfg.GetPrimaryIPFamily()

    # validate/normalize the node name

    self.hostname = netutils.GetHostname(name=self.op.node_name,

                                         family=self.primary_ip_family)

    self.op.node_name = self.hostname.name

    if self.op.readd and self.op.node_name == self.cfg.GetMasterNode():

      raise errors.OpPrereqError("Cannot readd the master node",

                                 errors.ECODE_STATE)

    if self.op.readd and self.op.group:

      raise errors.OpPrereqError("Cannot pass a node group when a node is"

                                 " being readded", errors.ECODE_INVAL)

  def BuildHooksEnv(self):

    """Build hooks env.

    This will run on all nodes before, and on all nodes + the new node after.

    """

    return {

      "OP_TARGET": self.op.node_name,

      "NODE_NAME": self.op.node_name,

      "NODE_PIP": self.op.primary_ip,

      "NODE_SIP": self.op.secondary_ip,

      "MASTER_CAPABLE": str(self.op.master_capable),

      "VM_CAPABLE": str(self.op.vm_capable),

      }

  def BuildHooksNodes(self):

    """Build hooks nodes.

    """

    # Exclude added node

    pre_nodes = list(set(self.cfg.GetNodeList()) - set([self.op.node_name]))

    post_nodes = pre_nodes + [self.op.node_name, ]

    return (pre_nodes, post_nodes)

  def CheckPrereq(self):

    """Check prerequisites.

    This checks:

     - the new node is not already in the config

     - it is resolvable

     - its parameters (single/dual homed) matches the cluster

    Any errors are signaled by raising errors.OpPrereqError.

    """

    cfg = self.cfg

    hostname = self.hostname

    node = hostname.name

    primary_ip = self.op.primary_ip = hostname.ip

    if self.op.secondary_ip is None:

      if self.primary_ip_family == netutils.IP6Address.family:

        raise errors.OpPrereqError("When using a IPv6 primary address, a valid"

                                   " IPv4 address must be given as secondary",

                                   errors.ECODE_INVAL)

      self.op.secondary_ip = primary_ip

    secondary_ip = self.op.secondary_ip

    if not netutils.IP4Address.IsValid(secondary_ip):

      raise errors.OpPrereqError("Secondary IP (%s) needs to be a valid IPv4"

                                 " address" % secondary_ip, errors.ECODE_INVAL)

    node_list = cfg.GetNodeList()

    if not self.op.readd and node in node_list:

      raise errors.OpPrereqError("Node %s is already in the configuration" %

                                 node, errors.ECODE_EXISTS)

    elif self.op.readd and node not in node_list:

      raise errors.OpPrereqError("Node %s is not in the configuration" % node,

                                 errors.ECODE_NOENT)

    self.changed_primary_ip = False

    for existing_node_name, existing_node in cfg.GetMultiNodeInfo(node_list):

      if self.op.readd and node == existing_node_name:

        if existing_node.secondary_ip != secondary_ip:

          raise errors.OpPrereqError("Readded node doesn't have the same IP"

                                     " address configuration as before",

                                     errors.ECODE_INVAL)

        if existing_node.primary_ip != primary_ip:

          self.changed_primary_ip = True

        continue

      if (existing_node.primary_ip == primary_ip or

          existing_node.secondary_ip == primary_ip or

          existing_node.primary_ip == secondary_ip or

          existing_node.secondary_ip == secondary_ip):

        raise errors.OpPrereqError("New node ip address(es) conflict with"

                                   " existing node %s" % existing_node.name,

                                   errors.ECODE_NOTUNIQUE)

    # After this 'if' block, None is no longer a valid value for the

    # _capable op attributes

    if self.op.readd:

      old_node = self.cfg.GetNodeInfo(node)

      assert old_node is not None, "Can't retrieve locked node %s" % node

      for attr in self._NFLAGS:

        if getattr(self.op, attr) is None:

          setattr(self.op, attr, getattr(old_node, attr))

    else:

      for attr in self._NFLAGS:

        if getattr(self.op, attr) is None:

          setattr(self.op, attr, True)

    if self.op.readd and not self.op.vm_capable:

      pri, sec = cfg.GetNodeInstances(node)

      if pri or sec:

        raise errors.OpPrereqError("Node %s being re-added with vm_capable"

                                   " flag set to false, but it already holds"

                                   " instances" % node,

                                   errors.ECODE_STATE)

    # check that the type of the node (single versus dual homed) is the

    # same as for the master

    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())

    master_singlehomed = myself.secondary_ip == myself.primary_ip

    newbie_singlehomed = secondary_ip == primary_ip

    if master_singlehomed != newbie_singlehomed:

      if master_singlehomed:

        raise errors.OpPrereqError("The master has no secondary ip but the"

                                   " new node has one",

                                   errors.ECODE_INVAL)

      else:

        raise errors.OpPrereqError("The master has a secondary ip but the"

                                   " new node doesn't have one",

                                   errors.ECODE_INVAL)

    # checks reachability

    if not netutils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):

      raise errors.OpPrereqError("Node not reachable by ping",

                                 errors.ECODE_ENVIRON)

    if not newbie_singlehomed:

      # check reachability from my secondary ip to newbie's secondary ip

      if not netutils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,

                              source=myself.secondary_ip):

        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"

                                   " based ping to node daemon port",

                                   errors.ECODE_ENVIRON)

    if self.op.readd:

      exceptions = [node]

    else:

      exceptions = []

    if self.op.master_capable:

      self.master_candidate = _DecideSelfPromotion(self, exceptions=exceptions)

    else:

      self.master_candidate = False

    if self.op.readd:

      self.new_node = old_node

    else:

      node_group = cfg.LookupNodeGroup(self.op.group)

      self.new_node = objects.Node(name=node,

                                   primary_ip=primary_ip,

                                   secondary_ip=secondary_ip,

                                   master_candidate=self.master_candidate,

                                   offline=False, drained=False,

                                   group=node_group, ndparams={})

    if self.op.ndparams:

      utils.ForceDictType(self.op.ndparams, constants.NDS_PARAMETER_TYPES)

      _CheckParamsNotGlobal(self.op.ndparams, constants.NDC_GLOBALS, "node",

                            "node", "cluster or group")

    if self.op.hv_state:

      self.new_hv_state = _MergeAndVerifyHvState(self.op.hv_state, None)

    if self.op.disk_state:

      self.new_disk_state = _MergeAndVerifyDiskState(self.op.disk_state, None)

    # TODO: If we need to have multiple DnsOnlyRunner we probably should make

    #       it a property on the base class.

    rpcrunner = rpc.DnsOnlyRunner()

    result = rpcrunner.call_version([node])[node]

    result.Raise("Can't get version information from node %s" % node)

    if constants.PROTOCOL_VERSION == result.payload:

      logging.info("Communication to node %s fine, sw version %s match",

                   node, result.payload)

    else:

      raise errors.OpPrereqError("Version mismatch master version %s,"

                                 " node version %s" %

                                 (constants.PROTOCOL_VERSION, result.payload),

                                 errors.ECODE_ENVIRON)

    vg_name = cfg.GetVGName()

    if vg_name is not None:

      vparams = {constants.NV_PVLIST: [vg_name]}

      excl_stor = _IsExclusiveStorageEnabledNode(cfg, self.new_node)

      cname = self.cfg.GetClusterName()

      result = rpcrunner.call_node_verify_light([node], vparams, cname)[node]

      (errmsgs, _) = _CheckNodePVs(result.payload, excl_stor)

      if errmsgs:

        raise errors.OpPrereqError("Checks on node PVs failed: %s" %

                                   "; ".join(errmsgs), errors.ECODE_ENVIRON)

  def Exec(self, feedback_fn):

    """Adds the new node to the cluster.

    """

    new_node = self.new_node

    node = new_node.name

    assert locking.BGL in self.owned_locks(locking.LEVEL_CLUSTER), \

      "Not owning BGL"

    # We adding a new node so we assume it's powered

    new_node.powered = True

    # for re-adds, reset the offline/drained/master-candidate flags;

    # we need to reset here, otherwise offline would prevent RPC calls

    # later in the procedure; this also means that if the re-add

    # fails, we are left with a non-offlined, broken node

    if self.op.readd:

      new_node.drained = new_node.offline = False # pylint: disable=W0201

      self.LogInfo("Readding a node, the offline/drained flags were reset")

      # if we demote the node, we do cleanup later in the procedure

      new_node.master_candidate = self.master_candidate

      if self.changed_primary_ip:

        new_node.primary_ip = self.op.primary_ip

    # copy the master/vm_capable flags

    for attr in self._NFLAGS:

      setattr(new_node, attr, getattr(self.op, attr))

    # notify the user about any possible mc promotion

    if new_node.master_candidate:

      self.LogInfo("Node will be a master candidate")

    if self.op.ndparams:

      new_node.ndparams = self.op.ndparams

    else:

      new_node.ndparams = {}

    if self.op.hv_state:

      new_node.hv_state_static = self.new_hv_state

    if self.op.disk_state:

      new_node.disk_state_static = self.new_disk_state

    # Add node to our /etc/hosts, and add key to known_hosts

    if self.cfg.GetClusterInfo().modify_etc_hosts:

      master_node = self.cfg.GetMasterNode()

      result = self.rpc.call_etc_hosts_modify(master_node,

                                              constants.ETC_HOSTS_ADD,

                                              self.hostname.name,

                                              self.hostname.ip)

      result.Raise("Can't update hosts file with new host data")

    if new_node.secondary_ip != new_node.primary_ip:

      _CheckNodeHasSecondaryIP(self, new_node.name, new_node.secondary_ip,

                               False)

    node_verify_list = [self.cfg.GetMasterNode()]

    node_verify_param = {

      constants.NV_NODELIST: ([node], {}),

      # TODO: do a node-net-test as well?

    }

    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,

                                       self.cfg.GetClusterName())

    for verifier in node_verify_list:

      result[verifier].Raise("Cannot communicate with node %s" % verifier)

      nl_payload = result[verifier].payload[constants.NV_NODELIST]

      if nl_payload:

        for failed in nl_payload:

          feedback_fn("ssh/hostname verification failed"

                      " (checking from %s): %s" %

                      (verifier, nl_payload[failed]))

        raise errors.OpExecError("ssh/hostname verification failed")

    if self.op.readd:

      _RedistributeAncillaryFiles(self)

      self.context.ReaddNode(new_node)

      # make sure we redistribute the config

      self.cfg.Update(new_node, feedback_fn)

      # and make sure the new node will not have old files around

      if not new_node.master_candidate:

        result = self.rpc.call_node_demote_from_mc(new_node.name)

        msg = result.fail_msg

        if msg:

          self.LogWarning("Node failed to demote itself from master"

                          " candidate status: %s" % msg)

    else:

      _RedistributeAncillaryFiles(self, additional_nodes=[node],

                                  additional_vm=self.op.vm_capable)

      self.context.AddNode(new_node, self.proc.GetECId())

class LUNodeSetParams(LogicalUnit):

  """Modifies the parameters of a node.

  @cvar _F2R: a dictionary from tuples of flags (mc, drained, offline)

      to the node role (as _ROLE_*)

  @cvar _R2F: a dictionary from node role to tuples of flags

  @cvar _FLAGS: a list of attribute names corresponding to the flags

  """

  HPATH = "node-modify"

  HTYPE = constants.HTYPE_NODE

  REQ_BGL = False

  (_ROLE_CANDIDATE, _ROLE_DRAINED, _ROLE_OFFLINE, _ROLE_REGULAR) = range(4)

  _F2R = {

    (True, False, False): _ROLE_CANDIDATE,

    (False, True, False): _ROLE_DRAINED,

    (False, False, True): _ROLE_OFFLINE,

    (False, False, False): _ROLE_REGULAR,

    }

  _R2F = dict((v, k) for k, v in _F2R.items())

  _FLAGS = ["master_candidate", "drained", "offline"]

  def CheckArguments(self):

    self.op.node_name = _ExpandNodeName(self.cfg, self.op.node_name)

    all_mods = [self.op.offline, self.op.master_candidate, self.op.drained,

                self.op.master_capable, self.op.vm_capable,

                self.op.secondary_ip, self.op.ndparams, self.op.hv_state,

                self.op.disk_state]

    if all_mods.count(None) == len(all_mods):

      raise errors.OpPrereqError("Please pass at least one modification",

                                 errors.ECODE_INVAL)

    if all_mods.count(True) > 1:

      raise errors.OpPrereqError("Can't set the node into more than one"

                                 " state at the same time",

                                 errors.ECODE_INVAL)

    # Boolean value that tells us whether we might be demoting from MC

    self.might_demote = (self.op.master_candidate is False or

                         self.op.offline is True or

                         self.op.drained is True or

                         self.op.master_capable is False)

    if self.op.secondary_ip:

      if not netutils.IP4Address.IsValid(self.op.secondary_ip):

        raise errors.OpPrereqError("Secondary IP (%s) needs to be a valid IPv4"

                                   " address" % self.op.secondary_ip,

                                   errors.ECODE_INVAL)

    self.lock_all = self.op.auto_promote and self.might_demote

    self.lock_instances = self.op.secondary_ip is not None

  def _InstanceFilter(self, instance):

    """Filter for getting affected instances.

    """

    return (instance.disk_template in constants.DTS_INT_MIRROR and

            self.op.node_name in instance.all_nodes)

  def ExpandNames(self):

    if self.lock_all:

      self.needed_locks = {

        locking.LEVEL_NODE: locking.ALL_SET,

        # Block allocations when all nodes are locked

        locking.LEVEL_NODE_ALLOC: locking.ALL_SET,

        }

    else:

      self.needed_locks = {

        locking.LEVEL_NODE: self.op.node_name,

        }

    # Since modifying a node can have severe effects on currently running

    # operations the resource lock is at least acquired in shared mode

    self.needed_locks[locking.LEVEL_NODE_RES] = \

      self.needed_locks[locking.LEVEL_NODE]

    # Get all locks except nodes in shared mode; they are not used for anything

    # but read-only access

    self.share_locks = _ShareAll()

    self.share_locks[locking.LEVEL_NODE] = 0

    self.share_locks[locking.LEVEL_NODE_RES] = 0

    self.share_locks[locking.LEVEL_NODE_ALLOC] = 0

    if self.lock_instances:

      self.needed_locks[locking.LEVEL_INSTANCE] = \

        frozenset(self.cfg.GetInstancesInfoByFilter(self._InstanceFilter))

  def BuildHooksEnv(self):

    """Build hooks env.

    This runs on the master node.

    """

    return {

      "OP_TARGET": self.op.node_name,

      "MASTER_CANDIDATE": str(self.op.master_candidate),

      "OFFLINE": str(self.op.offline),

      "DRAINED": str(self.op.drained),

      "MASTER_CAPABLE": str(self.op.master_capable),

      "VM_CAPABLE": str(self.op.vm_capable),

      }

  def BuildHooksNodes(self):

    """Build hooks nodes.

    """

    nl = [self.cfg.GetMasterNode(), self.op.node_name]

    return (nl, nl)

  def CheckPrereq(self):

    """Check prerequisites.

    This only checks the instance list against the existing names.

    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)

    if self.lock_instances:

      affected_instances = \

        self.cfg.GetInstancesInfoByFilter(self._InstanceFilter)

      # Verify instance locks

      owned_instances = self.owned_locks(locking.LEVEL_INSTANCE)

      wanted_instances = frozenset(affected_instances.keys())

      if wanted_instances - owned_instances:

        raise errors.OpPrereqError("Instances affected by changing node %s's"

                                   " secondary IP address have changed since"

                                   " locks were acquired, wanted '%s', have"

                                   " '%s'; retry the operation" %

                                   (self.op.node_name,

                                    utils.CommaJoin(wanted_instances),

                                    utils.CommaJoin(owned_instances)),

                                   errors.ECODE_STATE)

    else:

      affected_instances = None

    if (self.op.master_candidate is not None or

        self.op.drained is not None or

        self.op.offline is not None):

      # we can't change the master's node flags

      if self.op.node_name == self.cfg.GetMasterNode():

        raise errors.OpPrereqError("The master role can be changed"

                                   " only via master-failover",

                                   errors.ECODE_INVAL)

    if self.op.master_candidate and not node.master_capable:

      raise errors.OpPrereqError("Node %s is not master capable, cannot make"

                                 " it a master candidate" % node.name,

                                 errors.ECODE_STATE)

    if self.op.vm_capable is False:

      (ipri, isec) = self.cfg.GetNodeInstances(self.op.node_name)

      if ipri or isec:

        raise errors.OpPrereqError("Node %s hosts instances, cannot unset"

                                   " the vm_capable flag" % node.name,

                                   errors.ECODE_STATE)

    if node.master_candidate and self.might_demote and not self.lock_all:

      assert not self.op.auto_promote, "auto_promote set but lock_all not"

      # check if after removing the current node, we're missing master

      # candidates

      (mc_remaining, mc_should, _) = \

          self.cfg.GetMasterCandidateStats(exceptions=[node.name])

      if mc_remaining < mc_should:

        raise errors.OpPrereqError("Not enough master candidates, please"

                                   " pass auto promote option to allow"

                                   " promotion (--auto-promote or RAPI"

                                   " auto_promote=True)", errors.ECODE_STATE)

    self.old_flags = old_flags = (node.master_candidate,

                                  node.drained, node.offline)

    assert old_flags in self._F2R, "Un-handled old flags %s" % str(old_flags)

    self.old_role = old_role = self._F2R[old_flags]

    # Check for ineffective changes

    for attr in self._FLAGS:

      if (getattr(self.op, attr) is False and getattr(node, attr) is False):

        self.LogInfo("Ignoring request to unset flag %s, already unset", attr)

        setattr(self.op, attr, None)

    # Past this point, any flag change to False means a transition

    # away from the respective state, as only real changes are kept

    # TODO: We might query the real power state if it supports OOB

    if _SupportsOob(self.cfg, node):

      if self.op.offline is False and not (node.powered or

                                           self.op.powered is True):

        raise errors.OpPrereqError(("Node %s needs to be turned on before its"

                                    " offline status can be reset") %

                                   self.op.node_name, errors.ECODE_STATE)

    elif self.op.powered is not None:

      raise errors.OpPrereqError(("Unable to change powered state for node %s"

                                  " as it does not support out-of-band"

                                  " handling") % self.op.node_name,

                                 errors.ECODE_STATE)

    # If we're being deofflined/drained, we'll MC ourself if needed

    if (self.op.drained is False or self.op.offline is False or

        (self.op.master_capable and not node.master_capable)):

      if _DecideSelfPromotion(self):

        self.op.master_candidate = True

        self.LogInfo("Auto-promoting node to master candidate")

    # If we're no longer master capable, we'll demote ourselves from MC

    if self.op.master_capable is False and node.master_candidate:

      self.LogInfo("Demoting from master candidate")

      self.op.master_candidate = False

    # Compute new role

    assert [getattr(self.op, attr) for attr in self._FLAGS].count(True) <= 1

    if self.op.master_candidate:

      new_role = self._ROLE_CANDIDATE

    elif self.op.drained:

      new_role = self._ROLE_DRAINED

    elif self.op.offline:

      new_role = self._ROLE_OFFLINE

    elif False in [self.op.master_candidate, self.op.drained, self.op.offline]:

      # False is still in new flags, which means we're un-setting (the

      # only) True flag

      new_role = self._ROLE_REGULAR

    else: # no new flags, nothing, keep old role

      new_role = old_role

    self.new_role = new_role

    if old_role == self._ROLE_OFFLINE and new_role != old_role:

      # Trying to transition out of offline status

      result = self.rpc.call_version([node.name])[node.name]

      if result.fail_msg:

        raise errors.OpPrereqError("Node %s is being de-offlined but fails"

                                   " to report its version: %s" %

                                   (node.name, result.fail_msg),

                                   errors.ECODE_STATE)

      else:

        self.LogWarning("Transitioning node from offline to online state"

                        " without using re-add. Please make sure the node"

                        " is healthy!")

    # When changing the secondary ip, verify if this is a single-homed to

    # multi-homed transition or vice versa, and apply the relevant

    # restrictions.

    if self.op.secondary_ip:

      # Ok even without locking, because this can't be changed by any LU

      master = self.cfg.GetNodeInfo(self.cfg.GetMasterNode())

      master_singlehomed = master.secondary_ip == master.primary_ip

      if master_singlehomed and self.op.secondary_ip != node.primary_ip:

        if self.op.force and node.name == master.name:

          self.LogWarning("Transitioning from single-homed to multi-homed"

                          " cluster; all nodes will require a secondary IP"

                          " address")

        else:

          raise errors.OpPrereqError("Changing the secondary ip on a"

                                     " single-homed cluster requires the"

                                     " --force option to be passed, and the"

                                     " target node to be the master",

                                     errors.ECODE_INVAL)

      elif not master_singlehomed and self.op.secondary_ip == node.primary_ip:

        if self.op.force and node.name == master.name:

          self.LogWarning("Transitioning from multi-homed to single-homed"

                          " cluster; secondary IP addresses will have to be"

                          " removed")