Synnefo » snf-ganeti

#

#

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

"""Cluster related commands"""

# pylint: disable=W0401,W0613,W0614,C0103

# W0401: Wildcard import ganeti.cli

# W0613: Unused argument, since all functions follow the same API

# W0614: Unused import %s from wildcard import (since we need cli)

# C0103: Invalid name gnt-cluster

from cStringIO import StringIO

import os

import time

import OpenSSL

import itertools

from ganeti.cli import *

from ganeti import bootstrap

from ganeti import compat

from ganeti import constants

from ganeti import errors

from ganeti import netutils

from ganeti import objects

from ganeti import opcodes

from ganeti import pathutils

from ganeti import qlang

from ganeti import serializer

from ganeti import ssconf

from ganeti import ssh

from ganeti import uidpool

from ganeti import utils

from ganeti.client import base

ON_OPT = cli_option("--on", default=False,

                    action="store_true", dest="on",

                    help="Recover from an EPO")

GROUPS_OPT = cli_option("--groups", default=False,

                        action="store_true", dest="groups",

                        help="Arguments are node groups instead of nodes")

FORCE_FAILOVER = cli_option("--yes-do-it", dest="yes_do_it",

                            help="Override interactive check for --no-voting",

                            default=False, action="store_true")

FORCE_DISTRIBUTION = cli_option("--yes-do-it", dest="yes_do_it",

                                help="Unconditionally distribute the"

                                " configuration, even if the queue"

                                " is drained",

                                default=False, action="store_true")

TO_OPT = cli_option("--to", default=None, type="string",

                    help="The Ganeti version to upgrade to")

RESUME_OPT = cli_option("--resume", default=False, action="store_true",

                        help="Resume any pending Ganeti upgrades")

_EPO_PING_INTERVAL = 30 # 30 seconds between pings

_EPO_PING_TIMEOUT = 1 # 1 second

_EPO_REACHABLE_TIMEOUT = 15 * 60 # 15 minutes

def _InitEnabledDiskTemplates(opts):

  """Initialize the list of enabled disk templates.

  """

  if opts.enabled_disk_templates:

    return opts.enabled_disk_templates.split(",")

  else:

    return constants.DEFAULT_ENABLED_DISK_TEMPLATES

def _InitVgName(opts, enabled_disk_templates):

  """Initialize the volume group name.

  @type enabled_disk_templates: list of strings

  @param enabled_disk_templates: cluster-wide enabled disk templates

  """

  vg_name = None

  if opts.vg_name is not None:

    vg_name = opts.vg_name

    if vg_name:

      if not utils.IsLvmEnabled(enabled_disk_templates):

        ToStdout("You specified a volume group with --vg-name, but you did not"

                 " enable any disk template that uses lvm.")

    elif utils.IsLvmEnabled(enabled_disk_templates):

      raise errors.OpPrereqError(

          "LVM disk templates are enabled, but vg name not set.")

  elif utils.IsLvmEnabled(enabled_disk_templates):

    vg_name = constants.DEFAULT_VG

  return vg_name

def _InitDrbdHelper(opts, enabled_disk_templates):

  """Initialize the DRBD usermode helper.

  """

  drbd_enabled = constants.DT_DRBD8 in enabled_disk_templates

  if not drbd_enabled and opts.drbd_helper is not None:

    ToStdout("Note: You specified a DRBD usermode helper, while DRBD storage"

             " is not enabled.")

  if drbd_enabled:

    if opts.drbd_helper is None:

      return constants.DEFAULT_DRBD_HELPER

    if opts.drbd_helper == '':

      raise errors.OpPrereqError(

          "Unsetting the drbd usermode helper while enabling DRBD is not"

          " allowed.")

  return opts.drbd_helper

@UsesRPC

def InitCluster(opts, args):

  """Initialize the cluster.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should contain only one element, the desired

      cluster name

  @rtype: int

  @return: the desired exit code

  """

  enabled_disk_templates = _InitEnabledDiskTemplates(opts)

  try:

    vg_name = _InitVgName(opts, enabled_disk_templates)

    drbd_helper = _InitDrbdHelper(opts, enabled_disk_templates)

  except errors.OpPrereqError, e:

    ToStderr(str(e))

    return 1

  master_netdev = opts.master_netdev

  if master_netdev is None:

    nic_mode = opts.nicparams.get(constants.NIC_MODE, None)

    if not nic_mode:

      # default case, use bridging

      master_netdev = constants.DEFAULT_BRIDGE

    elif nic_mode == constants.NIC_MODE_OVS:

      # default ovs is different from default bridge

      master_netdev = constants.DEFAULT_OVS

      opts.nicparams[constants.NIC_LINK] = constants.DEFAULT_OVS

  hvlist = opts.enabled_hypervisors

  if hvlist is None:

    hvlist = constants.DEFAULT_ENABLED_HYPERVISOR

  hvlist = hvlist.split(",")

  hvparams = dict(opts.hvparams)

  beparams = opts.beparams

  nicparams = opts.nicparams

  diskparams = dict(opts.diskparams)

  # check the disk template types here, as we cannot rely on the type check done

  # by the opcode parameter types

  diskparams_keys = set(diskparams.keys())

  if not (diskparams_keys <= constants.DISK_TEMPLATES):

    unknown = utils.NiceSort(diskparams_keys - constants.DISK_TEMPLATES)

    ToStderr("Disk templates unknown: %s" % utils.CommaJoin(unknown))

    return 1

  # prepare beparams dict

  beparams = objects.FillDict(constants.BEC_DEFAULTS, beparams)

  utils.ForceDictType(beparams, constants.BES_PARAMETER_COMPAT)

  # prepare nicparams dict

  nicparams = objects.FillDict(constants.NICC_DEFAULTS, nicparams)

  utils.ForceDictType(nicparams, constants.NICS_PARAMETER_TYPES)

  # prepare ndparams dict

  if opts.ndparams is None:

    ndparams = dict(constants.NDC_DEFAULTS)

  else:

    ndparams = objects.FillDict(constants.NDC_DEFAULTS, opts.ndparams)

    utils.ForceDictType(ndparams, constants.NDS_PARAMETER_TYPES)

  # prepare hvparams dict

  for hv in constants.HYPER_TYPES:

    if hv not in hvparams:

      hvparams[hv] = {}

    hvparams[hv] = objects.FillDict(constants.HVC_DEFAULTS[hv], hvparams[hv])

    utils.ForceDictType(hvparams[hv], constants.HVS_PARAMETER_TYPES)

  # prepare diskparams dict

  for templ in constants.DISK_TEMPLATES:

    if templ not in diskparams:

      diskparams[templ] = {}

    diskparams[templ] = objects.FillDict(constants.DISK_DT_DEFAULTS[templ],

                                         diskparams[templ])

    utils.ForceDictType(diskparams[templ], constants.DISK_DT_TYPES)

  # prepare ipolicy dict

  ipolicy = CreateIPolicyFromOpts(

    ispecs_mem_size=opts.ispecs_mem_size,

    ispecs_cpu_count=opts.ispecs_cpu_count,

    ispecs_disk_count=opts.ispecs_disk_count,

    ispecs_disk_size=opts.ispecs_disk_size,

    ispecs_nic_count=opts.ispecs_nic_count,

    minmax_ispecs=opts.ipolicy_bounds_specs,

    std_ispecs=opts.ipolicy_std_specs,

    ipolicy_disk_templates=opts.ipolicy_disk_templates,

    ipolicy_vcpu_ratio=opts.ipolicy_vcpu_ratio,

    ipolicy_spindle_ratio=opts.ipolicy_spindle_ratio,

    fill_all=True)

  if opts.candidate_pool_size is None:

    opts.candidate_pool_size = constants.MASTER_POOL_SIZE_DEFAULT

  if opts.mac_prefix is None:

    opts.mac_prefix = constants.DEFAULT_MAC_PREFIX

  uid_pool = opts.uid_pool

  if uid_pool is not None:

    uid_pool = uidpool.ParseUidPool(uid_pool)

  if opts.prealloc_wipe_disks is None:

    opts.prealloc_wipe_disks = False

  external_ip_setup_script = opts.use_external_mip_script

  if external_ip_setup_script is None:

    external_ip_setup_script = False

  try:

    primary_ip_version = int(opts.primary_ip_version)

  except (ValueError, TypeError), err:

    ToStderr("Invalid primary ip version value: %s" % str(err))

    return 1

  master_netmask = opts.master_netmask

  try:

    if master_netmask is not None:

      master_netmask = int(master_netmask)

  except (ValueError, TypeError), err:

    ToStderr("Invalid master netmask value: %s" % str(err))

    return 1

  if opts.disk_state:

    disk_state = utils.FlatToDict(opts.disk_state)

  else:

    disk_state = {}

  hv_state = dict(opts.hv_state)

  default_ialloc_params = opts.default_iallocator_params

  bootstrap.InitCluster(cluster_name=args[0],

                        secondary_ip=opts.secondary_ip,

                        vg_name=vg_name,

                        mac_prefix=opts.mac_prefix,

                        master_netmask=master_netmask,

                        master_netdev=master_netdev,

                        file_storage_dir=opts.file_storage_dir,

                        shared_file_storage_dir=opts.shared_file_storage_dir,

                        gluster_storage_dir=opts.gluster_storage_dir,

                        enabled_hypervisors=hvlist,

                        hvparams=hvparams,

                        beparams=beparams,

                        nicparams=nicparams,

                        ndparams=ndparams,

                        diskparams=diskparams,

                        ipolicy=ipolicy,

                        candidate_pool_size=opts.candidate_pool_size,

                        modify_etc_hosts=opts.modify_etc_hosts,

                        modify_ssh_setup=opts.modify_ssh_setup,

                        maintain_node_health=opts.maintain_node_health,

                        drbd_helper=drbd_helper,

                        uid_pool=uid_pool,

                        default_iallocator=opts.default_iallocator,

                        default_iallocator_params=default_ialloc_params,

                        primary_ip_version=primary_ip_version,

                        prealloc_wipe_disks=opts.prealloc_wipe_disks,

                        use_external_mip_script=external_ip_setup_script,

                        hv_state=hv_state,

                        disk_state=disk_state,

                        enabled_disk_templates=enabled_disk_templates,

                        )

  op = opcodes.OpClusterPostInit()

  SubmitOpCode(op, opts=opts)

  return 0

@UsesRPC

def DestroyCluster(opts, args):

  """Destroy the cluster.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should be an empty list

  @rtype: int

  @return: the desired exit code

  """

  if not opts.yes_do_it:

    ToStderr("Destroying a cluster is irreversible. If you really want"

             " destroy this cluster, supply the --yes-do-it option.")

    return 1

  op = opcodes.OpClusterDestroy()

  master_uuid = SubmitOpCode(op, opts=opts)

  # if we reached this, the opcode didn't fail; we can proceed to

  # shutdown all the daemons

  bootstrap.FinalizeClusterDestroy(master_uuid)

  return 0

def RenameCluster(opts, args):

  """Rename the cluster.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should contain only one element, the new cluster name

  @rtype: int

  @return: the desired exit code

  """

  cl = GetClient()

  (cluster_name, ) = cl.QueryConfigValues(["cluster_name"])

  new_name = args[0]

  if not opts.force:

    usertext = ("This will rename the cluster from '%s' to '%s'. If you are"

                " connected over the network to the cluster name, the"

                " operation is very dangerous as the IP address will be"

                " removed from the node and the change may not go through."

                " Continue?") % (cluster_name, new_name)

    if not AskUser(usertext):

      return 1

  op = opcodes.OpClusterRename(name=new_name)

  result = SubmitOpCode(op, opts=opts, cl=cl)

  if result:

    ToStdout("Cluster renamed from '%s' to '%s'", cluster_name, result)

  return 0

def ActivateMasterIp(opts, args):

  """Activates the master IP.

  """

  op = opcodes.OpClusterActivateMasterIp()

  SubmitOpCode(op)

  return 0

def DeactivateMasterIp(opts, args):

  """Deactivates the master IP.

  """

  if not opts.confirm:

    usertext = ("This will disable the master IP. All the open connections to"

                " the master IP will be closed. To reach the master you will"

                " need to use its node IP."

                " Continue?")

    if not AskUser(usertext):

      return 1

  op = opcodes.OpClusterDeactivateMasterIp()

  SubmitOpCode(op)

  return 0

def RedistributeConfig(opts, args):

  """Forces push of the cluster configuration.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: empty list

  @rtype: int

  @return: the desired exit code

  """

  op = opcodes.OpClusterRedistConf()

  if opts.yes_do_it:

    SubmitOpCodeToDrainedQueue(op)

  else:

    SubmitOrSend(op, opts)

  return 0

def ShowClusterVersion(opts, args):

  """Write version of ganeti software to the standard output.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should be an empty list

  @rtype: int

  @return: the desired exit code

  """

  cl = GetClient()

  result = cl.QueryClusterInfo()

  ToStdout("Software version: %s", result["software_version"])

  ToStdout("Internode protocol: %s", result["protocol_version"])

  ToStdout("Configuration format: %s", result["config_version"])

  ToStdout("OS api version: %s", result["os_api_version"])

  ToStdout("Export interface: %s", result["export_version"])

  ToStdout("VCS version: %s", result["vcs_version"])

  return 0

def ShowClusterMaster(opts, args):

  """Write name of master node to the standard output.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should be an empty list

  @rtype: int

  @return: the desired exit code

  """

  master = bootstrap.GetMaster()

  ToStdout(master)

  return 0

def _FormatGroupedParams(paramsdict, roman=False):

  """Format Grouped parameters (be, nic, disk) by group.

  @type paramsdict: dict of dicts

  @param paramsdict: {group: {param: value, ...}, ...}

  @rtype: dict of dicts

  @return: copy of the input dictionaries with strings as values

  """

  ret = {}

  for (item, val) in paramsdict.items():

    if isinstance(val, dict):

      ret[item] = _FormatGroupedParams(val, roman=roman)

    elif roman and isinstance(val, int):

      ret[item] = compat.TryToRoman(val)

    else:

      ret[item] = str(val)

  return ret

def ShowClusterConfig(opts, args):

  """Shows cluster information.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should be an empty list

  @rtype: int

  @return: the desired exit code

  """

  cl = GetClient()

  result = cl.QueryClusterInfo()

  if result["tags"]:

    tags = utils.CommaJoin(utils.NiceSort(result["tags"]))

  else:

    tags = "(none)"

  if result["reserved_lvs"]:

    reserved_lvs = utils.CommaJoin(result["reserved_lvs"])

  else:

    reserved_lvs = "(none)"

  enabled_hv = result["enabled_hypervisors"]

  hvparams = dict((k, v) for k, v in result["hvparams"].iteritems()

                  if k in enabled_hv)

  info = [

    ("Cluster name", result["name"]),

    ("Cluster UUID", result["uuid"]),

    ("Creation time", utils.FormatTime(result["ctime"])),

    ("Modification time", utils.FormatTime(result["mtime"])),

    ("Master node", result["master"]),

    ("Architecture (this node)",

     "%s (%s)" % (result["architecture"][0], result["architecture"][1])),

    ("Tags", tags),

    ("Default hypervisor", result["default_hypervisor"]),

    ("Enabled hypervisors", utils.CommaJoin(enabled_hv)),

    ("Hypervisor parameters", _FormatGroupedParams(hvparams)),

    ("OS-specific hypervisor parameters",

     _FormatGroupedParams(result["os_hvp"])),

    ("OS parameters", _FormatGroupedParams(result["osparams"])),

    ("Hidden OSes", utils.CommaJoin(result["hidden_os"])),

    ("Blacklisted OSes", utils.CommaJoin(result["blacklisted_os"])),

    ("Cluster parameters", [

      ("candidate pool size",

       compat.TryToRoman(result["candidate_pool_size"],

                         convert=opts.roman_integers)),

      ("maximal number of jobs running simultaneously",

       compat.TryToRoman(result["max_running_jobs"],

                         convert=opts.roman_integers)),

      ("mac prefix", result["mac_prefix"]),

      ("master netdev", result["master_netdev"]),

      ("master netmask", result["master_netmask"]),

      ("use external master IP address setup script",

       result["use_external_mip_script"]),

      ("lvm volume group", result["volume_group_name"]),

      ("lvm reserved volumes", reserved_lvs),

      ("drbd usermode helper", result["drbd_usermode_helper"]),

      ("file storage path", result["file_storage_dir"]),

      ("shared file storage path", result["shared_file_storage_dir"]),

      ("gluster storage path", result["gluster_storage_dir"]),

      ("maintenance of node health", result["maintain_node_health"]),

      ("uid pool", uidpool.FormatUidPool(result["uid_pool"])),

      ("default instance allocator", result["default_iallocator"]),

      ("default instance allocator parameters",

       result["default_iallocator_params"]),

      ("primary ip version", result["primary_ip_version"]),

      ("preallocation wipe disks", result["prealloc_wipe_disks"]),

      ("OS search path", utils.CommaJoin(pathutils.OS_SEARCH_PATH)),

      ("ExtStorage Providers search path",

       utils.CommaJoin(pathutils.ES_SEARCH_PATH)),

      ("enabled disk templates",

       utils.CommaJoin(result["enabled_disk_templates"])),

      ("instance communication network",

       result["instance_communication_network"]),

      ]),

    ("Default node parameters",

     _FormatGroupedParams(result["ndparams"], roman=opts.roman_integers)),

    ("Default instance parameters",

     _FormatGroupedParams(result["beparams"], roman=opts.roman_integers)),

    ("Default nic parameters",

     _FormatGroupedParams(result["nicparams"], roman=opts.roman_integers)),

    ("Default disk parameters",

     _FormatGroupedParams(result["diskparams"], roman=opts.roman_integers)),

    ("Instance policy - limits for instances",

     FormatPolicyInfo(result["ipolicy"], None, True)),

    ]

  PrintGenericInfo(info)

  return 0

def ClusterCopyFile(opts, args):

  """Copy a file from master to some nodes.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should contain only one element, the path of

      the file to be copied

  @rtype: int

  @return: the desired exit code

  """

  filename = args[0]

  if not os.path.exists(filename):

    raise errors.OpPrereqError("No such filename '%s'" % filename,

                               errors.ECODE_INVAL)

  cl = GetClient()

  qcl = GetClient()

  try:

    cluster_name = cl.QueryConfigValues(["cluster_name"])[0]

    results = GetOnlineNodes(nodes=opts.nodes, cl=qcl, filter_master=True,

                             secondary_ips=opts.use_replication_network,

                             nodegroup=opts.nodegroup)

    ports = GetNodesSshPorts(opts.nodes, qcl)

  finally:

    cl.Close()

    qcl.Close()

  srun = ssh.SshRunner(cluster_name)

  for (node, port) in zip(results, ports):

    if not srun.CopyFileToNode(node, port, filename):

      ToStderr("Copy of file %s to node %s:%d failed", filename, node, port)

  return 0

def RunClusterCommand(opts, args):

  """Run a command on some nodes.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should contain the command to be run and its arguments

  @rtype: int

  @return: the desired exit code

  """

  cl = GetClient()

  qcl = GetClient()

  command = " ".join(args)

  nodes = GetOnlineNodes(nodes=opts.nodes, cl=qcl, nodegroup=opts.nodegroup)

  ports = GetNodesSshPorts(nodes, qcl)

  cluster_name, master_node = cl.QueryConfigValues(["cluster_name",

                                                    "master_node"])

  srun = ssh.SshRunner(cluster_name=cluster_name)

  # Make sure master node is at list end

  if master_node in nodes:

    nodes.remove(master_node)

    nodes.append(master_node)

  for (name, port) in zip(nodes, ports):

    result = srun.Run(name, constants.SSH_LOGIN_USER, command, port=port)

    if opts.failure_only and result.exit_code == constants.EXIT_SUCCESS:

      # Do not output anything for successful commands

      continue

    ToStdout("------------------------------------------------")

    if opts.show_machine_names:

      for line in result.output.splitlines():

        ToStdout("%s: %s", name, line)

    else:

      ToStdout("node: %s", name)

      ToStdout("%s", result.output)

    ToStdout("return code = %s", result.exit_code)

  return 0

def VerifyCluster(opts, args):

  """Verify integrity of cluster, performing various test on nodes.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should be an empty list

  @rtype: int

  @return: the desired exit code

  """

  skip_checks = []

  if opts.skip_nplusone_mem:

    skip_checks.append(constants.VERIFY_NPLUSONE_MEM)

  cl = GetClient()

  op = opcodes.OpClusterVerify(verbose=opts.verbose,

                               error_codes=opts.error_codes,

                               debug_simulate_errors=opts.simulate_errors,

                               skip_checks=skip_checks,

                               ignore_errors=opts.ignore_errors,

                               group_name=opts.nodegroup)

  result = SubmitOpCode(op, cl=cl, opts=opts)

  # Keep track of submitted jobs

  jex = JobExecutor(cl=cl, opts=opts)

  for (status, job_id) in result[constants.JOB_IDS_KEY]:

    jex.AddJobId(None, status, job_id)

  results = jex.GetResults()

  (bad_jobs, bad_results) = \

    map(len,

        # Convert iterators to lists

        map(list,

            # Count errors

            map(compat.partial(itertools.ifilterfalse, bool),

                # Convert result to booleans in a tuple

                zip(*((job_success, len(op_results) == 1 and op_results[0])

                      for (job_success, op_results) in results)))))

  if bad_jobs == 0 and bad_results == 0:

    rcode = constants.EXIT_SUCCESS

  else:

    rcode = constants.EXIT_FAILURE

    if bad_jobs > 0:

      ToStdout("%s job(s) failed while verifying the cluster.", bad_jobs)

  return rcode

def VerifyDisks(opts, args):

  """Verify integrity of cluster disks.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should be an empty list

  @rtype: int

  @return: the desired exit code

  """

  cl = GetClient()

  op = opcodes.OpClusterVerifyDisks()

  result = SubmitOpCode(op, cl=cl, opts=opts)

  # Keep track of submitted jobs

  jex = JobExecutor(cl=cl, opts=opts)

  for (status, job_id) in result[constants.JOB_IDS_KEY]:

    jex.AddJobId(None, status, job_id)

  retcode = constants.EXIT_SUCCESS

  for (status, result) in jex.GetResults():

    if not status:

      ToStdout("Job failed: %s", result)

      continue

    ((bad_nodes, instances, missing), ) = result

    for node, text in bad_nodes.items():

      ToStdout("Error gathering data on node %s: %s",

               node, utils.SafeEncode(text[-400:]))

      retcode = constants.EXIT_FAILURE

      ToStdout("You need to fix these nodes first before fixing instances")

    for iname in instances:

      if iname in missing:

        continue

      op = opcodes.OpInstanceActivateDisks(instance_name=iname)

      try:

        ToStdout("Activating disks for instance '%s'", iname)

        SubmitOpCode(op, opts=opts, cl=cl)

      except errors.GenericError, err:

        nret, msg = FormatError(err)

        retcode |= nret

        ToStderr("Error activating disks for instance %s: %s", iname, msg)

    if missing:

      for iname, ival in missing.iteritems():

        all_missing = compat.all(x[0] in bad_nodes for x in ival)

        if all_missing:

          ToStdout("Instance %s cannot be verified as it lives on"

                   " broken nodes", iname)

        else:

          ToStdout("Instance %s has missing logical volumes:", iname)

          ival.sort()

          for node, vol in ival:

            if node in bad_nodes:

              ToStdout("\tbroken node %s /dev/%s", node, vol)

            else:

              ToStdout("\t%s /dev/%s", node, vol)

      ToStdout("You need to replace or recreate disks for all the above"

               " instances if this message persists after fixing broken nodes.")

      retcode = constants.EXIT_FAILURE

    elif not instances:

      ToStdout("No disks need to be activated.")

  return retcode

def RepairDiskSizes(opts, args):

  """Verify sizes of cluster disks.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: optional list of instances to restrict check to

  @rtype: int

  @return: the desired exit code

  """

  op = opcodes.OpClusterRepairDiskSizes(instances=args)

  SubmitOpCode(op, opts=opts)

@UsesRPC

def MasterFailover(opts, args):

  """Failover the master node.

  This command, when run on a non-master node, will cause the current

  master to cease being master, and the non-master to become new

  master.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should be an empty list

  @rtype: int

  @return: the desired exit code

  """

  if opts.no_voting and not opts.yes_do_it:

    usertext = ("This will perform the failover even if most other nodes"

                " are down, or if this node is outdated. This is dangerous"

                " as it can lead to a non-consistent cluster. Check the"

                " gnt-cluster(8) man page before proceeding. Continue?")

    if not AskUser(usertext):

      return 1

  return bootstrap.MasterFailover(no_voting=opts.no_voting)

def MasterPing(opts, args):

  """Checks if the master is alive.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should be an empty list

  @rtype: int

  @return: the desired exit code

  """

  try:

    cl = GetClient()

    cl.QueryClusterInfo()

    return 0

  except Exception: # pylint: disable=W0703

    return 1

def SearchTags(opts, args):

  """Searches the tags on all the cluster.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should contain only one element, the tag pattern

  @rtype: int

  @return: the desired exit code

  """

  op = opcodes.OpTagsSearch(pattern=args[0])

  result = SubmitOpCode(op, opts=opts)

  if not result:

    return 1

  result = list(result)

  result.sort()

  for path, tag in result:

    ToStdout("%s %s", path, tag)

def _ReadAndVerifyCert(cert_filename, verify_private_key=False):

  """Reads and verifies an X509 certificate.

  @type cert_filename: string

  @param cert_filename: the path of the file containing the certificate to

                        verify encoded in PEM format

  @type verify_private_key: bool

  @param verify_private_key: whether to verify the private key in addition to

                             the public certificate

  @rtype: string

  @return: a string containing the PEM-encoded certificate.

  """

  try:

    pem = utils.ReadFile(cert_filename)

  except IOError, err:

    raise errors.X509CertError(cert_filename,

                               "Unable to read certificate: %s" % str(err))

  try:

    OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, pem)

  except Exception, err:

    raise errors.X509CertError(cert_filename,

                               "Unable to load certificate: %s" % str(err))

  if verify_private_key:

    try:

      OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, pem)

    except Exception, err:

      raise errors.X509CertError(cert_filename,

                                 "Unable to load private key: %s" % str(err))

  return pem

def _RenewCrypto(new_cluster_cert, new_rapi_cert, # pylint: disable=R0911

                 rapi_cert_filename, new_spice_cert, spice_cert_filename,

                 spice_cacert_filename, new_confd_hmac_key, new_cds,

                 cds_filename, force, new_node_cert):

  """Renews cluster certificates, keys and secrets.

  @type new_cluster_cert: bool

  @param new_cluster_cert: Whether to generate a new cluster certificate

  @type new_rapi_cert: bool

  @param new_rapi_cert: Whether to generate a new RAPI certificate

  @type rapi_cert_filename: string

  @param rapi_cert_filename: Path to file containing new RAPI certificate

  @type new_spice_cert: bool

  @param new_spice_cert: Whether to generate a new SPICE certificate

  @type spice_cert_filename: string

  @param spice_cert_filename: Path to file containing new SPICE certificate

  @type spice_cacert_filename: string

  @param spice_cacert_filename: Path to file containing the certificate of the

                                CA that signed the SPICE certificate

  @type new_confd_hmac_key: bool

  @param new_confd_hmac_key: Whether to generate a new HMAC key

  @type new_cds: bool

  @param new_cds: Whether to generate a new cluster domain secret

  @type cds_filename: string

  @param cds_filename: Path to file containing new cluster domain secret

  @type force: bool

  @param force: Whether to ask user for confirmation

  @type new_node_cert: string

  @param new_node_cert: Whether to generate new node certificates

  """

  if new_rapi_cert and rapi_cert_filename:

    ToStderr("Only one of the --new-rapi-certificate and --rapi-certificate"

             " options can be specified at the same time.")

    return 1

  if new_cds and cds_filename:

    ToStderr("Only one of the --new-cluster-domain-secret and"

             " --cluster-domain-secret options can be specified at"

             " the same time.")

    return 1

  if new_spice_cert and (spice_cert_filename or spice_cacert_filename):

    ToStderr("When using --new-spice-certificate, the --spice-certificate"

             " and --spice-ca-certificate must not be used.")

    return 1

  if bool(spice_cacert_filename) ^ bool(spice_cert_filename):

    ToStderr("Both --spice-certificate and --spice-ca-certificate must be"

             " specified.")

    return 1

  rapi_cert_pem, spice_cert_pem, spice_cacert_pem = (None, None, None)

  try:

    if rapi_cert_filename:

      rapi_cert_pem = _ReadAndVerifyCert(rapi_cert_filename, True)

    if spice_cert_filename:

      spice_cert_pem = _ReadAndVerifyCert(spice_cert_filename, True)

      spice_cacert_pem = _ReadAndVerifyCert(spice_cacert_filename)

  except errors.X509CertError, err:

    ToStderr("Unable to load X509 certificate from %s: %s", err[0], err[1])

    return 1

  if cds_filename:

    try:

      cds = utils.ReadFile(cds_filename)

    except Exception, err: # pylint: disable=W0703

      ToStderr("Can't load new cluster domain secret from %s: %s" %

               (cds_filename, str(err)))

      return 1

  else:

    cds = None

  if not force:

    usertext = ("This requires all daemons on all nodes to be restarted and"

                " may take some time. Continue?")

    if not AskUser(usertext):

      return 1

  def _RenewCryptoInner(ctx):

    ctx.feedback_fn("Updating certificates and keys")

    # Note: the node certificate will be generated in the LU

    bootstrap.GenerateClusterCrypto(new_cluster_cert,

                                    new_rapi_cert,

                                    new_spice_cert,

                                    new_confd_hmac_key,

                                    new_cds,

                                    rapi_cert_pem=rapi_cert_pem,

                                    spice_cert_pem=spice_cert_pem,

                                    spice_cacert_pem=spice_cacert_pem,

                                    cds=cds)

    files_to_copy = []

    if new_cluster_cert:

      files_to_copy.append(pathutils.NODED_CERT_FILE)

    if new_rapi_cert or rapi_cert_pem:

      files_to_copy.append(pathutils.RAPI_CERT_FILE)

    if new_spice_cert or spice_cert_pem:

      files_to_copy.append(pathutils.SPICE_CERT_FILE)

      files_to_copy.append(pathutils.SPICE_CACERT_FILE)

    if new_confd_hmac_key:

      files_to_copy.append(pathutils.CONFD_HMAC_KEY)

    if new_cds or cds:

      files_to_copy.append(pathutils.CLUSTER_DOMAIN_SECRET_FILE)

    if files_to_copy:

      for node_name in ctx.nonmaster_nodes:

        port = ctx.ssh_ports[node_name]

        ctx.feedback_fn("Copying %s to %s:%d" %

                        (", ".join(files_to_copy), node_name, port))

        for file_name in files_to_copy:

          ctx.ssh.CopyFileToNode(node_name, port, file_name)

  RunWhileClusterStopped(ToStdout, _RenewCryptoInner)

  ToStdout("All requested certificates and keys have been replaced."

           " Running \"gnt-cluster verify\" now is recommended.")

  if new_node_cert:

    cl = GetClient()

    renew_op = opcodes.OpClusterRenewCrypto()

    SubmitOpCode(renew_op, cl=cl)

  return 0

def RenewCrypto(opts, args):

  """Renews cluster certificates, keys and secrets.

  """

  return _RenewCrypto(opts.new_cluster_cert,

                      opts.new_rapi_cert,

                      opts.rapi_cert,

                      opts.new_spice_cert,

                      opts.spice_cert,

                      opts.spice_cacert,

                      opts.new_confd_hmac_key,

                      opts.new_cluster_domain_secret,

                      opts.cluster_domain_secret,

                      opts.force,

                      opts.new_node_cert)

def _GetEnabledDiskTemplates(opts):

  """Determine the list of enabled disk templates.

  """

  if opts.enabled_disk_templates:

    return opts.enabled_disk_templates.split(",")

  else:

    return None

def _GetVgName(opts, enabled_disk_templates):

  """Determine the volume group name.

  @type enabled_disk_templates: list of strings

  @param enabled_disk_templates: cluster-wide enabled disk-templates

  """

  # consistency between vg name and enabled disk templates

  vg_name = None

  if opts.vg_name is not None:

    vg_name = opts.vg_name

  if enabled_disk_templates:

    if vg_name and not utils.IsLvmEnabled(enabled_disk_templates):

      ToStdout("You specified a volume group with --vg-name, but you did not"

               " enable any of the following lvm-based disk templates: %s" %

               utils.CommaJoin(constants.DTS_LVM))

  return vg_name

def _GetDrbdHelper(opts, enabled_disk_templates):

  """Determine the DRBD usermode helper.

  """

  drbd_helper = opts.drbd_helper

  if enabled_disk_templates:

    drbd_enabled = constants.DT_DRBD8 in enabled_disk_templates

    if not drbd_enabled and opts.drbd_helper:

      ToStdout("You specified a DRBD usermode helper with "

               " --drbd-usermode-helper while DRBD is not enabled.")

  return drbd_helper

def SetClusterParams(opts, args):

  """Modify the cluster.

  @param opts: the command line options selected by the user

  @type args: list

  @param args: should be an empty list