code.grnet.gr Git - ganeti-local/blob - tools/burnin

   1 #!/usr/bin/python
   2 #
   3
   4 # Copyright (C) 2006, 2007 Google Inc.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Burnin program
  23
  24 """
  25
  26 import sys
  27 import optparse
  28 import time
  29 import socket
  30 import urllib
  31 from itertools import izip, islice, cycle
  32 from cStringIO import StringIO
  33
  34 from ganeti import opcodes
  35 from ganeti import constants
  36 from ganeti import cli
  37 from ganeti import errors
  38 from ganeti import utils
  39
  40
  41 USAGE = ("\tburnin -o OS_NAME [options...] instance_name ...")
  42
  43 MAX_RETRIES = 3
  44
  45 class InstanceDown(Exception):
  46   """The checked instance was not up"""
  47
  48
  49 class BurninFailure(Exception):
  50   """Failure detected during burning"""
  51
  52
  53 def Usage():
  54   """Shows program usage information and exits the program."""
  55
  56   print >> sys.stderr, "Usage:"
  57   print >> sys.stderr, USAGE
  58   sys.exit(2)
  59
  60
  61 def Log(msg, indent=0):
  62   """Simple function that prints out its argument.
  63
  64   """
  65   headers = {
  66     0: "- ",
  67     1: "* ",
  68     2: ""
  69     }
  70   sys.stdout.write("%*s%s%s\n" % (2*indent, "",
  71                                    headers.get(indent, "  "), msg))
  72   sys.stdout.flush()
  73
  74 def Err(msg, exit_code=1):
  75   """Simple error logging that prints to stderr.
  76
  77   """
  78   sys.stderr.write(msg + "\n")
  79   sys.stderr.flush()
  80   sys.exit(exit_code)
  81
  82
  83 class SimpleOpener(urllib.FancyURLopener):
  84   """A simple url opener"""
  85   # pylint: disable-msg=W0221
  86
  87   def prompt_user_passwd(self, host, realm, clear_cache=0):
  88     """No-interaction version of prompt_user_passwd."""
  89     # we follow parent class' API
  90     # pylint: disable-msg=W0613
  91     return None, None
  92
  93   def http_error_default(self, url, fp, errcode, errmsg, headers):
  94     """Custom error handling"""
  95     # make sure sockets are not left in CLOSE_WAIT, this is similar
  96     # but with a different exception to the BasicURLOpener class
  97     _ = fp.read() # throw away data
  98     fp.close()
  99     raise InstanceDown("HTTP error returned: code %s, msg %s" %
 100                        (errcode, errmsg))
 101
 102
 103 OPTIONS = [
 104   cli.cli_option("-o", "--os", dest="os", default=None,
 105                  help="OS to use during burnin",
 106                  metavar="<OS>",
 107                  completion_suggest=cli.OPT_COMPL_ONE_OS),
 108   cli.cli_option("--disk-size", dest="disk_size",
 109                  help="Disk size (determines disk count)",
 110                  default="128m", type="string", metavar="<size,size,...>",
 111                  completion_suggest=("128M 512M 1G 4G 1G,256M"
 112                                      " 4G,1G,1G 10G").split()),
 113   cli.cli_option("--disk-growth", dest="disk_growth", help="Disk growth",
 114                  default="128m", type="string", metavar="<size,size,...>"),
 115   cli.cli_option("--mem-size", dest="mem_size", help="Memory size",
 116                  default=128, type="unit", metavar="<size>",
 117                  completion_suggest=("128M 256M 512M 1G 4G 8G"
 118                                      " 12G 16G").split()),
 119   cli.DEBUG_OPT,
 120   cli.VERBOSE_OPT,
 121   cli.NOIPCHECK_OPT,
 122   cli.NONAMECHECK_OPT,
 123   cli.EARLY_RELEASE_OPT,
 124   cli.cli_option("--no-replace1", dest="do_replace1",
 125                  help="Skip disk replacement with the same secondary",
 126                  action="store_false", default=True),
 127   cli.cli_option("--no-replace2", dest="do_replace2",
 128                  help="Skip disk replacement with a different secondary",
 129                  action="store_false", default=True),
 130   cli.cli_option("--no-failover", dest="do_failover",
 131                  help="Skip instance failovers", action="store_false",
 132                  default=True),
 133   cli.cli_option("--no-migrate", dest="do_migrate",
 134                  help="Skip instance live migration",
 135                  action="store_false", default=True),
 136   cli.cli_option("--no-move", dest="do_move",
 137                  help="Skip instance moves", action="store_false",
 138                  default=True),
 139   cli.cli_option("--no-importexport", dest="do_importexport",
 140                  help="Skip instance export/import", action="store_false",
 141                  default=True),
 142   cli.cli_option("--no-startstop", dest="do_startstop",
 143                  help="Skip instance stop/start", action="store_false",
 144                  default=True),
 145   cli.cli_option("--no-reinstall", dest="do_reinstall",
 146                  help="Skip instance reinstall", action="store_false",
 147                  default=True),
 148   cli.cli_option("--no-reboot", dest="do_reboot",
 149                  help="Skip instance reboot", action="store_false",
 150                  default=True),
 151   cli.cli_option("--no-activate-disks", dest="do_activate_disks",
 152                  help="Skip disk activation/deactivation",
 153                  action="store_false", default=True),
 154   cli.cli_option("--no-add-disks", dest="do_addremove_disks",
 155                  help="Skip disk addition/removal",
 156                  action="store_false", default=True),
 157   cli.cli_option("--no-add-nics", dest="do_addremove_nics",
 158                  help="Skip NIC addition/removal",
 159                  action="store_false", default=True),
 160   cli.cli_option("--no-nics", dest="nics",
 161                  help="No network interfaces", action="store_const",
 162                  const=[], default=[{}]),
 163   cli.cli_option("--rename", dest="rename", default=None,
 164                  help=("Give one unused instance name which is taken"
 165                        " to start the renaming sequence"),
 166                  metavar="<instance_name>"),
 167   cli.cli_option("-t", "--disk-template", dest="disk_template",
 168                  choices=list(constants.DISK_TEMPLATES),
 169                  default=constants.DT_DRBD8,
 170                  help="Disk template (diskless, file, plain or drbd) [drbd]"),
 171   cli.cli_option("-n", "--nodes", dest="nodes", default="",
 172                  help=("Comma separated list of nodes to perform"
 173                        " the burnin on (defaults to all nodes)"),
 174                  completion_suggest=cli.OPT_COMPL_MANY_NODES),
 175   cli.cli_option("-I", "--iallocator", dest="iallocator",
 176                  default=None, type="string",
 177                  help=("Perform the allocation using an iallocator"
 178                        " instead of fixed node spread (node restrictions no"
 179                        " longer apply, therefore -n/--nodes must not be"
 180                        " used"),
 181                  completion_suggest=cli.OPT_COMPL_ONE_IALLOCATOR),
 182   cli.cli_option("-p", "--parallel", default=False, action="store_true",
 183                  dest="parallel",
 184                  help=("Enable parallelization of some operations in"
 185                        " order to speed burnin or to test granular locking")),
 186   cli.cli_option("--net-timeout", default=15, type="int",
 187                  dest="net_timeout",
 188                  help=("The instance check network timeout in seconds"
 189                        " (defaults to 15 seconds)"),
 190                  completion_suggest="15 60 300 900".split()),
 191   cli.cli_option("-C", "--http-check", default=False, action="store_true",
 192                  dest="http_check",
 193                  help=("Enable checking of instance status via http,"
 194                        " looking for /hostname.txt that should contain the"
 195                        " name of the instance")),
 196   cli.cli_option("-K", "--keep-instances", default=False,
 197                  action="store_true",
 198                  dest="keep_instances",
 199                  help=("Leave instances on the cluster after burnin,"
 200                        " for investigation in case of errors or simply"
 201                        " to use them")),
 202   ]
 203
 204 # Mainly used for bash completion
 205 ARGUMENTS = [cli.ArgInstance(min=1)]
 206
 207
 208 def _DoCheckInstances(fn):
 209   """Decorator for checking instances.
 210
 211   """
 212   def wrapper(self, *args, **kwargs):
 213     val = fn(self, *args, **kwargs)
 214     for instance in self.instances:
 215       self._CheckInstanceAlive(instance) # pylint: disable-msg=W0212
 216     return val
 217
 218   return wrapper
 219
 220
 221 def _DoBatch(retry):
 222   """Decorator for possible batch operations.
 223
 224   Must come after the _DoCheckInstances decorator (if any).
 225
 226   @param retry: whether this is a retryable batch, will be
 227       passed to StartBatch
 228
 229   """
 230   def wrap(fn):
 231     def batched(self, *args, **kwargs):
 232       self.StartBatch(retry)
 233       val = fn(self, *args, **kwargs)
 234       self.CommitQueue()
 235       return val
 236     return batched
 237
 238   return wrap
 239
 240
 241 class Burner(object):
 242   """Burner class."""
 243
 244   def __init__(self):
 245     """Constructor."""
 246     utils.SetupLogging(constants.LOG_BURNIN, debug=False, stderr_logging=True)
 247     self.url_opener = SimpleOpener()
 248     self._feed_buf = StringIO()
 249     self.nodes = []
 250     self.instances = []
 251     self.to_rem = []
 252     self.queued_ops = []
 253     self.opts = None
 254     self.queue_retry = False
 255     self.disk_count = self.disk_growth = self.disk_size = None
 256     self.hvp = self.bep = None
 257     self.ParseOptions()
 258     self.cl = cli.GetClient()
 259     self.GetState()
 260
 261   def ClearFeedbackBuf(self):
 262     """Clear the feedback buffer."""
 263     self._feed_buf.truncate(0)
 264
 265   def GetFeedbackBuf(self):
 266     """Return the contents of the buffer."""
 267     return self._feed_buf.getvalue()
 268
 269   def Feedback(self, msg):
 270     """Acumulate feedback in our buffer."""
 271     formatted_msg = "%s %s" % (time.ctime(utils.MergeTime(msg[0])), msg[2])
 272     self._feed_buf.write(formatted_msg + "\n")
 273     if self.opts.verbose:
 274       Log(formatted_msg, indent=3)
 275
 276   def MaybeRetry(self, retry_count, msg, fn, *args):
 277     """Possibly retry a given function execution.
 278
 279     @type retry_count: int
 280     @param retry_count: retry counter:
 281         - 0: non-retryable action
 282         - 1: last retry for a retryable action
 283         - MAX_RETRIES: original try for a retryable action
 284     @type msg: str
 285     @param msg: the kind of the operation
 286     @type fn: callable
 287     @param fn: the function to be called
 288
 289     """
 290     try:
 291       val = fn(*args)
 292       if retry_count > 0 and retry_count < MAX_RETRIES:
 293         Log("Idempotent %s succeeded after %d retries" %
 294             (msg, MAX_RETRIES - retry_count))
 295       return val
 296     except Exception, err: # pylint: disable-msg=W0703
 297       if retry_count == 0:
 298         Log("Non-idempotent %s failed, aborting" % (msg, ))
 299         raise
 300       elif retry_count == 1:
 301         Log("Idempotent %s repeated failure, aborting" % (msg, ))
 302         raise
 303       else:
 304         Log("Idempotent %s failed, retry #%d/%d: %s" %
 305             (msg, MAX_RETRIES - retry_count + 1, MAX_RETRIES, err))
 306         self.MaybeRetry(retry_count - 1, msg, fn, *args)
 307
 308   def _SetDebug(self, ops):
 309     """Set the debug value on the given opcodes"""
 310     for op in ops:
 311       op.debug_level = self.opts.debug
 312
 313   def _ExecOp(self, *ops):
 314     """Execute one or more opcodes and manage the exec buffer.
 315
 316     @result: if only opcode has been passed, we return its result;
 317         otherwise we return the list of results
 318
 319     """
 320     job_id = cli.SendJob(ops, cl=self.cl)
 321     results = cli.PollJob(job_id, cl=self.cl, feedback_fn=self.Feedback)
 322     if len(ops) == 1:
 323       return results[0]
 324     else:
 325       return results
 326
 327   def ExecOp(self, retry, *ops):
 328     """Execute one or more opcodes and manage the exec buffer.
 329
 330     @result: if only opcode has been passed, we return its result;
 331         otherwise we return the list of results
 332
 333     """
 334     if retry:
 335       rval = MAX_RETRIES
 336     else:
 337       rval = 0
 338     self._SetDebug(ops)
 339     return self.MaybeRetry(rval, "opcode", self._ExecOp, *ops)
 340
 341   def ExecOrQueue(self, name, *ops):
 342     """Execute an opcode and manage the exec buffer."""
 343     if self.opts.parallel:
 344       self._SetDebug(ops)
 345       self.queued_ops.append((ops, name))
 346     else:
 347       return self.ExecOp(self.queue_retry, *ops)
 348
 349   def StartBatch(self, retry):
 350     """Start a new batch of jobs.
 351
 352     @param retry: whether this is a retryable batch
 353
 354     """
 355     self.queued_ops = []
 356     self.queue_retry = retry
 357
 358   def CommitQueue(self):
 359     """Execute all submitted opcodes in case of parallel burnin"""
 360     if not self.opts.parallel:
 361       return
 362
 363     if self.queue_retry:
 364       rval = MAX_RETRIES
 365     else:
 366       rval = 0
 367
 368     try:
 369       results = self.MaybeRetry(rval, "jobset", self.ExecJobSet,
 370                                 self.queued_ops)
 371     finally:
 372       self.queued_ops = []
 373     return results
 374
 375   def ExecJobSet(self, jobs):
 376     """Execute a set of jobs and return once all are done.
 377
 378     The method will return the list of results, if all jobs are
 379     successful. Otherwise, OpExecError will be raised from within
 380     cli.py.
 381
 382     """
 383     self.ClearFeedbackBuf()
 384     job_ids = [cli.SendJob(row[0], cl=self.cl) for row in jobs]
 385     Log("Submitted job ID(s) %s" % utils.CommaJoin(job_ids), indent=1)
 386     results = []
 387     for jid, (_, iname) in zip(job_ids, jobs):
 388       Log("waiting for job %s for %s" % (jid, iname), indent=2)
 389       try:
 390         results.append(cli.PollJob(jid, cl=self.cl, feedback_fn=self.Feedback))
 391       except Exception, err: # pylint: disable-msg=W0703
 392         Log("Job for %s failed: %s" % (iname, err))
 393     if len(results) != len(jobs):
 394       raise BurninFailure()
 395     return results
 396
 397   def ParseOptions(self):
 398     """Parses the command line options.
 399
 400     In case of command line errors, it will show the usage and exit the
 401     program.
 402
 403     """
 404     parser = optparse.OptionParser(usage="\n%s" % USAGE,
 405                                    version=("%%prog (ganeti) %s" %
 406                                             constants.RELEASE_VERSION),
 407                                    option_list=OPTIONS)
 408
 409     options, args = parser.parse_args()
 410     if len(args) < 1 or options.os is None:
 411       Usage()
 412
 413     supported_disk_templates = (constants.DT_DISKLESS,
 414                                 constants.DT_FILE,
 415                                 constants.DT_PLAIN,
 416                                 constants.DT_DRBD8)
 417     if options.disk_template not in supported_disk_templates:
 418       Err("Unknown disk template '%s'" % options.disk_template)
 419
 420     if options.disk_template == constants.DT_DISKLESS:
 421       disk_size = disk_growth = []
 422       options.do_addremove_disks = False
 423     else:
 424       disk_size = [utils.ParseUnit(v) for v in options.disk_size.split(",")]
 425       disk_growth = [utils.ParseUnit(v)
 426                      for v in options.disk_growth.split(",")]
 427       if len(disk_growth) != len(disk_size):
 428         Err("Wrong disk sizes/growth combination")
 429     if ((disk_size and options.disk_template == constants.DT_DISKLESS) or
 430         (not disk_size and options.disk_template != constants.DT_DISKLESS)):
 431       Err("Wrong disk count/disk template combination")
 432
 433     self.disk_size = disk_size
 434     self.disk_growth = disk_growth
 435     self.disk_count = len(disk_size)
 436
 437     if options.nodes and options.iallocator:
 438       Err("Give either the nodes option or the iallocator option, not both")
 439
 440     if options.http_check and not options.name_check:
 441       Err("Can't enable HTTP checks without name checks")
 442
 443     self.opts = options
 444     self.instances = args
 445     self.bep = {
 446       constants.BE_MEMORY: options.mem_size,
 447       constants.BE_VCPUS: 1,
 448       }
 449     self.hvp = {}
 450
 451     socket.setdefaulttimeout(options.net_timeout)
 452
 453   def GetState(self):
 454     """Read the cluster state from the config."""
 455     if self.opts.nodes:
 456       names = self.opts.nodes.split(",")
 457     else:
 458       names = []
 459     try:
 460       op = opcodes.OpQueryNodes(output_fields=["name", "offline", "drained"],
 461                                 names=names, use_locking=True)
 462       result = self.ExecOp(True, op)
 463     except errors.GenericError, err:
 464       err_code, msg = cli.FormatError(err)
 465       Err(msg, exit_code=err_code)
 466     self.nodes = [data[0] for data in result if not (data[1] or data[2])]
 467
 468     op_diagnose = opcodes.OpDiagnoseOS(output_fields=["name", "valid",
 469                                                       "variants"], names=[])
 470     result = self.ExecOp(True, op_diagnose)
 471
 472     if not result:
 473       Err("Can't get the OS list")
 474
 475     found = False
 476     for (name, valid, variants) in result:
 477       if valid and self.opts.os in cli.CalculateOSNames(name, variants):
 478         found = True
 479         break
 480
 481     if not found:
 482       Err("OS '%s' not found" % self.opts.os)
 483
 484   @_DoCheckInstances
 485   @_DoBatch(False)
 486   def BurnCreateInstances(self):
 487     """Create the given instances.
 488
 489     """
 490     self.to_rem = []
 491     mytor = izip(cycle(self.nodes),
 492                  islice(cycle(self.nodes), 1, None),
 493                  self.instances)
 494
 495     Log("Creating instances")
 496     for pnode, snode, instance in mytor:
 497       Log("instance %s" % instance, indent=1)
 498       if self.opts.iallocator:
 499         pnode = snode = None
 500         msg = "with iallocator %s" % self.opts.iallocator
 501       elif self.opts.disk_template not in constants.DTS_NET_MIRROR:
 502         snode = None
 503         msg = "on %s" % pnode
 504       else:
 505         msg = "on %s, %s" % (pnode, snode)
 506
 507       Log(msg, indent=2)
 508
 509       op = opcodes.OpCreateInstance(instance_name=instance,
 510                                     disks = [ {"size": size}
 511                                               for size in self.disk_size],
 512                                     disk_template=self.opts.disk_template,
 513                                     nics=self.opts.nics,
 514                                     mode=constants.INSTANCE_CREATE,
 515                                     os_type=self.opts.os,
 516                                     pnode=pnode,
 517                                     snode=snode,
 518                                     start=True,
 519                                     ip_check=self.opts.ip_check,
 520                                     name_check=self.opts.name_check,
 521                                     wait_for_sync=True,
 522                                     file_driver="loop",
 523                                     file_storage_dir=None,
 524                                     iallocator=self.opts.iallocator,
 525                                     beparams=self.bep,
 526                                     hvparams=self.hvp,
 527                                     )
 528
 529       self.ExecOrQueue(instance, op)
 530       self.to_rem.append(instance)
 531
 532   @_DoBatch(False)
 533   def BurnGrowDisks(self):
 534     """Grow both the os and the swap disks by the requested amount, if any."""
 535     Log("Growing disks")
 536     for instance in self.instances:
 537       Log("instance %s" % instance, indent=1)
 538       for idx, growth in enumerate(self.disk_growth):
 539         if growth > 0:
 540           op = opcodes.OpGrowDisk(instance_name=instance, disk=idx,
 541                                   amount=growth, wait_for_sync=True)
 542           Log("increase disk/%s by %s MB" % (idx, growth), indent=2)
 543           self.ExecOrQueue(instance, op)
 544
 545   @_DoBatch(True)
 546   def BurnReplaceDisks1D8(self):
 547     """Replace disks on primary and secondary for drbd8."""
 548     Log("Replacing disks on the same nodes")
 549     for instance in self.instances:
 550       Log("instance %s" % instance, indent=1)
 551       ops = []
 552       for mode in constants.REPLACE_DISK_SEC, constants.REPLACE_DISK_PRI:
 553         op = opcodes.OpReplaceDisks(instance_name=instance,
 554                                     mode=mode,
 555                                     disks=[i for i in range(self.disk_count)],
 556                                     early_release=self.opts.early_release)
 557         Log("run %s" % mode, indent=2)
 558         ops.append(op)
 559       self.ExecOrQueue(instance, *ops) # pylint: disable-msg=W0142
 560
 561   @_DoBatch(True)
 562   def BurnReplaceDisks2(self):
 563     """Replace secondary node."""
 564     Log("Changing the secondary node")
 565     mode = constants.REPLACE_DISK_CHG
 566
 567     mytor = izip(islice(cycle(self.nodes), 2, None),
 568                  self.instances)
 569     for tnode, instance in mytor:
 570       Log("instance %s" % instance, indent=1)
 571       if self.opts.iallocator:
 572         tnode = None
 573         msg = "with iallocator %s" % self.opts.iallocator
 574       else:
 575         msg = tnode
 576       op = opcodes.OpReplaceDisks(instance_name=instance,
 577                                   mode=mode,
 578                                   remote_node=tnode,
 579                                   iallocator=self.opts.iallocator,
 580                                   disks=[],
 581                                   early_release=self.opts.early_release)
 582       Log("run %s %s" % (mode, msg), indent=2)
 583       self.ExecOrQueue(instance, op)
 584
 585   @_DoCheckInstances
 586   @_DoBatch(False)
 587   def BurnFailover(self):
 588     """Failover the instances."""
 589     Log("Failing over instances")
 590     for instance in self.instances:
 591       Log("instance %s" % instance, indent=1)
 592       op = opcodes.OpFailoverInstance(instance_name=instance,
 593                                       ignore_consistency=False)
 594       self.ExecOrQueue(instance, op)
 595
 596   @_DoCheckInstances
 597   @_DoBatch(False)
 598   def BurnMove(self):
 599     """Move the instances."""
 600     Log("Moving instances")
 601     mytor = izip(islice(cycle(self.nodes), 1, None),
 602                  self.instances)
 603     for tnode, instance in mytor:
 604       Log("instance %s" % instance, indent=1)
 605       op = opcodes.OpMoveInstance(instance_name=instance,
 606                                   target_node=tnode)
 607       self.ExecOrQueue(instance, op)
 608
 609   @_DoBatch(False)
 610   def BurnMigrate(self):
 611     """Migrate the instances."""
 612     Log("Migrating instances")
 613     for instance in self.instances:
 614       Log("instance %s" % instance, indent=1)
 615       op1 = opcodes.OpMigrateInstance(instance_name=instance, live=True,
 616                                       cleanup=False)
 617
 618       op2 = opcodes.OpMigrateInstance(instance_name=instance, live=True,
 619                                       cleanup=True)
 620       Log("migration and migration cleanup", indent=2)
 621       self.ExecOrQueue(instance, op1, op2)
 622
 623   @_DoCheckInstances
 624   @_DoBatch(False)
 625   def BurnImportExport(self):
 626     """Export the instance, delete it, and import it back.
 627
 628     """
 629     Log("Exporting and re-importing instances")
 630     mytor = izip(cycle(self.nodes),
 631                  islice(cycle(self.nodes), 1, None),
 632                  islice(cycle(self.nodes), 2, None),
 633                  self.instances)
 634
 635     for pnode, snode, enode, instance in mytor:
 636       Log("instance %s" % instance, indent=1)
 637       # read the full name of the instance
 638       nam_op = opcodes.OpQueryInstances(output_fields=["name"],
 639                                         names=[instance], use_locking=True)
 640       full_name = self.ExecOp(False, nam_op)[0][0]
 641
 642       if self.opts.iallocator:
 643         pnode = snode = None
 644         import_log_msg = ("import from %s"
 645                           " with iallocator %s" %
 646                           (enode, self.opts.iallocator))
 647       elif self.opts.disk_template not in constants.DTS_NET_MIRROR:
 648         snode = None
 649         import_log_msg = ("import from %s to %s" %
 650                           (enode, pnode))
 651       else:
 652         import_log_msg = ("import from %s to %s, %s" %
 653                           (enode, pnode, snode))
 654
 655       exp_op = opcodes.OpExportInstance(instance_name=instance,
 656                                            target_node=enode,
 657                                            shutdown=True)
 658       rem_op = opcodes.OpRemoveInstance(instance_name=instance,
 659                                         ignore_failures=True)
 660       imp_dir = utils.PathJoin(constants.EXPORT_DIR, full_name)
 661       imp_op = opcodes.OpCreateInstance(instance_name=instance,
 662                                         disks = [ {"size": size}
 663                                                   for size in self.disk_size],
 664                                         disk_template=self.opts.disk_template,
 665                                         nics=self.opts.nics,
 666                                         mode=constants.INSTANCE_IMPORT,
 667                                         src_node=enode,
 668                                         src_path=imp_dir,
 669                                         pnode=pnode,
 670                                         snode=snode,
 671                                         start=True,
 672                                         ip_check=self.opts.ip_check,
 673                                         name_check=self.opts.name_check,
 674                                         wait_for_sync=True,
 675                                         file_storage_dir=None,
 676                                         file_driver="loop",
 677                                         iallocator=self.opts.iallocator,
 678                                         beparams=self.bep,
 679                                         hvparams=self.hvp,
 680                                         )
 681
 682       erem_op = opcodes.OpRemoveExport(instance_name=instance)
 683
 684       Log("export to node %s" % enode, indent=2)
 685       Log("remove instance", indent=2)
 686       Log(import_log_msg, indent=2)
 687       Log("remove export", indent=2)
 688       self.ExecOrQueue(instance, exp_op, rem_op, imp_op, erem_op)
 689
 690   @staticmethod
 691   def StopInstanceOp(instance):
 692     """Stop given instance."""
 693     return opcodes.OpShutdownInstance(instance_name=instance)
 694
 695   @staticmethod
 696   def StartInstanceOp(instance):
 697     """Start given instance."""
 698     return opcodes.OpStartupInstance(instance_name=instance, force=False)
 699
 700   @staticmethod
 701   def RenameInstanceOp(instance, instance_new):
 702     """Rename instance."""
 703     return opcodes.OpRenameInstance(instance_name=instance,
 704                                     new_name=instance_new)
 705
 706   @_DoCheckInstances
 707   @_DoBatch(True)
 708   def BurnStopStart(self):
 709     """Stop/start the instances."""
 710     Log("Stopping and starting instances")
 711     for instance in self.instances:
 712       Log("instance %s" % instance, indent=1)
 713       op1 = self.StopInstanceOp(instance)
 714       op2 = self.StartInstanceOp(instance)
 715       self.ExecOrQueue(instance, op1, op2)
 716
 717   @_DoBatch(False)
 718   def BurnRemove(self):
 719     """Remove the instances."""
 720     Log("Removing instances")
 721     for instance in self.to_rem:
 722       Log("instance %s" % instance, indent=1)
 723       op = opcodes.OpRemoveInstance(instance_name=instance,
 724                                     ignore_failures=True)
 725       self.ExecOrQueue(instance, op)
 726
 727   def BurnRename(self):
 728     """Rename the instances.
 729
 730     Note that this function will not execute in parallel, since we
 731     only have one target for rename.
 732
 733     """
 734     Log("Renaming instances")
 735     rename = self.opts.rename
 736     for instance in self.instances:
 737       Log("instance %s" % instance, indent=1)
 738       op_stop1 = self.StopInstanceOp(instance)
 739       op_stop2 = self.StopInstanceOp(rename)
 740       op_rename1 = self.RenameInstanceOp(instance, rename)
 741       op_rename2 = self.RenameInstanceOp(rename, instance)
 742       op_start1 = self.StartInstanceOp(rename)
 743       op_start2 = self.StartInstanceOp(instance)
 744       self.ExecOp(False, op_stop1, op_rename1, op_start1)
 745       self._CheckInstanceAlive(rename)
 746       self.ExecOp(False, op_stop2, op_rename2, op_start2)
 747       self._CheckInstanceAlive(instance)
 748
 749   @_DoCheckInstances
 750   @_DoBatch(True)
 751   def BurnReinstall(self):
 752     """Reinstall the instances."""
 753     Log("Reinstalling instances")
 754     for instance in self.instances:
 755       Log("instance %s" % instance, indent=1)
 756       op1 = self.StopInstanceOp(instance)
 757       op2 = opcodes.OpReinstallInstance(instance_name=instance)
 758       Log("reinstall without passing the OS", indent=2)
 759       op3 = opcodes.OpReinstallInstance(instance_name=instance,
 760                                         os_type=self.opts.os)
 761       Log("reinstall specifying the OS", indent=2)
 762       op4 = self.StartInstanceOp(instance)
 763       self.ExecOrQueue(instance, op1, op2, op3, op4)
 764
 765   @_DoCheckInstances
 766   @_DoBatch(True)
 767   def BurnReboot(self):
 768     """Reboot the instances."""
 769     Log("Rebooting instances")
 770     for instance in self.instances:
 771       Log("instance %s" % instance, indent=1)
 772       ops = []
 773       for reboot_type in constants.REBOOT_TYPES:
 774         op = opcodes.OpRebootInstance(instance_name=instance,
 775                                       reboot_type=reboot_type,
 776                                       ignore_secondaries=False)
 777         Log("reboot with type '%s'" % reboot_type, indent=2)
 778         ops.append(op)
 779       self.ExecOrQueue(instance, *ops) # pylint: disable-msg=W0142
 780
 781   @_DoCheckInstances
 782   @_DoBatch(True)
 783   def BurnActivateDisks(self):
 784     """Activate and deactivate disks of the instances."""
 785     Log("Activating/deactivating disks")
 786     for instance in self.instances:
 787       Log("instance %s" % instance, indent=1)
 788       op_start = self.StartInstanceOp(instance)
 789       op_act = opcodes.OpActivateInstanceDisks(instance_name=instance)
 790       op_deact = opcodes.OpDeactivateInstanceDisks(instance_name=instance)
 791       op_stop = self.StopInstanceOp(instance)
 792       Log("activate disks when online", indent=2)
 793       Log("activate disks when offline", indent=2)
 794       Log("deactivate disks (when offline)", indent=2)
 795       self.ExecOrQueue(instance, op_act, op_stop, op_act, op_deact, op_start)
 796
 797   @_DoCheckInstances
 798   @_DoBatch(False)
 799   def BurnAddRemoveDisks(self):
 800     """Add and remove an extra disk for the instances."""
 801     Log("Adding and removing disks")
 802     for instance in self.instances:
 803       Log("instance %s" % instance, indent=1)
 804       op_add = opcodes.OpSetInstanceParams(\
 805         instance_name=instance,
 806         disks=[(constants.DDM_ADD, {"size": self.disk_size[0]})])
 807       op_rem = opcodes.OpSetInstanceParams(\
 808         instance_name=instance, disks=[(constants.DDM_REMOVE, {})])
 809       op_stop = self.StopInstanceOp(instance)
 810       op_start = self.StartInstanceOp(instance)
 811       Log("adding a disk", indent=2)
 812       Log("removing last disk", indent=2)
 813       self.ExecOrQueue(instance, op_add, op_stop, op_rem, op_start)
 814
 815   @_DoBatch(False)
 816   def BurnAddRemoveNICs(self):
 817     """Add and remove an extra NIC for the instances."""
 818     Log("Adding and removing NICs")
 819     for instance in self.instances:
 820       Log("instance %s" % instance, indent=1)
 821       op_add = opcodes.OpSetInstanceParams(\
 822         instance_name=instance, nics=[(constants.DDM_ADD, {})])
 823       op_rem = opcodes.OpSetInstanceParams(\
 824         instance_name=instance, nics=[(constants.DDM_REMOVE, {})])
 825       Log("adding a NIC", indent=2)
 826       Log("removing last NIC", indent=2)
 827       self.ExecOrQueue(instance, op_add, op_rem)
 828
 829   def _CheckInstanceAlive(self, instance):
 830     """Check if an instance is alive by doing http checks.
 831
 832     This will try to retrieve the url on the instance /hostname.txt
 833     and check that it contains the hostname of the instance. In case
 834     we get ECONNREFUSED, we retry up to the net timeout seconds, for
 835     any other error we abort.
 836
 837     """
 838     if not self.opts.http_check:
 839       return
 840     end_time = time.time() + self.opts.net_timeout
 841     url = None
 842     while time.time() < end_time and url is None:
 843       try:
 844         url = self.url_opener.open("http://%s/hostname.txt" % instance)
 845       except IOError:
 846         # here we can have connection refused, no route to host, etc.
 847         time.sleep(1)
 848     if url is None:
 849       raise InstanceDown(instance, "Cannot contact instance")
 850     hostname = url.read().strip()
 851     url.close()
 852     if hostname != instance:
 853       raise InstanceDown(instance, ("Hostname mismatch, expected %s, got %s" %
 854                                     (instance, hostname)))
 855
 856   def BurninCluster(self):
 857     """Test a cluster intensively.
 858
 859     This will create instances and then start/stop/failover them.
 860     It is safe for existing instances but could impact performance.
 861
 862     """
 863
 864     opts = self.opts
 865
 866     Log("Testing global parameters")
 867
 868     if (len(self.nodes) == 1 and
 869         opts.disk_template not in (constants.DT_DISKLESS, constants.DT_PLAIN,
 870                                    constants.DT_FILE)):
 871       Err("When one node is available/selected the disk template must"
 872           " be 'diskless', 'file' or 'plain'")
 873
 874     has_err = True
 875     try:
 876       self.BurnCreateInstances()
 877       if opts.do_replace1 and opts.disk_template in constants.DTS_NET_MIRROR:
 878         self.BurnReplaceDisks1D8()
 879       if (opts.do_replace2 and len(self.nodes) > 2 and
 880           opts.disk_template in constants.DTS_NET_MIRROR) :
 881         self.BurnReplaceDisks2()
 882
 883       if (opts.disk_template != constants.DT_DISKLESS and
 884           utils.any(self.disk_growth, lambda n: n > 0)):
 885         self.BurnGrowDisks()
 886
 887       if opts.do_failover and opts.disk_template in constants.DTS_NET_MIRROR:
 888         self.BurnFailover()
 889
 890       if opts.do_migrate and opts.disk_template == constants.DT_DRBD8:
 891         self.BurnMigrate()
 892
 893       if (opts.do_move and len(self.nodes) > 1 and
 894           opts.disk_template in [constants.DT_PLAIN, constants.DT_FILE]):
 895         self.BurnMove()
 896
 897       if (opts.do_importexport and
 898           opts.disk_template not in (constants.DT_DISKLESS,
 899                                      constants.DT_FILE)):
 900         self.BurnImportExport()
 901
 902       if opts.do_reinstall:
 903         self.BurnReinstall()
 904
 905       if opts.do_reboot:
 906         self.BurnReboot()
 907
 908       if opts.do_addremove_disks:
 909         self.BurnAddRemoveDisks()
 910
 911       if opts.do_addremove_nics:
 912         self.BurnAddRemoveNICs()
 913
 914       if opts.do_activate_disks:
 915         self.BurnActivateDisks()
 916
 917       if opts.rename:
 918         self.BurnRename()
 919
 920       if opts.do_startstop:
 921         self.BurnStopStart()
 922
 923       has_err = False
 924     finally:
 925       if has_err:
 926         Log("Error detected: opcode buffer follows:\n\n")
 927         Log(self.GetFeedbackBuf())
 928         Log("\n\n")
 929       if not self.opts.keep_instances:
 930         try:
 931           self.BurnRemove()
 932         except Exception, err:  # pylint: disable-msg=W0703
 933           if has_err: # already detected errors, so errors in removal
 934                       # are quite expected
 935             Log("Note: error detected during instance remove: %s" % str(err))
 936           else: # non-expected error
 937             raise
 938
 939     return 0
 940
 941
 942 def main():
 943   """Main function"""
 944
 945   burner = Burner()
 946   return burner.BurninCluster()
 947
 948
 949 if __name__ == "__main__":
 950   main()