Statistics
| Branch: | Tag: | Revision:

root / tools / burnin @ 265e6244

History | View | Annotate | Download (33.3 kB)

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 os
27
import sys
28
import optparse
29
import time
30
import socket
31
import urllib
32
from itertools import izip, islice, cycle
33
from cStringIO import StringIO
34

    
35
from ganeti import opcodes
36
from ganeti import constants
37
from ganeti import cli
38
from ganeti import errors
39
from ganeti import utils
40

    
41

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

    
44
MAX_RETRIES = 3
45

    
46
class InstanceDown(Exception):
47
  """The checked instance was not up"""
48

    
49

    
50
class BurninFailure(Exception):
51
  """Failure detected during burning"""
52

    
53

    
54
def Usage():
55
  """Shows program usage information and exits the program."""
56

    
57
  print >> sys.stderr, "Usage:"
58
  print >> sys.stderr, USAGE
59
  sys.exit(2)
60

    
61

    
62
def Log(msg, indent=0):
63
  """Simple function that prints out its argument.
64

    
65
  """
66
  headers = {
67
    0: "- ",
68
    1: "* ",
69
    2: ""
70
    }
71
  sys.stdout.write("%*s%s%s\n" % (2*indent, "",
72
                                   headers.get(indent, "  "), msg))
73
  sys.stdout.flush()
74

    
75
def Err(msg, exit_code=1):
76
  """Simple error logging that prints to stderr.
77

    
78
  """
79
  sys.stderr.write(msg + "\n")
80
  sys.stderr.flush()
81
  sys.exit(exit_code)
82

    
83

    
84
class SimpleOpener(urllib.FancyURLopener):
85
  """A simple url opener"""
86
  # pylint: disable-msg=W0221
87

    
88
  def prompt_user_passwd(self, host, realm, clear_cache=0):
89
    """No-interaction version of prompt_user_passwd."""
90
    # we follow parent class' API
91
    # pylint: disable-msg=W0613
92
    return None, None
93

    
94
  def http_error_default(self, url, fp, errcode, errmsg, headers):
95
    """Custom error handling"""
96
    # make sure sockets are not left in CLOSE_WAIT, this is similar
97
    # but with a different exception to the BasicURLOpener class
98
    _ = fp.read() # throw away data
99
    fp.close()
100
    raise InstanceDown("HTTP error returned: code %s, msg %s" %
101
                       (errcode, errmsg))
102

    
103

    
104
OPTIONS = [
105
  cli.cli_option("-o", "--os", dest="os", default=None,
106
                 help="OS to use during burnin",
107
                 metavar="<OS>",
108
                 completion_suggest=cli.OPT_COMPL_ONE_OS),
109
  cli.cli_option("--disk-size", dest="disk_size",
110
                 help="Disk size (determines disk count)",
111
                 default="128m", type="string", metavar="<size,size,...>",
112
                 completion_suggest=("128M 512M 1G 4G 1G,256M"
113
                                     " 4G,1G,1G 10G").split()),
114
  cli.cli_option("--disk-growth", dest="disk_growth", help="Disk growth",
115
                 default="128m", type="string", metavar="<size,size,...>"),
116
  cli.cli_option("--mem-size", dest="mem_size", help="Memory size",
117
                 default=128, type="unit", metavar="<size>",
118
                 completion_suggest=("128M 256M 512M 1G 4G 8G"
119
                                     " 12G 16G").split()),
120
  cli.DEBUG_OPT,
121
  cli.VERBOSE_OPT,
122
  cli.NOIPCHECK_OPT,
123
  cli.NONAMECHECK_OPT,
124
  cli.EARLY_RELEASE_OPT,
125
  cli.cli_option("--no-replace1", dest="do_replace1",
126
                 help="Skip disk replacement with the same secondary",
127
                 action="store_false", default=True),
128
  cli.cli_option("--no-replace2", dest="do_replace2",
129
                 help="Skip disk replacement with a different secondary",
130
                 action="store_false", default=True),
131
  cli.cli_option("--no-failover", dest="do_failover",
132
                 help="Skip instance failovers", action="store_false",
133
                 default=True),
134
  cli.cli_option("--no-migrate", dest="do_migrate",
135
                 help="Skip instance live migration",
136
                 action="store_false", default=True),
137
  cli.cli_option("--no-move", dest="do_move",
138
                 help="Skip instance moves", action="store_false",
139
                 default=True),
140
  cli.cli_option("--no-importexport", dest="do_importexport",
141
                 help="Skip instance export/import", action="store_false",
142
                 default=True),
143
  cli.cli_option("--no-startstop", dest="do_startstop",
144
                 help="Skip instance stop/start", action="store_false",
145
                 default=True),
146
  cli.cli_option("--no-reinstall", dest="do_reinstall",
147
                 help="Skip instance reinstall", action="store_false",
148
                 default=True),
149
  cli.cli_option("--no-reboot", dest="do_reboot",
150
                 help="Skip instance reboot", action="store_false",
151
                 default=True),
152
  cli.cli_option("--no-activate-disks", dest="do_activate_disks",
153
                 help="Skip disk activation/deactivation",
154
                 action="store_false", default=True),
155
  cli.cli_option("--no-add-disks", dest="do_addremove_disks",
156
                 help="Skip disk addition/removal",
157
                 action="store_false", default=True),
158
  cli.cli_option("--no-add-nics", dest="do_addremove_nics",
159
                 help="Skip NIC addition/removal",
160
                 action="store_false", default=True),
161
  cli.cli_option("--no-nics", dest="nics",
162
                 help="No network interfaces", action="store_const",
163
                 const=[], default=[{}]),
164
  cli.cli_option("--rename", dest="rename", default=None,
165
                 help=("Give one unused instance name which is taken"
166
                       " to start the renaming sequence"),
167
                 metavar="<instance_name>"),
168
  cli.cli_option("-t", "--disk-template", dest="disk_template",
169
                 choices=list(constants.DISK_TEMPLATES),
170
                 default=constants.DT_DRBD8,
171
                 help="Disk template (diskless, file, plain or drbd) [drbd]"),
172
  cli.cli_option("-n", "--nodes", dest="nodes", default="",
173
                 help=("Comma separated list of nodes to perform"
174
                       " the burnin on (defaults to all nodes)"),
175
                 completion_suggest=cli.OPT_COMPL_MANY_NODES),
176
  cli.cli_option("-I", "--iallocator", dest="iallocator",
177
                 default=None, type="string",
178
                 help=("Perform the allocation using an iallocator"
179
                       " instead of fixed node spread (node restrictions no"
180
                       " longer apply, therefore -n/--nodes must not be"
181
                       " used"),
182
                 completion_suggest=cli.OPT_COMPL_ONE_IALLOCATOR),
183
  cli.cli_option("-p", "--parallel", default=False, action="store_true",
184
                 dest="parallel",
185
                 help=("Enable parallelization of some operations in"
186
                       " order to speed burnin or to test granular locking")),
187
  cli.cli_option("--net-timeout", default=15, type="int",
188
                 dest="net_timeout",
189
                 help=("The instance check network timeout in seconds"
190
                       " (defaults to 15 seconds)"),
191
                 completion_suggest="15 60 300 900".split()),
192
  cli.cli_option("-C", "--http-check", default=False, action="store_true",
193
                 dest="http_check",
194
                 help=("Enable checking of instance status via http,"
195
                       " looking for /hostname.txt that should contain the"
196
                       " name of the instance")),
197
  cli.cli_option("-K", "--keep-instances", default=False,
198
                 action="store_true",
199
                 dest="keep_instances",
200
                 help=("Leave instances on the cluster after burnin,"
201
                       " for investigation in case of errors or simply"
202
                       " to use them")),
203
  ]
204

    
205
# Mainly used for bash completion
206
ARGUMENTS = [cli.ArgInstance(min=1)]
207

    
208

    
209
def _DoCheckInstances(fn):
210
  """Decorator for checking instances.
211

    
212
  """
213
  def wrapper(self, *args, **kwargs):
214
    val = fn(self, *args, **kwargs)
215
    for instance in self.instances:
216
      self._CheckInstanceAlive(instance) # pylint: disable-msg=W0212
217
    return val
218

    
219
  return wrapper
220

    
221

    
222
def _DoBatch(retry):
223
  """Decorator for possible batch operations.
224

    
225
  Must come after the _DoCheckInstances decorator (if any).
226

    
227
  @param retry: whether this is a retryable batch, will be
228
      passed to StartBatch
229

    
230
  """
231
  def wrap(fn):
232
    def batched(self, *args, **kwargs):
233
      self.StartBatch(retry)
234
      val = fn(self, *args, **kwargs)
235
      self.CommitQueue()
236
      return val
237
    return batched
238

    
239
  return wrap
240

    
241

    
242
class Burner(object):
243
  """Burner class."""
244

    
245
  def __init__(self):
246
    """Constructor."""
247
    utils.SetupLogging(constants.LOG_BURNIN, debug=False, stderr_logging=True)
248
    self.url_opener = SimpleOpener()
249
    self._feed_buf = StringIO()
250
    self.nodes = []
251
    self.instances = []
252
    self.to_rem = []
253
    self.queued_ops = []
254
    self.opts = None
255
    self.queue_retry = False
256
    self.disk_count = self.disk_growth = self.disk_size = None
257
    self.hvp = self.bep = None
258
    self.ParseOptions()
259
    self.cl = cli.GetClient()
260
    self.GetState()
261

    
262
  def ClearFeedbackBuf(self):
263
    """Clear the feedback buffer."""
264
    self._feed_buf.truncate(0)
265

    
266
  def GetFeedbackBuf(self):
267
    """Return the contents of the buffer."""
268
    return self._feed_buf.getvalue()
269

    
270
  def Feedback(self, msg):
271
    """Acumulate feedback in our buffer."""
272
    formatted_msg = "%s %s" % (time.ctime(utils.MergeTime(msg[0])), msg[2])
273
    self._feed_buf.write(formatted_msg + "\n")
274
    if self.opts.verbose:
275
      Log(formatted_msg, indent=3)
276

    
277
  def MaybeRetry(self, retry_count, msg, fn, *args):
278
    """Possibly retry a given function execution.
279

    
280
    @type retry_count: int
281
    @param retry_count: retry counter:
282
        - 0: non-retryable action
283
        - 1: last retry for a retryable action
284
        - MAX_RETRIES: original try for a retryable action
285
    @type msg: str
286
    @param msg: the kind of the operation
287
    @type fn: callable
288
    @param fn: the function to be called
289

    
290
    """
291
    try:
292
      val = fn(*args)
293
      if retry_count > 0 and retry_count < MAX_RETRIES:
294
        Log("Idempotent %s succeeded after %d retries" %
295
            (msg, MAX_RETRIES - retry_count))
296
      return val
297
    except Exception, err: # pylint: disable-msg=W0703
298
      if retry_count == 0:
299
        Log("Non-idempotent %s failed, aborting" % (msg, ))
300
        raise
301
      elif retry_count == 1:
302
        Log("Idempotent %s repeated failure, aborting" % (msg, ))
303
        raise
304
      else:
305
        Log("Idempotent %s failed, retry #%d/%d: %s" %
306
            (msg, MAX_RETRIES - retry_count + 1, MAX_RETRIES, err))
307
        self.MaybeRetry(retry_count - 1, msg, fn, *args)
308

    
309
  def _SetDebug(self, ops):
310
    """Set the debug value on the given opcodes"""
311
    for op in ops:
312
      op.debug_level = self.opts.debug
313

    
314
  def _ExecOp(self, *ops):
315
    """Execute one or more opcodes and manage the exec buffer.
316

    
317
    @result: if only opcode has been passed, we return its result;
318
        otherwise we return the list of results
319

    
320
    """
321
    job_id = cli.SendJob(ops, cl=self.cl)
322
    results = cli.PollJob(job_id, cl=self.cl, feedback_fn=self.Feedback)
323
    if len(ops) == 1:
324
      return results[0]
325
    else:
326
      return results
327

    
328
  def ExecOp(self, retry, *ops):
329
    """Execute one or more opcodes and manage the exec buffer.
330

    
331
    @result: if only opcode has been passed, we return its result;
332
        otherwise we return the list of results
333

    
334
    """
335
    if retry:
336
      rval = MAX_RETRIES
337
    else:
338
      rval = 0
339
    self._SetDebug(ops)
340
    return self.MaybeRetry(rval, "opcode", self._ExecOp, *ops)
341

    
342
  def ExecOrQueue(self, name, *ops):
343
    """Execute an opcode and manage the exec buffer."""
344
    if self.opts.parallel:
345
      self._SetDebug(ops)
346
      self.queued_ops.append((ops, name))
347
    else:
348
      return self.ExecOp(self.queue_retry, *ops)
349

    
350
  def StartBatch(self, retry):
351
    """Start a new batch of jobs.
352

    
353
    @param retry: whether this is a retryable batch
354

    
355
    """
356
    self.queued_ops = []
357
    self.queue_retry = retry
358

    
359
  def CommitQueue(self):
360
    """Execute all submitted opcodes in case of parallel burnin"""
361
    if not self.opts.parallel:
362
      return
363

    
364
    if self.queue_retry:
365
      rval = MAX_RETRIES
366
    else:
367
      rval = 0
368

    
369
    try:
370
      results = self.MaybeRetry(rval, "jobset", self.ExecJobSet,
371
                                self.queued_ops)
372
    finally:
373
      self.queued_ops = []
374
    return results
375

    
376
  def ExecJobSet(self, jobs):
377
    """Execute a set of jobs and return once all are done.
378

    
379
    The method will return the list of results, if all jobs are
380
    successful. Otherwise, OpExecError will be raised from within
381
    cli.py.
382

    
383
    """
384
    self.ClearFeedbackBuf()
385
    job_ids = [cli.SendJob(row[0], cl=self.cl) for row in jobs]
386
    Log("Submitted job ID(s) %s" % utils.CommaJoin(job_ids), indent=1)
387
    results = []
388
    for jid, (_, iname) in zip(job_ids, jobs):
389
      Log("waiting for job %s for %s" % (jid, iname), indent=2)
390
      try:
391
        results.append(cli.PollJob(jid, cl=self.cl, feedback_fn=self.Feedback))
392
      except Exception, err: # pylint: disable-msg=W0703
393
        Log("Job for %s failed: %s" % (iname, err))
394
    if len(results) != len(jobs):
395
      raise BurninFailure()
396
    return results
397

    
398
  def ParseOptions(self):
399
    """Parses the command line options.
400

    
401
    In case of command line errors, it will show the usage and exit the
402
    program.
403

    
404
    """
405
    parser = optparse.OptionParser(usage="\n%s" % USAGE,
406
                                   version=("%%prog (ganeti) %s" %
407
                                            constants.RELEASE_VERSION),
408
                                   option_list=OPTIONS)
409

    
410
    options, args = parser.parse_args()
411
    if len(args) < 1 or options.os is None:
412
      Usage()
413

    
414
    supported_disk_templates = (constants.DT_DISKLESS,
415
                                constants.DT_FILE,
416
                                constants.DT_PLAIN,
417
                                constants.DT_DRBD8)
418
    if options.disk_template not in supported_disk_templates:
419
      Err("Unknown disk template '%s'" % options.disk_template)
420

    
421
    if options.disk_template == constants.DT_DISKLESS:
422
      disk_size = disk_growth = []
423
      options.do_addremove_disks = False
424
    else:
425
      disk_size = [utils.ParseUnit(v) for v in options.disk_size.split(",")]
426
      disk_growth = [utils.ParseUnit(v)
427
                     for v in options.disk_growth.split(",")]
428
      if len(disk_growth) != len(disk_size):
429
        Err("Wrong disk sizes/growth combination")
430
    if ((disk_size and options.disk_template == constants.DT_DISKLESS) or
431
        (not disk_size and options.disk_template != constants.DT_DISKLESS)):
432
      Err("Wrong disk count/disk template combination")
433

    
434
    self.disk_size = disk_size
435
    self.disk_growth = disk_growth
436
    self.disk_count = len(disk_size)
437

    
438
    if options.nodes and options.iallocator:
439
      Err("Give either the nodes option or the iallocator option, not both")
440

    
441
    if options.http_check and not options.name_check:
442
      Err("Can't enable HTTP checks without name checks")
443

    
444
    self.opts = options
445
    self.instances = args
446
    self.bep = {
447
      constants.BE_MEMORY: options.mem_size,
448
      constants.BE_VCPUS: 1,
449
      }
450
    self.hvp = {}
451

    
452
    socket.setdefaulttimeout(options.net_timeout)
453

    
454
  def GetState(self):
455
    """Read the cluster state from the config."""
456
    if self.opts.nodes:
457
      names = self.opts.nodes.split(",")
458
    else:
459
      names = []
460
    try:
461
      op = opcodes.OpQueryNodes(output_fields=["name", "offline", "drained"],
462
                                names=names, use_locking=True)
463
      result = self.ExecOp(True, op)
464
    except errors.GenericError, err:
465
      err_code, msg = cli.FormatError(err)
466
      Err(msg, exit_code=err_code)
467
    self.nodes = [data[0] for data in result if not (data[1] or data[2])]
468

    
469
    op_diagnose = opcodes.OpDiagnoseOS(output_fields=["name", "valid",
470
                                                      "variants"], names=[])
471
    result = self.ExecOp(True, op_diagnose)
472

    
473
    if not result:
474
      Err("Can't get the OS list")
475

    
476
    found = False
477
    for (name, valid, variants) in result:
478
      if valid and self.opts.os in cli.CalculateOSNames(name, variants):
479
        found = True
480
        break
481

    
482
    if not found:
483
      Err("OS '%s' not found" % self.opts.os)
484

    
485
  @_DoCheckInstances
486
  @_DoBatch(False)
487
  def BurnCreateInstances(self):
488
    """Create the given instances.
489

    
490
    """
491
    self.to_rem = []
492
    mytor = izip(cycle(self.nodes),
493
                 islice(cycle(self.nodes), 1, None),
494
                 self.instances)
495

    
496
    Log("Creating instances")
497
    for pnode, snode, instance in mytor:
498
      Log("instance %s" % instance, indent=1)
499
      if self.opts.iallocator:
500
        pnode = snode = None
501
        msg = "with iallocator %s" % self.opts.iallocator
502
      elif self.opts.disk_template not in constants.DTS_NET_MIRROR:
503
        snode = None
504
        msg = "on %s" % pnode
505
      else:
506
        msg = "on %s, %s" % (pnode, snode)
507

    
508
      Log(msg, indent=2)
509

    
510
      op = opcodes.OpCreateInstance(instance_name=instance,
511
                                    disks = [ {"size": size}
512
                                              for size in self.disk_size],
513
                                    disk_template=self.opts.disk_template,
514
                                    nics=self.opts.nics,
515
                                    mode=constants.INSTANCE_CREATE,
516
                                    os_type=self.opts.os,
517
                                    pnode=pnode,
518
                                    snode=snode,
519
                                    start=True,
520
                                    ip_check=self.opts.ip_check,
521
                                    name_check=self.opts.name_check,
522
                                    wait_for_sync=True,
523
                                    file_driver="loop",
524
                                    file_storage_dir=None,
525
                                    iallocator=self.opts.iallocator,
526
                                    beparams=self.bep,
527
                                    hvparams=self.hvp,
528
                                    )
529

    
530
      self.ExecOrQueue(instance, op)
531
      self.to_rem.append(instance)
532

    
533
  @_DoBatch(False)
534
  def BurnGrowDisks(self):
535
    """Grow both the os and the swap disks by the requested amount, if any."""
536
    Log("Growing disks")
537
    for instance in self.instances:
538
      Log("instance %s" % instance, indent=1)
539
      for idx, growth in enumerate(self.disk_growth):
540
        if growth > 0:
541
          op = opcodes.OpGrowDisk(instance_name=instance, disk=idx,
542
                                  amount=growth, wait_for_sync=True)
543
          Log("increase disk/%s by %s MB" % (idx, growth), indent=2)
544
          self.ExecOrQueue(instance, op)
545

    
546
  @_DoBatch(True)
547
  def BurnReplaceDisks1D8(self):
548
    """Replace disks on primary and secondary for drbd8."""
549
    Log("Replacing disks on the same nodes")
550
    for instance in self.instances:
551
      Log("instance %s" % instance, indent=1)
552
      ops = []
553
      for mode in constants.REPLACE_DISK_SEC, constants.REPLACE_DISK_PRI:
554
        op = opcodes.OpReplaceDisks(instance_name=instance,
555
                                    mode=mode,
556
                                    disks=[i for i in range(self.disk_count)],
557
                                    early_release=self.opts.early_release)
558
        Log("run %s" % mode, indent=2)
559
        ops.append(op)
560
      self.ExecOrQueue(instance, *ops) # pylint: disable-msg=W0142
561

    
562
  @_DoBatch(True)
563
  def BurnReplaceDisks2(self):
564
    """Replace secondary node."""
565
    Log("Changing the secondary node")
566
    mode = constants.REPLACE_DISK_CHG
567

    
568
    mytor = izip(islice(cycle(self.nodes), 2, None),
569
                 self.instances)
570
    for tnode, instance in mytor:
571
      Log("instance %s" % instance, indent=1)
572
      if self.opts.iallocator:
573
        tnode = None
574
        msg = "with iallocator %s" % self.opts.iallocator
575
      else:
576
        msg = tnode
577
      op = opcodes.OpReplaceDisks(instance_name=instance,
578
                                  mode=mode,
579
                                  remote_node=tnode,
580
                                  iallocator=self.opts.iallocator,
581
                                  disks=[],
582
                                  early_release=self.opts.early_release)
583
      Log("run %s %s" % (mode, msg), indent=2)
584
      self.ExecOrQueue(instance, op)
585

    
586
  @_DoCheckInstances
587
  @_DoBatch(False)
588
  def BurnFailover(self):
589
    """Failover the instances."""
590
    Log("Failing over instances")
591
    for instance in self.instances:
592
      Log("instance %s" % instance, indent=1)
593
      op = opcodes.OpFailoverInstance(instance_name=instance,
594
                                      ignore_consistency=False)
595
      self.ExecOrQueue(instance, op)
596

    
597
  @_DoCheckInstances
598
  @_DoBatch(False)
599
  def BurnMove(self):
600
    """Move the instances."""
601
    Log("Moving instances")
602
    mytor = izip(islice(cycle(self.nodes), 1, None),
603
                 self.instances)
604
    for tnode, instance in mytor:
605
      Log("instance %s" % instance, indent=1)
606
      op = opcodes.OpMoveInstance(instance_name=instance,
607
                                  target_node=tnode)
608
      self.ExecOrQueue(instance, op)
609

    
610
  @_DoBatch(False)
611
  def BurnMigrate(self):
612
    """Migrate the instances."""
613
    Log("Migrating instances")
614
    for instance in self.instances:
615
      Log("instance %s" % instance, indent=1)
616
      op1 = opcodes.OpMigrateInstance(instance_name=instance, live=True,
617
                                      cleanup=False)
618

    
619
      op2 = opcodes.OpMigrateInstance(instance_name=instance, live=True,
620
                                      cleanup=True)
621
      Log("migration and migration cleanup", indent=2)
622
      self.ExecOrQueue(instance, op1, op2)
623

    
624
  @_DoCheckInstances
625
  @_DoBatch(False)
626
  def BurnImportExport(self):
627
    """Export the instance, delete it, and import it back.
628

    
629
    """
630
    Log("Exporting and re-importing instances")
631
    mytor = izip(cycle(self.nodes),
632
                 islice(cycle(self.nodes), 1, None),
633
                 islice(cycle(self.nodes), 2, None),
634
                 self.instances)
635

    
636
    for pnode, snode, enode, instance in mytor:
637
      Log("instance %s" % instance, indent=1)
638
      # read the full name of the instance
639
      nam_op = opcodes.OpQueryInstances(output_fields=["name"],
640
                                        names=[instance], use_locking=True)
641
      full_name = self.ExecOp(False, nam_op)[0][0]
642

    
643
      if self.opts.iallocator:
644
        pnode = snode = None
645
        import_log_msg = ("import from %s"
646
                          " with iallocator %s" %
647
                          (enode, self.opts.iallocator))
648
      elif self.opts.disk_template not in constants.DTS_NET_MIRROR:
649
        snode = None
650
        import_log_msg = ("import from %s to %s" %
651
                          (enode, pnode))
652
      else:
653
        import_log_msg = ("import from %s to %s, %s" %
654
                          (enode, pnode, snode))
655

    
656
      exp_op = opcodes.OpExportInstance(instance_name=instance,
657
                                           target_node=enode,
658
                                           shutdown=True)
659
      rem_op = opcodes.OpRemoveInstance(instance_name=instance,
660
                                        ignore_failures=True)
661
      imp_dir = os.path.join(constants.EXPORT_DIR, full_name)
662
      imp_op = opcodes.OpCreateInstance(instance_name=instance,
663
                                        disks = [ {"size": size}
664
                                                  for size in self.disk_size],
665
                                        disk_template=self.opts.disk_template,
666
                                        nics=self.opts.nics,
667
                                        mode=constants.INSTANCE_IMPORT,
668
                                        src_node=enode,
669
                                        src_path=imp_dir,
670
                                        pnode=pnode,
671
                                        snode=snode,
672
                                        start=True,
673
                                        ip_check=self.opts.ip_check,
674
                                        name_check=self.opts.name_check,
675
                                        wait_for_sync=True,
676
                                        file_storage_dir=None,
677
                                        file_driver="loop",
678
                                        iallocator=self.opts.iallocator,
679
                                        beparams=self.bep,
680
                                        hvparams=self.hvp,
681
                                        )
682

    
683
      erem_op = opcodes.OpRemoveExport(instance_name=instance)
684

    
685
      Log("export to node %s" % enode, indent=2)
686
      Log("remove instance", indent=2)
687
      Log(import_log_msg, indent=2)
688
      Log("remove export", indent=2)
689
      self.ExecOrQueue(instance, exp_op, rem_op, imp_op, erem_op)
690

    
691
  @staticmethod
692
  def StopInstanceOp(instance):
693
    """Stop given instance."""
694
    return opcodes.OpShutdownInstance(instance_name=instance)
695

    
696
  @staticmethod
697
  def StartInstanceOp(instance):
698
    """Start given instance."""
699
    return opcodes.OpStartupInstance(instance_name=instance, force=False)
700

    
701
  @staticmethod
702
  def RenameInstanceOp(instance, instance_new):
703
    """Rename instance."""
704
    return opcodes.OpRenameInstance(instance_name=instance,
705
                                    new_name=instance_new)
706

    
707
  @_DoCheckInstances
708
  @_DoBatch(True)
709
  def BurnStopStart(self):
710
    """Stop/start the instances."""
711
    Log("Stopping and starting instances")
712
    for instance in self.instances:
713
      Log("instance %s" % instance, indent=1)
714
      op1 = self.StopInstanceOp(instance)
715
      op2 = self.StartInstanceOp(instance)
716
      self.ExecOrQueue(instance, op1, op2)
717

    
718
  @_DoBatch(False)
719
  def BurnRemove(self):
720
    """Remove the instances."""
721
    Log("Removing instances")
722
    for instance in self.to_rem:
723
      Log("instance %s" % instance, indent=1)
724
      op = opcodes.OpRemoveInstance(instance_name=instance,
725
                                    ignore_failures=True)
726
      self.ExecOrQueue(instance, op)
727

    
728
  def BurnRename(self):
729
    """Rename the instances.
730

    
731
    Note that this function will not execute in parallel, since we
732
    only have one target for rename.
733

    
734
    """
735
    Log("Renaming instances")
736
    rename = self.opts.rename
737
    for instance in self.instances:
738
      Log("instance %s" % instance, indent=1)
739
      op_stop1 = self.StopInstanceOp(instance)
740
      op_stop2 = self.StopInstanceOp(rename)
741
      op_rename1 = self.RenameInstanceOp(instance, rename)
742
      op_rename2 = self.RenameInstanceOp(rename, instance)
743
      op_start1 = self.StartInstanceOp(rename)
744
      op_start2 = self.StartInstanceOp(instance)
745
      self.ExecOp(False, op_stop1, op_rename1, op_start1)
746
      self._CheckInstanceAlive(rename)
747
      self.ExecOp(False, op_stop2, op_rename2, op_start2)
748
      self._CheckInstanceAlive(instance)
749

    
750
  @_DoCheckInstances
751
  @_DoBatch(True)
752
  def BurnReinstall(self):
753
    """Reinstall the instances."""
754
    Log("Reinstalling instances")
755
    for instance in self.instances:
756
      Log("instance %s" % instance, indent=1)
757
      op1 = self.StopInstanceOp(instance)
758
      op2 = opcodes.OpReinstallInstance(instance_name=instance)
759
      Log("reinstall without passing the OS", indent=2)
760
      op3 = opcodes.OpReinstallInstance(instance_name=instance,
761
                                        os_type=self.opts.os)
762
      Log("reinstall specifying the OS", indent=2)
763
      op4 = self.StartInstanceOp(instance)
764
      self.ExecOrQueue(instance, op1, op2, op3, op4)
765

    
766
  @_DoCheckInstances
767
  @_DoBatch(True)
768
  def BurnReboot(self):
769
    """Reboot the instances."""
770
    Log("Rebooting instances")
771
    for instance in self.instances:
772
      Log("instance %s" % instance, indent=1)
773
      ops = []
774
      for reboot_type in constants.REBOOT_TYPES:
775
        op = opcodes.OpRebootInstance(instance_name=instance,
776
                                      reboot_type=reboot_type,
777
                                      ignore_secondaries=False)
778
        Log("reboot with type '%s'" % reboot_type, indent=2)
779
        ops.append(op)
780
      self.ExecOrQueue(instance, *ops) # pylint: disable-msg=W0142
781

    
782
  @_DoCheckInstances
783
  @_DoBatch(True)
784
  def BurnActivateDisks(self):
785
    """Activate and deactivate disks of the instances."""
786
    Log("Activating/deactivating disks")
787
    for instance in self.instances:
788
      Log("instance %s" % instance, indent=1)
789
      op_start = self.StartInstanceOp(instance)
790
      op_act = opcodes.OpActivateInstanceDisks(instance_name=instance)
791
      op_deact = opcodes.OpDeactivateInstanceDisks(instance_name=instance)
792
      op_stop = self.StopInstanceOp(instance)
793
      Log("activate disks when online", indent=2)
794
      Log("activate disks when offline", indent=2)
795
      Log("deactivate disks (when offline)", indent=2)
796
      self.ExecOrQueue(instance, op_act, op_stop, op_act, op_deact, op_start)
797

    
798
  @_DoCheckInstances
799
  @_DoBatch(False)
800
  def BurnAddRemoveDisks(self):
801
    """Add and remove an extra disk for the instances."""
802
    Log("Adding and removing disks")
803
    for instance in self.instances:
804
      Log("instance %s" % instance, indent=1)
805
      op_add = opcodes.OpSetInstanceParams(\
806
        instance_name=instance,
807
        disks=[(constants.DDM_ADD, {"size": self.disk_size[0]})])
808
      op_rem = opcodes.OpSetInstanceParams(\
809
        instance_name=instance, disks=[(constants.DDM_REMOVE, {})])
810
      op_stop = self.StopInstanceOp(instance)
811
      op_start = self.StartInstanceOp(instance)
812
      Log("adding a disk", indent=2)
813
      Log("removing last disk", indent=2)
814
      self.ExecOrQueue(instance, op_add, op_stop, op_rem, op_start)
815

    
816
  @_DoBatch(False)
817
  def BurnAddRemoveNICs(self):
818
    """Add and remove an extra NIC for the instances."""
819
    Log("Adding and removing NICs")
820
    for instance in self.instances:
821
      Log("instance %s" % instance, indent=1)
822
      op_add = opcodes.OpSetInstanceParams(\
823
        instance_name=instance, nics=[(constants.DDM_ADD, {})])
824
      op_rem = opcodes.OpSetInstanceParams(\
825
        instance_name=instance, nics=[(constants.DDM_REMOVE, {})])
826
      Log("adding a NIC", indent=2)
827
      Log("removing last NIC", indent=2)
828
      self.ExecOrQueue(instance, op_add, op_rem)
829

    
830
  def _CheckInstanceAlive(self, instance):
831
    """Check if an instance is alive by doing http checks.
832

    
833
    This will try to retrieve the url on the instance /hostname.txt
834
    and check that it contains the hostname of the instance. In case
835
    we get ECONNREFUSED, we retry up to the net timeout seconds, for
836
    any other error we abort.
837

    
838
    """
839
    if not self.opts.http_check:
840
      return
841
    end_time = time.time() + self.opts.net_timeout
842
    url = None
843
    while time.time() < end_time and url is None:
844
      try:
845
        url = self.url_opener.open("http://%s/hostname.txt" % instance)
846
      except IOError:
847
        # here we can have connection refused, no route to host, etc.
848
        time.sleep(1)
849
    if url is None:
850
      raise InstanceDown(instance, "Cannot contact instance")
851
    hostname = url.read().strip()
852
    url.close()
853
    if hostname != instance:
854
      raise InstanceDown(instance, ("Hostname mismatch, expected %s, got %s" %
855
                                    (instance, hostname)))
856

    
857
  def BurninCluster(self):
858
    """Test a cluster intensively.
859

    
860
    This will create instances and then start/stop/failover them.
861
    It is safe for existing instances but could impact performance.
862

    
863
    """
864

    
865
    opts = self.opts
866

    
867
    Log("Testing global parameters")
868

    
869
    if (len(self.nodes) == 1 and
870
        opts.disk_template not in (constants.DT_DISKLESS, constants.DT_PLAIN,
871
                                   constants.DT_FILE)):
872
      Err("When one node is available/selected the disk template must"
873
          " be 'diskless', 'file' or 'plain'")
874

    
875
    has_err = True
876
    try:
877
      self.BurnCreateInstances()
878
      if opts.do_replace1 and opts.disk_template in constants.DTS_NET_MIRROR:
879
        self.BurnReplaceDisks1D8()
880
      if (opts.do_replace2 and len(self.nodes) > 2 and
881
          opts.disk_template in constants.DTS_NET_MIRROR) :
882
        self.BurnReplaceDisks2()
883

    
884
      if (opts.disk_template != constants.DT_DISKLESS and
885
          utils.any(self.disk_growth, lambda n: n > 0)):
886
        self.BurnGrowDisks()
887

    
888
      if opts.do_failover and opts.disk_template in constants.DTS_NET_MIRROR:
889
        self.BurnFailover()
890

    
891
      if opts.do_migrate and opts.disk_template == constants.DT_DRBD8:
892
        self.BurnMigrate()
893

    
894
      if (opts.do_move and len(self.nodes) > 1 and
895
          opts.disk_template in [constants.DT_PLAIN, constants.DT_FILE]):
896
        self.BurnMove()
897

    
898
      if (opts.do_importexport and
899
          opts.disk_template not in (constants.DT_DISKLESS,
900
                                     constants.DT_FILE)):
901
        self.BurnImportExport()
902

    
903
      if opts.do_reinstall:
904
        self.BurnReinstall()
905

    
906
      if opts.do_reboot:
907
        self.BurnReboot()
908

    
909
      if opts.do_addremove_disks:
910
        self.BurnAddRemoveDisks()
911

    
912
      if opts.do_addremove_nics:
913
        self.BurnAddRemoveNICs()
914

    
915
      if opts.do_activate_disks:
916
        self.BurnActivateDisks()
917

    
918
      if opts.rename:
919
        self.BurnRename()
920

    
921
      if opts.do_startstop:
922
        self.BurnStopStart()
923

    
924
      has_err = False
925
    finally:
926
      if has_err:
927
        Log("Error detected: opcode buffer follows:\n\n")
928
        Log(self.GetFeedbackBuf())
929
        Log("\n\n")
930
      if not self.opts.keep_instances:
931
        try:
932
          self.BurnRemove()
933
        except Exception, err:  # pylint: disable-msg=W0703
934
          if has_err: # already detected errors, so errors in removal
935
                      # are quite expected
936
            Log("Note: error detected during instance remove: %s" % str(err))
937
          else: # non-expected error
938
            raise
939

    
940
    return 0
941

    
942

    
943
def main():
944
  """Main function"""
945

    
946
  burner = Burner()
947
  return burner.BurninCluster()
948

    
949

    
950
if __name__ == "__main__":
951
  main()