Synnefo » snf-ganeti

#!/usr/bin/python

#

# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Burnin program

"""

import sys

import optparse

import time

import socket

import urllib

from itertools import izip, islice, cycle

from cStringIO import StringIO

from ganeti import opcodes

from ganeti import constants

from ganeti import cli

from ganeti import errors

from ganeti import utils

from ganeti import hypervisor

from ganeti import compat

from ganeti import pathutils

from ganeti.confd import client as confd_client

from ganeti.runtime import (GetClient)

USAGE = ("\tburnin -o OS_NAME [options...] instance_name ...")

MAX_RETRIES = 3

LOG_HEADERS = {

  0: "- ",

  1: "* ",

  2: "",

  }

#: Disk templates supporting a single node

_SINGLE_NODE_DISK_TEMPLATES = compat.UniqueFrozenset([

  constants.DT_DISKLESS,

  constants.DT_PLAIN,

  constants.DT_FILE,

  constants.DT_SHARED_FILE,

  constants.DT_EXT,

  constants.DT_RBD,

  constants.DT_GLUSTER

  ])

_SUPPORTED_DISK_TEMPLATES = compat.UniqueFrozenset([

  constants.DT_DISKLESS,

  constants.DT_DRBD8,

  constants.DT_EXT,

  constants.DT_FILE,

  constants.DT_PLAIN,

  constants.DT_RBD,

  constants.DT_SHARED_FILE,

  constants.DT_GLUSTER

  ])

#: Disk templates for which import/export is tested

_IMPEXP_DISK_TEMPLATES = (_SUPPORTED_DISK_TEMPLATES - frozenset([

  constants.DT_DISKLESS,

  constants.DT_FILE,

  constants.DT_SHARED_FILE,

  constants.DT_GLUSTER

  ]))

class InstanceDown(Exception):

  """The checked instance was not up"""

class BurninFailure(Exception):

  """Failure detected during burning"""

def Usage():

  """Shows program usage information and exits the program."""

  print >> sys.stderr, "Usage:"

  print >> sys.stderr, USAGE

  sys.exit(2)

def Log(msg, *args, **kwargs):

  """Simple function that prints out its argument.

  """

  if args:

    msg = msg % args

  indent = kwargs.get("indent", 0)

  sys.stdout.write("%*s%s%s\n" % (2 * indent, "",

                                  LOG_HEADERS.get(indent, "  "), msg))

  sys.stdout.flush()

def Err(msg, exit_code=1):

  """Simple error logging that prints to stderr.

  """

  sys.stderr.write(msg + "\n")

  sys.stderr.flush()

  sys.exit(exit_code)

class SimpleOpener(urllib.FancyURLopener):

  """A simple url opener"""

  # pylint: disable=W0221

  def prompt_user_passwd(self, host, realm, clear_cache=0):

    """No-interaction version of prompt_user_passwd."""

    # we follow parent class' API

    # pylint: disable=W0613

    return None, None

  def http_error_default(self, url, fp, errcode, errmsg, headers):

    """Custom error handling"""

    # make sure sockets are not left in CLOSE_WAIT, this is similar

    # but with a different exception to the BasicURLOpener class

    _ = fp.read() # throw away data

    fp.close()

    raise InstanceDown("HTTP error returned: code %s, msg %s" %

                       (errcode, errmsg))

OPTIONS = [

  cli.cli_option("-o", "--os", dest="os", default=None,

                 help="OS to use during burnin",

                 metavar="<OS>",

                 completion_suggest=cli.OPT_COMPL_ONE_OS),

  cli.HYPERVISOR_OPT,

  cli.OSPARAMS_OPT,

  cli.cli_option("--disk-size", dest="disk_size",

                 help="Disk size (determines disk count)",

                 default="128m", type="string", metavar="<size,size,...>",

                 completion_suggest=("128M 512M 1G 4G 1G,256M"

                                     " 4G,1G,1G 10G").split()),

  cli.cli_option("--disk-growth", dest="disk_growth", help="Disk growth",

                 default="128m", type="string", metavar="<size,size,...>"),

  cli.cli_option("--mem-size", dest="mem_size", help="Memory size",

                 default=None, type="unit", metavar="<size>",

                 completion_suggest=("128M 256M 512M 1G 4G 8G"

                                     " 12G 16G").split()),

  cli.cli_option("--maxmem-size", dest="maxmem_size", help="Max Memory size",

                 default=256, type="unit", metavar="<size>",

                 completion_suggest=("128M 256M 512M 1G 4G 8G"

                                     " 12G 16G").split()),

  cli.cli_option("--minmem-size", dest="minmem_size", help="Min Memory size",

                 default=128, type="unit", metavar="<size>",

                 completion_suggest=("128M 256M 512M 1G 4G 8G"

                                     " 12G 16G").split()),

  cli.cli_option("--vcpu-count", dest="vcpu_count", help="VCPU count",

                 default=3, type="unit", metavar="<count>",

                 completion_suggest=("1 2 3 4").split()),

  cli.DEBUG_OPT,

  cli.VERBOSE_OPT,

  cli.NOIPCHECK_OPT,

  cli.NONAMECHECK_OPT,

  cli.EARLY_RELEASE_OPT,

  cli.cli_option("--no-replace1", dest="do_replace1",

                 help="Skip disk replacement with the same secondary",

                 action="store_false", default=True),

  cli.cli_option("--no-replace2", dest="do_replace2",

                 help="Skip disk replacement with a different secondary",

                 action="store_false", default=True),

  cli.cli_option("--no-failover", dest="do_failover",

                 help="Skip instance failovers", action="store_false",

                 default=True),

  cli.cli_option("--no-migrate", dest="do_migrate",

                 help="Skip instance live migration",

                 action="store_false", default=True),

  cli.cli_option("--no-move", dest="do_move",

                 help="Skip instance moves", action="store_false",

                 default=True),

  cli.cli_option("--no-importexport", dest="do_importexport",

                 help="Skip instance export/import", action="store_false",

                 default=True),

  cli.cli_option("--no-startstop", dest="do_startstop",

                 help="Skip instance stop/start", action="store_false",

                 default=True),

  cli.cli_option("--no-reinstall", dest="do_reinstall",

                 help="Skip instance reinstall", action="store_false",

                 default=True),

  cli.cli_option("--no-reboot", dest="do_reboot",

                 help="Skip instance reboot", action="store_false",

                 default=True),

  cli.cli_option("--no-renamesame", dest="do_renamesame",

                 help="Skip instance rename to same name", action="store_false",

                 default=True),

  cli.cli_option("--reboot-types", dest="reboot_types",

                 help="Specify the reboot types", default=None),

  cli.cli_option("--no-activate-disks", dest="do_activate_disks",

                 help="Skip disk activation/deactivation",

                 action="store_false", default=True),

  cli.cli_option("--no-add-disks", dest="do_addremove_disks",

                 help="Skip disk addition/removal",

                 action="store_false", default=True),

  cli.cli_option("--no-add-nics", dest="do_addremove_nics",

                 help="Skip NIC addition/removal",

                 action="store_false", default=True),

  cli.cli_option("--no-nics", dest="nics",

                 help="No network interfaces", action="store_const",

                 const=[], default=[{}]),

  cli.cli_option("--no-confd", dest="do_confd_tests",

                 help="Skip confd queries",

                 action="store_false", default=constants.ENABLE_CONFD),

  cli.cli_option("--rename", dest="rename", default=None,

                 help=("Give one unused instance name which is taken"

                       " to start the renaming sequence"),

                 metavar="<instance_name>"),

  cli.cli_option("-t", "--disk-template", dest="disk_template",

                 choices=list(_SUPPORTED_DISK_TEMPLATES),

                 default=constants.DT_DRBD8,

                 help=("Disk template (default %s, otherwise one of %s)" %

                       (constants.DT_DRBD8,

                        utils.CommaJoin(_SUPPORTED_DISK_TEMPLATES)))),

  cli.cli_option("-n", "--nodes", dest="nodes", default="",

                 help=("Comma separated list of nodes to perform"

                       " the burnin on (defaults to all nodes)"),

                 completion_suggest=cli.OPT_COMPL_MANY_NODES),

  cli.cli_option("-I", "--iallocator", dest="iallocator",

                 default=None, type="string",

                 help=("Perform the allocation using an iallocator"

                       " instead of fixed node spread (node restrictions no"

                       " longer apply, therefore -n/--nodes must not be"

                       " used"),

                 completion_suggest=cli.OPT_COMPL_ONE_IALLOCATOR),

  cli.cli_option("-p", "--parallel", default=False, action="store_true",

                 dest="parallel",

                 help=("Enable parallelization of some operations in"

                       " order to speed burnin or to test granular locking")),

  cli.cli_option("--net-timeout", default=15, type="int",

                 dest="net_timeout",

                 help=("The instance check network timeout in seconds"

                       " (defaults to 15 seconds)"),

                 completion_suggest="15 60 300 900".split()),

  cli.cli_option("-C", "--http-check", default=False, action="store_true",

                 dest="http_check",

                 help=("Enable checking of instance status via http,"

                       " looking for /hostname.txt that should contain the"

                       " name of the instance")),

  cli.cli_option("-K", "--keep-instances", default=False,

                 action="store_true",

                 dest="keep_instances",

                 help=("Leave instances on the cluster after burnin,"

                       " for investigation in case of errors or simply"

                       " to use them")),

  cli.REASON_OPT,

  ]

# Mainly used for bash completion

ARGUMENTS = [cli.ArgInstance(min=1)]

def _DoCheckInstances(fn):

  """Decorator for checking instances.

  """

  def wrapper(self, *args, **kwargs):

    val = fn(self, *args, **kwargs)

    for instance in self.instances:

      self._CheckInstanceAlive(instance) # pylint: disable=W0212

    return val

  return wrapper

def _DoBatch(retry):

  """Decorator for possible batch operations.

  Must come after the _DoCheckInstances decorator (if any).

  @param retry: whether this is a retryable batch, will be

      passed to StartBatch

  """

  def wrap(fn):

    def batched(self, *args, **kwargs):

      self.StartBatch(retry)

      val = fn(self, *args, **kwargs)

      self.CommitQueue()

      return val

    return batched

  return wrap

class Burner(object):

  """Burner class."""

  def __init__(self):

    """Constructor."""

    self.url_opener = SimpleOpener()

    self._feed_buf = StringIO()

    self.nodes = []

    self.instances = []

    self.to_rem = []

    self.queued_ops = []

    self.opts = None

    self.queue_retry = False

    self.disk_count = self.disk_growth = self.disk_size = None

    self.hvp = self.bep = None

    self.ParseOptions()

    self.cl = cli.GetClient()

    self.GetState()

  def ClearFeedbackBuf(self):

    """Clear the feedback buffer."""

    self._feed_buf.truncate(0)

  def GetFeedbackBuf(self):

    """Return the contents of the buffer."""

    return self._feed_buf.getvalue()

  def Feedback(self, msg):

    """Acumulate feedback in our buffer."""

    formatted_msg = "%s %s" % (time.ctime(utils.MergeTime(msg[0])), msg[2])

    self._feed_buf.write(formatted_msg + "\n")

    if self.opts.verbose:

      Log(formatted_msg, indent=3)

  def MaybeRetry(self, retry_count, msg, fn, *args):

    """Possibly retry a given function execution.

    @type retry_count: int

    @param retry_count: retry counter:

        - 0: non-retryable action

        - 1: last retry for a retryable action

        - MAX_RETRIES: original try for a retryable action

    @type msg: str

    @param msg: the kind of the operation

    @type fn: callable

    @param fn: the function to be called

    """

    try:

      val = fn(*args)

      if retry_count > 0 and retry_count < MAX_RETRIES:

        Log("Idempotent %s succeeded after %d retries",

            msg, MAX_RETRIES - retry_count)

      return val

    except Exception, err: # pylint: disable=W0703

      if retry_count == 0:

        Log("Non-idempotent %s failed, aborting", msg)

        raise

      elif retry_count == 1:

        Log("Idempotent %s repeated failure, aborting", msg)

        raise

      else:

        Log("Idempotent %s failed, retry #%d/%d: %s",

            msg, MAX_RETRIES - retry_count + 1, MAX_RETRIES, err)

        self.MaybeRetry(retry_count - 1, msg, fn, *args)

  def _ExecOp(self, *ops):

    """Execute one or more opcodes and manage the exec buffer.

    @return: if only opcode has been passed, we return its result;

        otherwise we return the list of results

    """

    job_id = cli.SendJob(ops, cl=self.cl)

    results = cli.PollJob(job_id, cl=self.cl, feedback_fn=self.Feedback)

    if len(ops) == 1:

      return results[0]

    else:

      return results

  def ExecOp(self, retry, *ops):

    """Execute one or more opcodes and manage the exec buffer.

    @return: if only opcode has been passed, we return its result;

        otherwise we return the list of results

    """

    if retry:

      rval = MAX_RETRIES

    else:

      rval = 0

    cli.SetGenericOpcodeOpts(ops, self.opts)

    return self.MaybeRetry(rval, "opcode", self._ExecOp, *ops)

  def ExecOrQueue(self, name, ops, post_process=None):

    """Execute an opcode and manage the exec buffer."""

    if self.opts.parallel:

      cli.SetGenericOpcodeOpts(ops, self.opts)

      self.queued_ops.append((ops, name, post_process))

    else:

      val = self.ExecOp(self.queue_retry, *ops) # pylint: disable=W0142

      if post_process is not None:

        post_process()

      return val

  def StartBatch(self, retry):

    """Start a new batch of jobs.

    @param retry: whether this is a retryable batch

    """

    self.queued_ops = []

    self.queue_retry = retry

  def CommitQueue(self):

    """Execute all submitted opcodes in case of parallel burnin"""

    if not self.opts.parallel or not self.queued_ops:

      return

    if self.queue_retry:

      rval = MAX_RETRIES

    else:

      rval = 0

    try:

      results = self.MaybeRetry(rval, "jobset", self.ExecJobSet,

                                self.queued_ops)

    finally:

      self.queued_ops = []

    return results

  def ExecJobSet(self, jobs):

    """Execute a set of jobs and return once all are done.

    The method will return the list of results, if all jobs are

    successful. Otherwise, OpExecError will be raised from within

    cli.py.

    """

    self.ClearFeedbackBuf()

    jex = cli.JobExecutor(cl=self.cl, feedback_fn=self.Feedback)

    for ops, name, _ in jobs:

      jex.QueueJob(name, *ops) # pylint: disable=W0142

    try:

      results = jex.GetResults()

    except Exception, err: # pylint: disable=W0703

      Log("Jobs failed: %s", err)

      raise BurninFailure()

    fail = False

    val = []

    for (_, name, post_process), (success, result) in zip(jobs, results):

      if success:

        if post_process:

          try:

            post_process()

          except Exception, err: # pylint: disable=W0703

            Log("Post process call for job %s failed: %s", name, err)

            fail = True

        val.append(result)

      else:

        fail = True

    if fail:

      raise BurninFailure()

    return val

  def ParseOptions(self):

    """Parses the command line options.

    In case of command line errors, it will show the usage and exit the

    program.

    """

    parser = optparse.OptionParser(usage="\n%s" % USAGE,

                                   version=("%%prog (ganeti) %s" %

                                            constants.RELEASE_VERSION),

                                   option_list=OPTIONS)

    options, args = parser.parse_args()

    if len(args) < 1 or options.os is None:

      Usage()

    if options.mem_size:

      options.maxmem_size = options.mem_size

      options.minmem_size = options.mem_size

    elif options.minmem_size > options.maxmem_size:

      Err("Maximum memory lower than minimum memory")

    if options.disk_template not in _SUPPORTED_DISK_TEMPLATES:

      Err("Unknown or unsupported disk template '%s'" % options.disk_template)

    if options.disk_template == constants.DT_DISKLESS:

      disk_size = disk_growth = []

      options.do_addremove_disks = False

    else:

      disk_size = [utils.ParseUnit(v) for v in options.disk_size.split(",")]

      disk_growth = [utils.ParseUnit(v)

                     for v in options.disk_growth.split(",")]

      if len(disk_growth) != len(disk_size):

        Err("Wrong disk sizes/growth combination")

    if ((disk_size and options.disk_template == constants.DT_DISKLESS) or

        (not disk_size and options.disk_template != constants.DT_DISKLESS)):

      Err("Wrong disk count/disk template combination")

    self.disk_size = disk_size

    self.disk_growth = disk_growth

    self.disk_count = len(disk_size)

    if options.nodes and options.iallocator:

      Err("Give either the nodes option or the iallocator option, not both")

    if options.http_check and not options.name_check:

      Err("Can't enable HTTP checks without name checks")

    self.opts = options

    self.instances = args

    self.bep = {

      constants.BE_MINMEM: options.minmem_size,

      constants.BE_MAXMEM: options.maxmem_size,

      constants.BE_VCPUS: options.vcpu_count,

      }

    self.hypervisor = None

    self.hvp = {}

    if options.hypervisor:

      self.hypervisor, self.hvp = options.hypervisor

    if options.reboot_types is None:

      options.reboot_types = constants.REBOOT_TYPES

    else:

      options.reboot_types = options.reboot_types.split(",")

      rt_diff = set(options.reboot_types).difference(constants.REBOOT_TYPES)

      if rt_diff:

        Err("Invalid reboot types specified: %s" % utils.CommaJoin(rt_diff))

    socket.setdefaulttimeout(options.net_timeout)

  def GetState(self):

    """Read the cluster state from the master daemon."""

    if self.opts.nodes:

      names = self.opts.nodes.split(",")

    else:

      names = []

    try:

      qcl = GetClient(query=True)

      result = qcl.QueryNodes(names, ["name", "offline", "drained"], False)

    except errors.GenericError, err:

      err_code, msg = cli.FormatError(err)

      Err(msg, exit_code=err_code)

    finally:

      qcl.Close()

    self.nodes = [data[0] for data in result if not (data[1] or data[2])]

    op_diagnose = opcodes.OpOsDiagnose(output_fields=["name",

                                                      "variants",

                                                      "hidden"],

                                       names=[])

    result = self.ExecOp(True, op_diagnose)

    if not result:

      Err("Can't get the OS list")

    found = False

    for (name, variants, _) in result:

      if self.opts.os in cli.CalculateOSNames(name, variants):

        found = True

        break

    if not found:

      Err("OS '%s' not found" % self.opts.os)

    cluster_info = self.cl.QueryClusterInfo()

    self.cluster_info = cluster_info

    if not self.cluster_info:

      Err("Can't get cluster info")

    default_nic_params = self.cluster_info["nicparams"][constants.PP_DEFAULT]

    self.cluster_default_nicparams = default_nic_params

    if self.hypervisor is None:

      self.hypervisor = self.cluster_info["default_hypervisor"]

    self.hv_can_migrate = \

      hypervisor.GetHypervisorClass(self.hypervisor).CAN_MIGRATE

  @_DoCheckInstances

  @_DoBatch(False)

  def BurnCreateInstances(self):

    """Create the given instances.

    """

    self.to_rem = []

    mytor = izip(cycle(self.nodes),

                 islice(cycle(self.nodes), 1, None),

                 self.instances)

    Log("Creating instances")

    for pnode, snode, instance in mytor:

      Log("instance %s", instance, indent=1)

      if self.opts.iallocator:

        pnode = snode = None

        msg = "with iallocator %s" % self.opts.iallocator

      elif self.opts.disk_template not in constants.DTS_INT_MIRROR:

        snode = None

        msg = "on %s" % pnode

      else:

        msg = "on %s, %s" % (pnode, snode)

      Log(msg, indent=2)

      op = opcodes.OpInstanceCreate(instance_name=instance,

                                    disks=[{"size": size}

                                           for size in self.disk_size],

                                    disk_template=self.opts.disk_template,

                                    nics=self.opts.nics,

                                    mode=constants.INSTANCE_CREATE,

                                    os_type=self.opts.os,

                                    pnode=pnode,

                                    snode=snode,

                                    start=True,

                                    ip_check=self.opts.ip_check,

                                    name_check=self.opts.name_check,

                                    wait_for_sync=True,

                                    file_driver="loop",

                                    file_storage_dir=None,

                                    iallocator=self.opts.iallocator,

                                    beparams=self.bep,

                                    hvparams=self.hvp,

                                    hypervisor=self.hypervisor,

                                    osparams=self.opts.osparams,

                                    )

      remove_instance = lambda name: lambda: self.to_rem.append(name)

      self.ExecOrQueue(instance, [op], post_process=remove_instance(instance))

  @_DoBatch(False)

  def BurnModifyRuntimeMemory(self):

    """Alter the runtime memory."""

    Log("Setting instance runtime memory")

    for instance in self.instances:

      Log("instance %s", instance, indent=1)

      tgt_mem = self.bep[constants.BE_MINMEM]

      op = opcodes.OpInstanceSetParams(instance_name=instance,

                                       runtime_mem=tgt_mem)

      Log("Set memory to %s MB", tgt_mem, indent=2)

      self.ExecOrQueue(instance, [op])

  @_DoBatch(False)

  def BurnGrowDisks(self):

    """Grow both the os and the swap disks by the requested amount, if any."""

    Log("Growing disks")

    for instance in self.instances:

      Log("instance %s", instance, indent=1)

      for idx, growth in enumerate(self.disk_growth):

        if growth > 0:

          op = opcodes.OpInstanceGrowDisk(instance_name=instance, disk=idx,

                                          amount=growth, wait_for_sync=True)

          Log("increase disk/%s by %s MB", idx, growth, indent=2)

          self.ExecOrQueue(instance, [op])

  @_DoBatch(True)

  def BurnReplaceDisks1D8(self):

    """Replace disks on primary and secondary for drbd8."""

    Log("Replacing disks on the same nodes")

    early_release = self.opts.early_release

    for instance in self.instances:

      Log("instance %s", instance, indent=1)

      ops = []

      for mode in constants.REPLACE_DISK_SEC, constants.REPLACE_DISK_PRI:

        op = opcodes.OpInstanceReplaceDisks(instance_name=instance,

                                            mode=mode,

                                            disks=list(range(self.disk_count)),

                                            early_release=early_release)

        Log("run %s", mode, indent=2)

        ops.append(op)

      self.ExecOrQueue(instance, ops)

  @_DoBatch(True)

  def BurnReplaceDisks2(self):

    """Replace secondary node."""

    Log("Changing the secondary node")

    mode = constants.REPLACE_DISK_CHG

    mytor = izip(islice(cycle(self.nodes), 2, None),

                 self.instances)

    for tnode, instance in mytor:

      Log("instance %s", instance, indent=1)

      if self.opts.iallocator:

        tnode = None

        msg = "with iallocator %s" % self.opts.iallocator

      else:

        msg = tnode

      op = opcodes.OpInstanceReplaceDisks(instance_name=instance,

                                          mode=mode,

                                          remote_node=tnode,

                                          iallocator=self.opts.iallocator,

                                          disks=[],

                                          early_release=self.opts.early_release)

      Log("run %s %s", mode, msg, indent=2)

      self.ExecOrQueue(instance, [op])

  @_DoCheckInstances

  @_DoBatch(False)

  def BurnFailover(self):

    """Failover the instances."""

    Log("Failing over instances")

    for instance in self.instances:

      Log("instance %s", instance, indent=1)

      op = opcodes.OpInstanceFailover(instance_name=instance,

                                      ignore_consistency=False)

      self.ExecOrQueue(instance, [op])

  @_DoCheckInstances

  @_DoBatch(False)

  def BurnMove(self):

    """Move the instances."""

    Log("Moving instances")

    mytor = izip(islice(cycle(self.nodes), 1, None),

                 self.instances)

    for tnode, instance in mytor:

      Log("instance %s", instance, indent=1)

      op = opcodes.OpInstanceMove(instance_name=instance,

                                  target_node=tnode)

      self.ExecOrQueue(instance, [op])

  @_DoBatch(False)

  def BurnMigrate(self):

    """Migrate the instances."""

    Log("Migrating instances")

    for instance in self.instances: