Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 82187135

History | View | Annotate | Download (309.6 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008 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
"""Module implementing the master-side code."""
23

    
24
# pylint: disable-msg=W0201
25

    
26
# W0201 since most LU attributes are defined in CheckPrereq or similar
27
# functions
28

    
29
import os
30
import os.path
31
import time
32
import re
33
import platform
34
import logging
35
import copy
36

    
37
from ganeti import ssh
38
from ganeti import utils
39
from ganeti import errors
40
from ganeti import hypervisor
41
from ganeti import locking
42
from ganeti import constants
43
from ganeti import objects
44
from ganeti import serializer
45
from ganeti import ssconf
46

    
47

    
48
class LogicalUnit(object):
49
  """Logical Unit base class.
50

51
  Subclasses must follow these rules:
52
    - implement ExpandNames
53
    - implement CheckPrereq (except when tasklets are used)
54
    - implement Exec (except when tasklets are used)
55
    - implement BuildHooksEnv
56
    - redefine HPATH and HTYPE
57
    - optionally redefine their run requirements:
58
        REQ_BGL: the LU needs to hold the Big Ganeti Lock exclusively
59

60
  Note that all commands require root permissions.
61

62
  @ivar dry_run_result: the value (if any) that will be returned to the caller
63
      in dry-run mode (signalled by opcode dry_run parameter)
64

65
  """
66
  HPATH = None
67
  HTYPE = None
68
  _OP_REQP = []
69
  REQ_BGL = True
70

    
71
  def __init__(self, processor, op, context, rpc):
72
    """Constructor for LogicalUnit.
73

74
    This needs to be overridden in derived classes in order to check op
75
    validity.
76

77
    """
78
    self.proc = processor
79
    self.op = op
80
    self.cfg = context.cfg
81
    self.context = context
82
    self.rpc = rpc
83
    # Dicts used to declare locking needs to mcpu
84
    self.needed_locks = None
85
    self.acquired_locks = {}
86
    self.share_locks = dict.fromkeys(locking.LEVELS, 0)
87
    self.add_locks = {}
88
    self.remove_locks = {}
89
    # Used to force good behavior when calling helper functions
90
    self.recalculate_locks = {}
91
    self.__ssh = None
92
    # logging
93
    self.LogWarning = processor.LogWarning # pylint: disable-msg=C0103
94
    self.LogInfo = processor.LogInfo # pylint: disable-msg=C0103
95
    self.LogStep = processor.LogStep # pylint: disable-msg=C0103
96
    # support for dry-run
97
    self.dry_run_result = None
98

    
99
    # Tasklets
100
    self.tasklets = None
101

    
102
    for attr_name in self._OP_REQP:
103
      attr_val = getattr(op, attr_name, None)
104
      if attr_val is None:
105
        raise errors.OpPrereqError("Required parameter '%s' missing" %
106
                                   attr_name, errors.ECODE_INVAL)
107

    
108
    self.CheckArguments()
109

    
110
  def __GetSSH(self):
111
    """Returns the SshRunner object
112

113
    """
114
    if not self.__ssh:
115
      self.__ssh = ssh.SshRunner(self.cfg.GetClusterName())
116
    return self.__ssh
117

    
118
  ssh = property(fget=__GetSSH)
119

    
120
  def CheckArguments(self):
121
    """Check syntactic validity for the opcode arguments.
122

123
    This method is for doing a simple syntactic check and ensure
124
    validity of opcode parameters, without any cluster-related
125
    checks. While the same can be accomplished in ExpandNames and/or
126
    CheckPrereq, doing these separate is better because:
127

128
      - ExpandNames is left as as purely a lock-related function
129
      - CheckPrereq is run after we have acquired locks (and possible
130
        waited for them)
131

132
    The function is allowed to change the self.op attribute so that
133
    later methods can no longer worry about missing parameters.
134

135
    """
136
    pass
137

    
138
  def ExpandNames(self):
139
    """Expand names for this LU.
140

141
    This method is called before starting to execute the opcode, and it should
142
    update all the parameters of the opcode to their canonical form (e.g. a
143
    short node name must be fully expanded after this method has successfully
144
    completed). This way locking, hooks, logging, ecc. can work correctly.
145

146
    LUs which implement this method must also populate the self.needed_locks
147
    member, as a dict with lock levels as keys, and a list of needed lock names
148
    as values. Rules:
149

150
      - use an empty dict if you don't need any lock
151
      - if you don't need any lock at a particular level omit that level
152
      - don't put anything for the BGL level
153
      - if you want all locks at a level use locking.ALL_SET as a value
154

155
    If you need to share locks (rather than acquire them exclusively) at one
156
    level you can modify self.share_locks, setting a true value (usually 1) for
157
    that level. By default locks are not shared.
158

159
    This function can also define a list of tasklets, which then will be
160
    executed in order instead of the usual LU-level CheckPrereq and Exec
161
    functions, if those are not defined by the LU.
162

163
    Examples::
164

165
      # Acquire all nodes and one instance
166
      self.needed_locks = {
167
        locking.LEVEL_NODE: locking.ALL_SET,
168
        locking.LEVEL_INSTANCE: ['instance1.example.tld'],
169
      }
170
      # Acquire just two nodes
171
      self.needed_locks = {
172
        locking.LEVEL_NODE: ['node1.example.tld', 'node2.example.tld'],
173
      }
174
      # Acquire no locks
175
      self.needed_locks = {} # No, you can't leave it to the default value None
176

177
    """
178
    # The implementation of this method is mandatory only if the new LU is
179
    # concurrent, so that old LUs don't need to be changed all at the same
180
    # time.
181
    if self.REQ_BGL:
182
      self.needed_locks = {} # Exclusive LUs don't need locks.
183
    else:
184
      raise NotImplementedError
185

    
186
  def DeclareLocks(self, level):
187
    """Declare LU locking needs for a level
188

189
    While most LUs can just declare their locking needs at ExpandNames time,
190
    sometimes there's the need to calculate some locks after having acquired
191
    the ones before. This function is called just before acquiring locks at a
192
    particular level, but after acquiring the ones at lower levels, and permits
193
    such calculations. It can be used to modify self.needed_locks, and by
194
    default it does nothing.
195

196
    This function is only called if you have something already set in
197
    self.needed_locks for the level.
198

199
    @param level: Locking level which is going to be locked
200
    @type level: member of ganeti.locking.LEVELS
201

202
    """
203

    
204
  def CheckPrereq(self):
205
    """Check prerequisites for this LU.
206

207
    This method should check that the prerequisites for the execution
208
    of this LU are fulfilled. It can do internode communication, but
209
    it should be idempotent - no cluster or system changes are
210
    allowed.
211

212
    The method should raise errors.OpPrereqError in case something is
213
    not fulfilled. Its return value is ignored.
214

215
    This method should also update all the parameters of the opcode to
216
    their canonical form if it hasn't been done by ExpandNames before.
217

218
    """
219
    if self.tasklets is not None:
220
      for (idx, tl) in enumerate(self.tasklets):
221
        logging.debug("Checking prerequisites for tasklet %s/%s",
222
                      idx + 1, len(self.tasklets))
223
        tl.CheckPrereq()
224
    else:
225
      raise NotImplementedError
226

    
227
  def Exec(self, feedback_fn):
228
    """Execute the LU.
229

230
    This method should implement the actual work. It should raise
231
    errors.OpExecError for failures that are somewhat dealt with in
232
    code, or expected.
233

234
    """
235
    if self.tasklets is not None:
236
      for (idx, tl) in enumerate(self.tasklets):
237
        logging.debug("Executing tasklet %s/%s", idx + 1, len(self.tasklets))
238
        tl.Exec(feedback_fn)
239
    else:
240
      raise NotImplementedError
241

    
242
  def BuildHooksEnv(self):
243
    """Build hooks environment for this LU.
244

245
    This method should return a three-node tuple consisting of: a dict
246
    containing the environment that will be used for running the
247
    specific hook for this LU, a list of node names on which the hook
248
    should run before the execution, and a list of node names on which
249
    the hook should run after the execution.
250

251
    The keys of the dict must not have 'GANETI_' prefixed as this will
252
    be handled in the hooks runner. Also note additional keys will be
253
    added by the hooks runner. If the LU doesn't define any
254
    environment, an empty dict (and not None) should be returned.
255

256
    No nodes should be returned as an empty list (and not None).
257

258
    Note that if the HPATH for a LU class is None, this function will
259
    not be called.
260

261
    """
262
    raise NotImplementedError
263

    
264
  def HooksCallBack(self, phase, hook_results, feedback_fn, lu_result):
265
    """Notify the LU about the results of its hooks.
266

267
    This method is called every time a hooks phase is executed, and notifies
268
    the Logical Unit about the hooks' result. The LU can then use it to alter
269
    its result based on the hooks.  By default the method does nothing and the
270
    previous result is passed back unchanged but any LU can define it if it
271
    wants to use the local cluster hook-scripts somehow.
272

273
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
274
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
275
    @param hook_results: the results of the multi-node hooks rpc call
276
    @param feedback_fn: function used send feedback back to the caller
277
    @param lu_result: the previous Exec result this LU had, or None
278
        in the PRE phase
279
    @return: the new Exec result, based on the previous result
280
        and hook results
281

282
    """
283
    # API must be kept, thus we ignore the unused argument and could
284
    # be a function warnings
285
    # pylint: disable-msg=W0613,R0201
286
    return lu_result
287

    
288
  def _ExpandAndLockInstance(self):
289
    """Helper function to expand and lock an instance.
290

291
    Many LUs that work on an instance take its name in self.op.instance_name
292
    and need to expand it and then declare the expanded name for locking. This
293
    function does it, and then updates self.op.instance_name to the expanded
294
    name. It also initializes needed_locks as a dict, if this hasn't been done
295
    before.
296

297
    """
298
    if self.needed_locks is None:
299
      self.needed_locks = {}
300
    else:
301
      assert locking.LEVEL_INSTANCE not in self.needed_locks, \
302
        "_ExpandAndLockInstance called with instance-level locks set"
303
    expanded_name = self.cfg.ExpandInstanceName(self.op.instance_name)
304
    if expanded_name is None:
305
      raise errors.OpPrereqError("Instance '%s' not known" %
306
                                 self.op.instance_name, errors.ECODE_NOENT)
307
    self.needed_locks[locking.LEVEL_INSTANCE] = expanded_name
308
    self.op.instance_name = expanded_name
309

    
310
  def _LockInstancesNodes(self, primary_only=False):
311
    """Helper function to declare instances' nodes for locking.
312

313
    This function should be called after locking one or more instances to lock
314
    their nodes. Its effect is populating self.needed_locks[locking.LEVEL_NODE]
315
    with all primary or secondary nodes for instances already locked and
316
    present in self.needed_locks[locking.LEVEL_INSTANCE].
317

318
    It should be called from DeclareLocks, and for safety only works if
319
    self.recalculate_locks[locking.LEVEL_NODE] is set.
320

321
    In the future it may grow parameters to just lock some instance's nodes, or
322
    to just lock primaries or secondary nodes, if needed.
323

324
    If should be called in DeclareLocks in a way similar to::
325

326
      if level == locking.LEVEL_NODE:
327
        self._LockInstancesNodes()
328

329
    @type primary_only: boolean
330
    @param primary_only: only lock primary nodes of locked instances
331

332
    """
333
    assert locking.LEVEL_NODE in self.recalculate_locks, \
334
      "_LockInstancesNodes helper function called with no nodes to recalculate"
335

    
336
    # TODO: check if we're really been called with the instance locks held
337

    
338
    # For now we'll replace self.needed_locks[locking.LEVEL_NODE], but in the
339
    # future we might want to have different behaviors depending on the value
340
    # of self.recalculate_locks[locking.LEVEL_NODE]
341
    wanted_nodes = []
342
    for instance_name in self.acquired_locks[locking.LEVEL_INSTANCE]:
343
      instance = self.context.cfg.GetInstanceInfo(instance_name)
344
      wanted_nodes.append(instance.primary_node)
345
      if not primary_only:
346
        wanted_nodes.extend(instance.secondary_nodes)
347

    
348
    if self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_REPLACE:
349
      self.needed_locks[locking.LEVEL_NODE] = wanted_nodes
350
    elif self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_APPEND:
351
      self.needed_locks[locking.LEVEL_NODE].extend(wanted_nodes)
352

    
353
    del self.recalculate_locks[locking.LEVEL_NODE]
354

    
355

    
356
class NoHooksLU(LogicalUnit): # pylint: disable-msg=W0223
357
  """Simple LU which runs no hooks.
358

359
  This LU is intended as a parent for other LogicalUnits which will
360
  run no hooks, in order to reduce duplicate code.
361

362
  """
363
  HPATH = None
364
  HTYPE = None
365

    
366
  def BuildHooksEnv(self):
367
    """Empty BuildHooksEnv for NoHooksLu.
368

369
    This just raises an error.
370

371
    """
372
    assert False, "BuildHooksEnv called for NoHooksLUs"
373

    
374

    
375
class Tasklet:
376
  """Tasklet base class.
377

378
  Tasklets are subcomponents for LUs. LUs can consist entirely of tasklets or
379
  they can mix legacy code with tasklets. Locking needs to be done in the LU,
380
  tasklets know nothing about locks.
381

382
  Subclasses must follow these rules:
383
    - Implement CheckPrereq
384
    - Implement Exec
385

386
  """
387
  def __init__(self, lu):
388
    self.lu = lu
389

    
390
    # Shortcuts
391
    self.cfg = lu.cfg
392
    self.rpc = lu.rpc
393

    
394
  def CheckPrereq(self):
395
    """Check prerequisites for this tasklets.
396

397
    This method should check whether the prerequisites for the execution of
398
    this tasklet are fulfilled. It can do internode communication, but it
399
    should be idempotent - no cluster or system changes are allowed.
400

401
    The method should raise errors.OpPrereqError in case something is not
402
    fulfilled. Its return value is ignored.
403

404
    This method should also update all parameters to their canonical form if it
405
    hasn't been done before.
406

407
    """
408
    raise NotImplementedError
409

    
410
  def Exec(self, feedback_fn):
411
    """Execute the tasklet.
412

413
    This method should implement the actual work. It should raise
414
    errors.OpExecError for failures that are somewhat dealt with in code, or
415
    expected.
416

417
    """
418
    raise NotImplementedError
419

    
420

    
421
def _GetWantedNodes(lu, nodes):
422
  """Returns list of checked and expanded node names.
423

424
  @type lu: L{LogicalUnit}
425
  @param lu: the logical unit on whose behalf we execute
426
  @type nodes: list
427
  @param nodes: list of node names or None for all nodes
428
  @rtype: list
429
  @return: the list of nodes, sorted
430
  @raise errors.OpProgrammerError: if the nodes parameter is wrong type
431

432
  """
433
  if not isinstance(nodes, list):
434
    raise errors.OpPrereqError("Invalid argument type 'nodes'",
435
                               errors.ECODE_INVAL)
436

    
437
  if not nodes:
438
    raise errors.ProgrammerError("_GetWantedNodes should only be called with a"
439
      " non-empty list of nodes whose name is to be expanded.")
440

    
441
  wanted = []
442
  for name in nodes:
443
    node = lu.cfg.ExpandNodeName(name)
444
    if node is None:
445
      raise errors.OpPrereqError("No such node name '%s'" % name,
446
                                 errors.ECODE_NOENT)
447
    wanted.append(node)
448

    
449
  return utils.NiceSort(wanted)
450

    
451

    
452
def _GetWantedInstances(lu, instances):
453
  """Returns list of checked and expanded instance names.
454

455
  @type lu: L{LogicalUnit}
456
  @param lu: the logical unit on whose behalf we execute
457
  @type instances: list
458
  @param instances: list of instance names or None for all instances
459
  @rtype: list
460
  @return: the list of instances, sorted
461
  @raise errors.OpPrereqError: if the instances parameter is wrong type
462
  @raise errors.OpPrereqError: if any of the passed instances is not found
463

464
  """
465
  if not isinstance(instances, list):
466
    raise errors.OpPrereqError("Invalid argument type 'instances'",
467
                               errors.ECODE_INVAL)
468

    
469
  if instances:
470
    wanted = []
471

    
472
    for name in instances:
473
      instance = lu.cfg.ExpandInstanceName(name)
474
      if instance is None:
475
        raise errors.OpPrereqError("No such instance name '%s'" % name,
476
                                   errors.ECODE_NOENT)
477
      wanted.append(instance)
478

    
479
  else:
480
    wanted = utils.NiceSort(lu.cfg.GetInstanceList())
481
  return wanted
482

    
483

    
484
def _CheckOutputFields(static, dynamic, selected):
485
  """Checks whether all selected fields are valid.
486

487
  @type static: L{utils.FieldSet}
488
  @param static: static fields set
489
  @type dynamic: L{utils.FieldSet}
490
  @param dynamic: dynamic fields set
491

492
  """
493
  f = utils.FieldSet()
494
  f.Extend(static)
495
  f.Extend(dynamic)
496

    
497
  delta = f.NonMatching(selected)
498
  if delta:
499
    raise errors.OpPrereqError("Unknown output fields selected: %s"
500
                               % ",".join(delta), errors.ECODE_INVAL)
501

    
502

    
503
def _CheckBooleanOpField(op, name):
504
  """Validates boolean opcode parameters.
505

506
  This will ensure that an opcode parameter is either a boolean value,
507
  or None (but that it always exists).
508

509
  """
510
  val = getattr(op, name, None)
511
  if not (val is None or isinstance(val, bool)):
512
    raise errors.OpPrereqError("Invalid boolean parameter '%s' (%s)" %
513
                               (name, str(val)), errors.ECODE_INVAL)
514
  setattr(op, name, val)
515

    
516

    
517
def _CheckGlobalHvParams(params):
518
  """Validates that given hypervisor params are not global ones.
519

520
  This will ensure that instances don't get customised versions of
521
  global params.
522

523
  """
524
  used_globals = constants.HVC_GLOBALS.intersection(params)
525
  if used_globals:
526
    msg = ("The following hypervisor parameters are global and cannot"
527
           " be customized at instance level, please modify them at"
528
           " cluster level: %s" % utils.CommaJoin(used_globals))
529
    raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
530

    
531

    
532
def _CheckNodeOnline(lu, node):
533
  """Ensure that a given node is online.
534

535
  @param lu: the LU on behalf of which we make the check
536
  @param node: the node to check
537
  @raise errors.OpPrereqError: if the node is offline
538

539
  """
540
  if lu.cfg.GetNodeInfo(node).offline:
541
    raise errors.OpPrereqError("Can't use offline node %s" % node,
542
                               errors.ECODE_INVAL)
543

    
544

    
545
def _CheckNodeNotDrained(lu, node):
546
  """Ensure that a given node is not drained.
547

548
  @param lu: the LU on behalf of which we make the check
549
  @param node: the node to check
550
  @raise errors.OpPrereqError: if the node is drained
551

552
  """
553
  if lu.cfg.GetNodeInfo(node).drained:
554
    raise errors.OpPrereqError("Can't use drained node %s" % node,
555
                               errors.ECODE_INVAL)
556

    
557

    
558
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
559
                          memory, vcpus, nics, disk_template, disks,
560
                          bep, hvp, hypervisor_name):
561
  """Builds instance related env variables for hooks
562

563
  This builds the hook environment from individual variables.
564

565
  @type name: string
566
  @param name: the name of the instance
567
  @type primary_node: string
568
  @param primary_node: the name of the instance's primary node
569
  @type secondary_nodes: list
570
  @param secondary_nodes: list of secondary nodes as strings
571
  @type os_type: string
572
  @param os_type: the name of the instance's OS
573
  @type status: boolean
574
  @param status: the should_run status of the instance
575
  @type memory: string
576
  @param memory: the memory size of the instance
577
  @type vcpus: string
578
  @param vcpus: the count of VCPUs the instance has
579
  @type nics: list
580
  @param nics: list of tuples (ip, mac, mode, link) representing
581
      the NICs the instance has
582
  @type disk_template: string
583
  @param disk_template: the disk template of the instance
584
  @type disks: list
585
  @param disks: the list of (size, mode) pairs
586
  @type bep: dict
587
  @param bep: the backend parameters for the instance
588
  @type hvp: dict
589
  @param hvp: the hypervisor parameters for the instance
590
  @type hypervisor_name: string
591
  @param hypervisor_name: the hypervisor for the instance
592
  @rtype: dict
593
  @return: the hook environment for this instance
594

595
  """
596
  if status:
597
    str_status = "up"
598
  else:
599
    str_status = "down"
600
  env = {
601
    "OP_TARGET": name,
602
    "INSTANCE_NAME": name,
603
    "INSTANCE_PRIMARY": primary_node,
604
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
605
    "INSTANCE_OS_TYPE": os_type,
606
    "INSTANCE_STATUS": str_status,
607
    "INSTANCE_MEMORY": memory,
608
    "INSTANCE_VCPUS": vcpus,
609
    "INSTANCE_DISK_TEMPLATE": disk_template,
610
    "INSTANCE_HYPERVISOR": hypervisor_name,
611
  }
612

    
613
  if nics:
614
    nic_count = len(nics)
615
    for idx, (ip, mac, mode, link) in enumerate(nics):
616
      if ip is None:
617
        ip = ""
618
      env["INSTANCE_NIC%d_IP" % idx] = ip
619
      env["INSTANCE_NIC%d_MAC" % idx] = mac
620
      env["INSTANCE_NIC%d_MODE" % idx] = mode
621
      env["INSTANCE_NIC%d_LINK" % idx] = link
622
      if mode == constants.NIC_MODE_BRIDGED:
623
        env["INSTANCE_NIC%d_BRIDGE" % idx] = link
624
  else:
625
    nic_count = 0
626

    
627
  env["INSTANCE_NIC_COUNT"] = nic_count
628

    
629
  if disks:
630
    disk_count = len(disks)
631
    for idx, (size, mode) in enumerate(disks):
632
      env["INSTANCE_DISK%d_SIZE" % idx] = size
633
      env["INSTANCE_DISK%d_MODE" % idx] = mode
634
  else:
635
    disk_count = 0
636

    
637
  env["INSTANCE_DISK_COUNT"] = disk_count
638

    
639
  for source, kind in [(bep, "BE"), (hvp, "HV")]:
640
    for key, value in source.items():
641
      env["INSTANCE_%s_%s" % (kind, key)] = value
642

    
643
  return env
644

    
645

    
646
def _NICListToTuple(lu, nics):
647
  """Build a list of nic information tuples.
648

649
  This list is suitable to be passed to _BuildInstanceHookEnv or as a return
650
  value in LUQueryInstanceData.
651

652
  @type lu:  L{LogicalUnit}
653
  @param lu: the logical unit on whose behalf we execute
654
  @type nics: list of L{objects.NIC}
655
  @param nics: list of nics to convert to hooks tuples
656

657
  """
658
  hooks_nics = []
659
  c_nicparams = lu.cfg.GetClusterInfo().nicparams[constants.PP_DEFAULT]
660
  for nic in nics:
661
    ip = nic.ip
662
    mac = nic.mac
663
    filled_params = objects.FillDict(c_nicparams, nic.nicparams)
664
    mode = filled_params[constants.NIC_MODE]
665
    link = filled_params[constants.NIC_LINK]
666
    hooks_nics.append((ip, mac, mode, link))
667
  return hooks_nics
668

    
669

    
670
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
671
  """Builds instance related env variables for hooks from an object.
672

673
  @type lu: L{LogicalUnit}
674
  @param lu: the logical unit on whose behalf we execute
675
  @type instance: L{objects.Instance}
676
  @param instance: the instance for which we should build the
677
      environment
678
  @type override: dict
679
  @param override: dictionary with key/values that will override
680
      our values
681
  @rtype: dict
682
  @return: the hook environment dictionary
683

684
  """
685
  cluster = lu.cfg.GetClusterInfo()
686
  bep = cluster.FillBE(instance)
687
  hvp = cluster.FillHV(instance)
688
  args = {
689
    'name': instance.name,
690
    'primary_node': instance.primary_node,
691
    'secondary_nodes': instance.secondary_nodes,
692
    'os_type': instance.os,
693
    'status': instance.admin_up,
694
    'memory': bep[constants.BE_MEMORY],
695
    'vcpus': bep[constants.BE_VCPUS],
696
    'nics': _NICListToTuple(lu, instance.nics),
697
    'disk_template': instance.disk_template,
698
    'disks': [(disk.size, disk.mode) for disk in instance.disks],
699
    'bep': bep,
700
    'hvp': hvp,
701
    'hypervisor_name': instance.hypervisor,
702
  }
703
  if override:
704
    args.update(override)
705
  return _BuildInstanceHookEnv(**args) # pylint: disable-msg=W0142
706

    
707

    
708
def _AdjustCandidatePool(lu, exceptions):
709
  """Adjust the candidate pool after node operations.
710

711
  """
712
  mod_list = lu.cfg.MaintainCandidatePool(exceptions)
713
  if mod_list:
714
    lu.LogInfo("Promoted nodes to master candidate role: %s",
715
               utils.CommaJoin(node.name for node in mod_list))
716
    for name in mod_list:
717
      lu.context.ReaddNode(name)
718
  mc_now, mc_max, _ = lu.cfg.GetMasterCandidateStats(exceptions)
719
  if mc_now > mc_max:
720
    lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
721
               (mc_now, mc_max))
722

    
723

    
724
def _DecideSelfPromotion(lu, exceptions=None):
725
  """Decide whether I should promote myself as a master candidate.
726

727
  """
728
  cp_size = lu.cfg.GetClusterInfo().candidate_pool_size
729
  mc_now, mc_should, _ = lu.cfg.GetMasterCandidateStats(exceptions)
730
  # the new node will increase mc_max with one, so:
731
  mc_should = min(mc_should + 1, cp_size)
732
  return mc_now < mc_should
733

    
734

    
735
def _CheckNicsBridgesExist(lu, target_nics, target_node,
736
                               profile=constants.PP_DEFAULT):
737
  """Check that the brigdes needed by a list of nics exist.
738

739
  """
740
  c_nicparams = lu.cfg.GetClusterInfo().nicparams[profile]
741
  paramslist = [objects.FillDict(c_nicparams, nic.nicparams)
742
                for nic in target_nics]
743
  brlist = [params[constants.NIC_LINK] for params in paramslist
744
            if params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED]
745
  if brlist:
746
    result = lu.rpc.call_bridges_exist(target_node, brlist)
747
    result.Raise("Error checking bridges on destination node '%s'" %
748
                 target_node, prereq=True, ecode=errors.ECODE_ENVIRON)
749

    
750

    
751
def _CheckInstanceBridgesExist(lu, instance, node=None):
752
  """Check that the brigdes needed by an instance exist.
753

754
  """
755
  if node is None:
756
    node = instance.primary_node
757
  _CheckNicsBridgesExist(lu, instance.nics, node)
758

    
759

    
760
def _CheckOSVariant(os_obj, name):
761
  """Check whether an OS name conforms to the os variants specification.
762

763
  @type os_obj: L{objects.OS}
764
  @param os_obj: OS object to check
765
  @type name: string
766
  @param name: OS name passed by the user, to check for validity
767

768
  """
769
  if not os_obj.supported_variants:
770
    return
771
  try:
772
    variant = name.split("+", 1)[1]
773
  except IndexError:
774
    raise errors.OpPrereqError("OS name must include a variant",
775
                               errors.ECODE_INVAL)
776

    
777
  if variant not in os_obj.supported_variants:
778
    raise errors.OpPrereqError("Unsupported OS variant", errors.ECODE_INVAL)
779

    
780

    
781
def _GetNodeInstancesInner(cfg, fn):
782
  return [i for i in cfg.GetAllInstancesInfo().values() if fn(i)]
783

    
784

    
785
def _GetNodeInstances(cfg, node_name):
786
  """Returns a list of all primary and secondary instances on a node.
787

788
  """
789

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

    
792

    
793
def _GetNodePrimaryInstances(cfg, node_name):
794
  """Returns primary instances on a node.
795

796
  """
797
  return _GetNodeInstancesInner(cfg,
798
                                lambda inst: node_name == inst.primary_node)
799

    
800

    
801
def _GetNodeSecondaryInstances(cfg, node_name):
802
  """Returns secondary instances on a node.
803

804
  """
805
  return _GetNodeInstancesInner(cfg,
806
                                lambda inst: node_name in inst.secondary_nodes)
807

    
808

    
809
def _GetStorageTypeArgs(cfg, storage_type):
810
  """Returns the arguments for a storage type.
811

812
  """
813
  # Special case for file storage
814
  if storage_type == constants.ST_FILE:
815
    # storage.FileStorage wants a list of storage directories
816
    return [[cfg.GetFileStorageDir()]]
817

    
818
  return []
819

    
820

    
821
def _FindFaultyInstanceDisks(cfg, rpc, instance, node_name, prereq):
822
  faulty = []
823

    
824
  for dev in instance.disks:
825
    cfg.SetDiskID(dev, node_name)
826

    
827
  result = rpc.call_blockdev_getmirrorstatus(node_name, instance.disks)
828
  result.Raise("Failed to get disk status from node %s" % node_name,
829
               prereq=prereq, ecode=errors.ECODE_ENVIRON)
830

    
831
  for idx, bdev_status in enumerate(result.payload):
832
    if bdev_status and bdev_status.ldisk_status == constants.LDS_FAULTY:
833
      faulty.append(idx)
834

    
835
  return faulty
836

    
837

    
838
class LUPostInitCluster(LogicalUnit):
839
  """Logical unit for running hooks after cluster initialization.
840

841
  """
842
  HPATH = "cluster-init"
843
  HTYPE = constants.HTYPE_CLUSTER
844
  _OP_REQP = []
845

    
846
  def BuildHooksEnv(self):
847
    """Build hooks env.
848

849
    """
850
    env = {"OP_TARGET": self.cfg.GetClusterName()}
851
    mn = self.cfg.GetMasterNode()
852
    return env, [], [mn]
853

    
854
  def CheckPrereq(self):
855
    """No prerequisites to check.
856

857
    """
858
    return True
859

    
860
  def Exec(self, feedback_fn):
861
    """Nothing to do.
862

863
    """
864
    return True
865

    
866

    
867
class LUDestroyCluster(LogicalUnit):
868
  """Logical unit for destroying the cluster.
869

870
  """
871
  HPATH = "cluster-destroy"
872
  HTYPE = constants.HTYPE_CLUSTER
873
  _OP_REQP = []
874

    
875
  def BuildHooksEnv(self):
876
    """Build hooks env.
877

878
    """
879
    env = {"OP_TARGET": self.cfg.GetClusterName()}
880
    return env, [], []
881

    
882
  def CheckPrereq(self):
883
    """Check prerequisites.
884

885
    This checks whether the cluster is empty.
886

887
    Any errors are signaled by raising errors.OpPrereqError.
888

889
    """
890
    master = self.cfg.GetMasterNode()
891

    
892
    nodelist = self.cfg.GetNodeList()
893
    if len(nodelist) != 1 or nodelist[0] != master:
894
      raise errors.OpPrereqError("There are still %d node(s) in"
895
                                 " this cluster." % (len(nodelist) - 1),
896
                                 errors.ECODE_INVAL)
897
    instancelist = self.cfg.GetInstanceList()
898
    if instancelist:
899
      raise errors.OpPrereqError("There are still %d instance(s) in"
900
                                 " this cluster." % len(instancelist),
901
                                 errors.ECODE_INVAL)
902

    
903
  def Exec(self, feedback_fn):
904
    """Destroys the cluster.
905

906
    """
907
    master = self.cfg.GetMasterNode()
908
    modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup
909

    
910
    # Run post hooks on master node before it's removed
911
    hm = self.proc.hmclass(self.rpc.call_hooks_runner, self)
912
    try:
913
      hm.RunPhase(constants.HOOKS_PHASE_POST, [master])
914
    except:
915
      # pylint: disable-msg=W0702
916
      self.LogWarning("Errors occurred running hooks on %s" % master)
917

    
918
    result = self.rpc.call_node_stop_master(master, False)
919
    result.Raise("Could not disable the master role")
920

    
921
    if modify_ssh_setup:
922
      priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
923
      utils.CreateBackup(priv_key)
924
      utils.CreateBackup(pub_key)
925

    
926
    return master
927

    
928

    
929
class LUVerifyCluster(LogicalUnit):
930
  """Verifies the cluster status.
931

932
  """
933
  HPATH = "cluster-verify"
934
  HTYPE = constants.HTYPE_CLUSTER
935
  _OP_REQP = ["skip_checks", "verbose", "error_codes", "debug_simulate_errors"]
936
  REQ_BGL = False
937

    
938
  TCLUSTER = "cluster"
939
  TNODE = "node"
940
  TINSTANCE = "instance"
941

    
942
  ECLUSTERCFG = (TCLUSTER, "ECLUSTERCFG")
943
  EINSTANCEBADNODE = (TINSTANCE, "EINSTANCEBADNODE")
944
  EINSTANCEDOWN = (TINSTANCE, "EINSTANCEDOWN")
945
  EINSTANCELAYOUT = (TINSTANCE, "EINSTANCELAYOUT")
946
  EINSTANCEMISSINGDISK = (TINSTANCE, "EINSTANCEMISSINGDISK")
947
  EINSTANCEMISSINGDISK = (TINSTANCE, "EINSTANCEMISSINGDISK")
948
  EINSTANCEWRONGNODE = (TINSTANCE, "EINSTANCEWRONGNODE")
949
  ENODEDRBD = (TNODE, "ENODEDRBD")
950
  ENODEFILECHECK = (TNODE, "ENODEFILECHECK")
951
  ENODEHOOKS = (TNODE, "ENODEHOOKS")
952
  ENODEHV = (TNODE, "ENODEHV")
953
  ENODELVM = (TNODE, "ENODELVM")
954
  ENODEN1 = (TNODE, "ENODEN1")
955
  ENODENET = (TNODE, "ENODENET")
956
  ENODEORPHANINSTANCE = (TNODE, "ENODEORPHANINSTANCE")
957
  ENODEORPHANLV = (TNODE, "ENODEORPHANLV")
958
  ENODERPC = (TNODE, "ENODERPC")
959
  ENODESSH = (TNODE, "ENODESSH")
960
  ENODEVERSION = (TNODE, "ENODEVERSION")
961
  ENODESETUP = (TNODE, "ENODESETUP")
962
  ENODETIME = (TNODE, "ENODETIME")
963

    
964
  ETYPE_FIELD = "code"
965
  ETYPE_ERROR = "ERROR"
966
  ETYPE_WARNING = "WARNING"
967

    
968
  def ExpandNames(self):
969
    self.needed_locks = {
970
      locking.LEVEL_NODE: locking.ALL_SET,
971
      locking.LEVEL_INSTANCE: locking.ALL_SET,
972
    }
973
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
974

    
975
  def _Error(self, ecode, item, msg, *args, **kwargs):
976
    """Format an error message.
977

978
    Based on the opcode's error_codes parameter, either format a
979
    parseable error code, or a simpler error string.
980

981
    This must be called only from Exec and functions called from Exec.
982

983
    """
984
    ltype = kwargs.get(self.ETYPE_FIELD, self.ETYPE_ERROR)
985
    itype, etxt = ecode
986
    # first complete the msg
987
    if args:
988
      msg = msg % args
989
    # then format the whole message
990
    if self.op.error_codes:
991
      msg = "%s:%s:%s:%s:%s" % (ltype, etxt, itype, item, msg)
992
    else:
993
      if item:
994
        item = " " + item
995
      else:
996
        item = ""
997
      msg = "%s: %s%s: %s" % (ltype, itype, item, msg)
998
    # and finally report it via the feedback_fn
999
    self._feedback_fn("  - %s" % msg)
1000

    
1001
  def _ErrorIf(self, cond, *args, **kwargs):
1002
    """Log an error message if the passed condition is True.
1003

1004
    """
1005
    cond = bool(cond) or self.op.debug_simulate_errors
1006
    if cond:
1007
      self._Error(*args, **kwargs)
1008
    # do not mark the operation as failed for WARN cases only
1009
    if kwargs.get(self.ETYPE_FIELD, self.ETYPE_ERROR) == self.ETYPE_ERROR:
1010
      self.bad = self.bad or cond
1011

    
1012
  def _VerifyNode(self, nodeinfo, file_list, local_cksum,
1013
                  node_result, master_files, drbd_map, vg_name):
1014
    """Run multiple tests against a node.
1015

1016
    Test list:
1017

1018
      - compares ganeti version
1019
      - checks vg existence and size > 20G
1020
      - checks config file checksum
1021
      - checks ssh to other nodes
1022

1023
    @type nodeinfo: L{objects.Node}
1024
    @param nodeinfo: the node to check
1025
    @param file_list: required list of files
1026
    @param local_cksum: dictionary of local files and their checksums
1027
    @param node_result: the results from the node
1028
    @param master_files: list of files that only masters should have
1029
    @param drbd_map: the useddrbd minors for this node, in
1030
        form of minor: (instance, must_exist) which correspond to instances
1031
        and their running status
1032
    @param vg_name: Ganeti Volume Group (result of self.cfg.GetVGName())
1033

1034
    """
1035
    node = nodeinfo.name
1036
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1037

    
1038
    # main result, node_result should be a non-empty dict
1039
    test = not node_result or not isinstance(node_result, dict)
1040
    _ErrorIf(test, self.ENODERPC, node,
1041
                  "unable to verify node: no data returned")
1042
    if test:
1043
      return
1044

    
1045
    # compares ganeti version
1046
    local_version = constants.PROTOCOL_VERSION
1047
    remote_version = node_result.get('version', None)
1048
    test = not (remote_version and
1049
                isinstance(remote_version, (list, tuple)) and
1050
                len(remote_version) == 2)
1051
    _ErrorIf(test, self.ENODERPC, node,
1052
             "connection to node returned invalid data")
1053
    if test:
1054
      return
1055

    
1056
    test = local_version != remote_version[0]
1057
    _ErrorIf(test, self.ENODEVERSION, node,
1058
             "incompatible protocol versions: master %s,"
1059
             " node %s", local_version, remote_version[0])
1060
    if test:
1061
      return
1062

    
1063
    # node seems compatible, we can actually try to look into its results
1064

    
1065
    # full package version
1066
    self._ErrorIf(constants.RELEASE_VERSION != remote_version[1],
1067
                  self.ENODEVERSION, node,
1068
                  "software version mismatch: master %s, node %s",
1069
                  constants.RELEASE_VERSION, remote_version[1],
1070
                  code=self.ETYPE_WARNING)
1071

    
1072
    # checks vg existence and size > 20G
1073
    if vg_name is not None:
1074
      vglist = node_result.get(constants.NV_VGLIST, None)
1075
      test = not vglist
1076
      _ErrorIf(test, self.ENODELVM, node, "unable to check volume groups")
1077
      if not test:
1078
        vgstatus = utils.CheckVolumeGroupSize(vglist, vg_name,
1079
                                              constants.MIN_VG_SIZE)
1080
        _ErrorIf(vgstatus, self.ENODELVM, node, vgstatus)
1081

    
1082
    # checks config file checksum
1083

    
1084
    remote_cksum = node_result.get(constants.NV_FILELIST, None)
1085
    test = not isinstance(remote_cksum, dict)
1086
    _ErrorIf(test, self.ENODEFILECHECK, node,
1087
             "node hasn't returned file checksum data")
1088
    if not test:
1089
      for file_name in file_list:
1090
        node_is_mc = nodeinfo.master_candidate
1091
        must_have = (file_name not in master_files) or node_is_mc
1092
        # missing
1093
        test1 = file_name not in remote_cksum
1094
        # invalid checksum
1095
        test2 = not test1 and remote_cksum[file_name] != local_cksum[file_name]
1096
        # existing and good
1097
        test3 = not test1 and remote_cksum[file_name] == local_cksum[file_name]
1098
        _ErrorIf(test1 and must_have, self.ENODEFILECHECK, node,
1099
                 "file '%s' missing", file_name)
1100
        _ErrorIf(test2 and must_have, self.ENODEFILECHECK, node,
1101
                 "file '%s' has wrong checksum", file_name)
1102
        # not candidate and this is not a must-have file
1103
        _ErrorIf(test2 and not must_have, self.ENODEFILECHECK, node,
1104
                 "file '%s' should not exist on non master"
1105
                 " candidates (and the file is outdated)", file_name)
1106
        # all good, except non-master/non-must have combination
1107
        _ErrorIf(test3 and not must_have, self.ENODEFILECHECK, node,
1108
                 "file '%s' should not exist"
1109
                 " on non master candidates", file_name)
1110

    
1111
    # checks ssh to any
1112

    
1113
    test = constants.NV_NODELIST not in node_result
1114
    _ErrorIf(test, self.ENODESSH, node,
1115
             "node hasn't returned node ssh connectivity data")
1116
    if not test:
1117
      if node_result[constants.NV_NODELIST]:
1118
        for a_node, a_msg in node_result[constants.NV_NODELIST].items():
1119
          _ErrorIf(True, self.ENODESSH, node,
1120
                   "ssh communication with node '%s': %s", a_node, a_msg)
1121

    
1122
    test = constants.NV_NODENETTEST not in node_result
1123
    _ErrorIf(test, self.ENODENET, node,
1124
             "node hasn't returned node tcp connectivity data")
1125
    if not test:
1126
      if node_result[constants.NV_NODENETTEST]:
1127
        nlist = utils.NiceSort(node_result[constants.NV_NODENETTEST].keys())
1128
        for anode in nlist:
1129
          _ErrorIf(True, self.ENODENET, node,
1130
                   "tcp communication with node '%s': %s",
1131
                   anode, node_result[constants.NV_NODENETTEST][anode])
1132

    
1133
    hyp_result = node_result.get(constants.NV_HYPERVISOR, None)
1134
    if isinstance(hyp_result, dict):
1135
      for hv_name, hv_result in hyp_result.iteritems():
1136
        test = hv_result is not None
1137
        _ErrorIf(test, self.ENODEHV, node,
1138
                 "hypervisor %s verify failure: '%s'", hv_name, hv_result)
1139

    
1140
    # check used drbd list
1141
    if vg_name is not None:
1142
      used_minors = node_result.get(constants.NV_DRBDLIST, [])
1143
      test = not isinstance(used_minors, (tuple, list))
1144
      _ErrorIf(test, self.ENODEDRBD, node,
1145
               "cannot parse drbd status file: %s", str(used_minors))
1146
      if not test:
1147
        for minor, (iname, must_exist) in drbd_map.items():
1148
          test = minor not in used_minors and must_exist
1149
          _ErrorIf(test, self.ENODEDRBD, node,
1150
                   "drbd minor %d of instance %s is not active",
1151
                   minor, iname)
1152
        for minor in used_minors:
1153
          test = minor not in drbd_map
1154
          _ErrorIf(test, self.ENODEDRBD, node,
1155
                   "unallocated drbd minor %d is in use", minor)
1156
    test = node_result.get(constants.NV_NODESETUP,
1157
                           ["Missing NODESETUP results"])
1158
    _ErrorIf(test, self.ENODESETUP, node, "node setup error: %s",
1159
             "; ".join(test))
1160

    
1161
    # check pv names
1162
    if vg_name is not None:
1163
      pvlist = node_result.get(constants.NV_PVLIST, None)
1164
      test = pvlist is None
1165
      _ErrorIf(test, self.ENODELVM, node, "Can't get PV list from node")
1166
      if not test:
1167
        # check that ':' is not present in PV names, since it's a
1168
        # special character for lvcreate (denotes the range of PEs to
1169
        # use on the PV)
1170
        for _, pvname, owner_vg in pvlist:
1171
          test = ":" in pvname
1172
          _ErrorIf(test, self.ENODELVM, node, "Invalid character ':' in PV"
1173
                   " '%s' of VG '%s'", pvname, owner_vg)
1174

    
1175
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
1176
                      node_instance, n_offline):
1177
    """Verify an instance.
1178

1179
    This function checks to see if the required block devices are
1180
    available on the instance's node.
1181

1182
    """
1183
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1184
    node_current = instanceconfig.primary_node
1185

    
1186
    node_vol_should = {}
1187
    instanceconfig.MapLVsByNode(node_vol_should)
1188

    
1189
    for node in node_vol_should:
1190
      if node in n_offline:
1191
        # ignore missing volumes on offline nodes
1192
        continue
1193
      for volume in node_vol_should[node]:
1194
        test = node not in node_vol_is or volume not in node_vol_is[node]
1195
        _ErrorIf(test, self.EINSTANCEMISSINGDISK, instance,
1196
                 "volume %s missing on node %s", volume, node)
1197

    
1198
    if instanceconfig.admin_up:
1199
      test = ((node_current not in node_instance or
1200
               not instance in node_instance[node_current]) and
1201
              node_current not in n_offline)
1202
      _ErrorIf(test, self.EINSTANCEDOWN, instance,
1203
               "instance not running on its primary node %s",
1204
               node_current)
1205

    
1206
    for node in node_instance:
1207
      if (not node == node_current):
1208
        test = instance in node_instance[node]
1209
        _ErrorIf(test, self.EINSTANCEWRONGNODE, instance,
1210
                 "instance should not run on node %s", node)
1211

    
1212
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is):
1213
    """Verify if there are any unknown volumes in the cluster.
1214

1215
    The .os, .swap and backup volumes are ignored. All other volumes are
1216
    reported as unknown.
1217

1218
    """
1219
    for node in node_vol_is:
1220
      for volume in node_vol_is[node]:
1221
        test = (node not in node_vol_should or
1222
                volume not in node_vol_should[node])
1223
        self._ErrorIf(test, self.ENODEORPHANLV, node,
1224
                      "volume %s is unknown", volume)
1225

    
1226
  def _VerifyOrphanInstances(self, instancelist, node_instance):
1227
    """Verify the list of running instances.
1228

1229
    This checks what instances are running but unknown to the cluster.
1230

1231
    """
1232
    for node in node_instance:
1233
      for o_inst in node_instance[node]:
1234
        test = o_inst not in instancelist
1235
        self._ErrorIf(test, self.ENODEORPHANINSTANCE, node,
1236
                      "instance %s on node %s should not exist", o_inst, node)
1237

    
1238
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg):
1239
    """Verify N+1 Memory Resilience.
1240

1241
    Check that if one single node dies we can still start all the instances it
1242
    was primary for.
1243

1244
    """
1245
    for node, nodeinfo in node_info.iteritems():
1246
      # This code checks that every node which is now listed as secondary has
1247
      # enough memory to host all instances it is supposed to should a single
1248
      # other node in the cluster fail.
1249
      # FIXME: not ready for failover to an arbitrary node
1250
      # FIXME: does not support file-backed instances
1251
      # WARNING: we currently take into account down instances as well as up
1252
      # ones, considering that even if they're down someone might want to start
1253
      # them even in the event of a node failure.
1254
      for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
1255
        needed_mem = 0
1256
        for instance in instances:
1257
          bep = self.cfg.GetClusterInfo().FillBE(instance_cfg[instance])
1258
          if bep[constants.BE_AUTO_BALANCE]:
1259
            needed_mem += bep[constants.BE_MEMORY]
1260
        test = nodeinfo['mfree'] < needed_mem
1261
        self._ErrorIf(test, self.ENODEN1, node,
1262
                      "not enough memory on to accommodate"
1263
                      " failovers should peer node %s fail", prinode)
1264

    
1265
  def CheckPrereq(self):
1266
    """Check prerequisites.
1267

1268
    Transform the list of checks we're going to skip into a set and check that
1269
    all its members are valid.
1270

1271
    """
1272
    self.skip_set = frozenset(self.op.skip_checks)
1273
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
1274
      raise errors.OpPrereqError("Invalid checks to be skipped specified",
1275
                                 errors.ECODE_INVAL)
1276

    
1277
  def BuildHooksEnv(self):
1278
    """Build hooks env.
1279

1280
    Cluster-Verify hooks just ran in the post phase and their failure makes
1281
    the output be logged in the verify output and the verification to fail.
1282

1283
    """
1284
    all_nodes = self.cfg.GetNodeList()
1285
    env = {
1286
      "CLUSTER_TAGS": " ".join(self.cfg.GetClusterInfo().GetTags())
1287
      }
1288
    for node in self.cfg.GetAllNodesInfo().values():
1289
      env["NODE_TAGS_%s" % node.name] = " ".join(node.GetTags())
1290

    
1291
    return env, [], all_nodes
1292

    
1293
  def Exec(self, feedback_fn):
1294
    """Verify integrity of cluster, performing various test on nodes.
1295

1296
    """
1297
    self.bad = False
1298
    _ErrorIf = self._ErrorIf # pylint: disable-msg=C0103
1299
    verbose = self.op.verbose
1300
    self._feedback_fn = feedback_fn
1301
    feedback_fn("* Verifying global settings")
1302
    for msg in self.cfg.VerifyConfig():
1303
      _ErrorIf(True, self.ECLUSTERCFG, None, msg)
1304

    
1305
    vg_name = self.cfg.GetVGName()
1306
    hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
1307
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
1308
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
1309
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
1310
    instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
1311
                        for iname in instancelist)
1312
    i_non_redundant = [] # Non redundant instances
1313
    i_non_a_balanced = [] # Non auto-balanced instances
1314
    n_offline = [] # List of offline nodes
1315
    n_drained = [] # List of nodes being drained
1316
    node_volume = {}
1317
    node_instance = {}
1318
    node_info = {}
1319
    instance_cfg = {}
1320

    
1321
    # FIXME: verify OS list
1322
    # do local checksums
1323
    master_files = [constants.CLUSTER_CONF_FILE]
1324

    
1325
    file_names = ssconf.SimpleStore().GetFileList()
1326
    file_names.append(constants.SSL_CERT_FILE)
1327
    file_names.append(constants.RAPI_CERT_FILE)
1328
    file_names.extend(master_files)
1329

    
1330
    local_checksums = utils.FingerprintFiles(file_names)
1331

    
1332
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
1333
    node_verify_param = {
1334
      constants.NV_FILELIST: file_names,
1335
      constants.NV_NODELIST: [node.name for node in nodeinfo
1336
                              if not node.offline],
1337
      constants.NV_HYPERVISOR: hypervisors,
1338
      constants.NV_NODENETTEST: [(node.name, node.primary_ip,
1339
                                  node.secondary_ip) for node in nodeinfo
1340
                                 if not node.offline],
1341
      constants.NV_INSTANCELIST: hypervisors,
1342
      constants.NV_VERSION: None,
1343
      constants.NV_HVINFO: self.cfg.GetHypervisorType(),
1344
      constants.NV_NODESETUP: None,
1345
      constants.NV_TIME: None,
1346
      }
1347

    
1348
    if vg_name is not None:
1349
      node_verify_param[constants.NV_VGLIST] = None
1350
      node_verify_param[constants.NV_LVLIST] = vg_name
1351
      node_verify_param[constants.NV_PVLIST] = [vg_name]
1352
      node_verify_param[constants.NV_DRBDLIST] = None
1353

    
1354
    # Due to the way our RPC system works, exact response times cannot be
1355
    # guaranteed (e.g. a broken node could run into a timeout). By keeping the
1356
    # time before and after executing the request, we can at least have a time
1357
    # window.
1358
    nvinfo_starttime = time.time()
1359
    all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
1360
                                           self.cfg.GetClusterName())
1361
    nvinfo_endtime = time.time()
1362

    
1363
    cluster = self.cfg.GetClusterInfo()
1364
    master_node = self.cfg.GetMasterNode()
1365
    all_drbd_map = self.cfg.ComputeDRBDMap()
1366

    
1367
    feedback_fn("* Verifying node status")
1368
    for node_i in nodeinfo:
1369
      node = node_i.name
1370

    
1371
      if node_i.offline:
1372
        if verbose:
1373
          feedback_fn("* Skipping offline node %s" % (node,))
1374
        n_offline.append(node)
1375
        continue
1376

    
1377
      if node == master_node:
1378
        ntype = "master"
1379
      elif node_i.master_candidate:
1380
        ntype = "master candidate"
1381
      elif node_i.drained:
1382
        ntype = "drained"
1383
        n_drained.append(node)
1384
      else:
1385
        ntype = "regular"
1386
      if verbose:
1387
        feedback_fn("* Verifying node %s (%s)" % (node, ntype))
1388

    
1389
      msg = all_nvinfo[node].fail_msg
1390
      _ErrorIf(msg, self.ENODERPC, node, "while contacting node: %s", msg)
1391
      if msg:
1392
        continue
1393

    
1394
      nresult = all_nvinfo[node].payload
1395
      node_drbd = {}
1396
      for minor, instance in all_drbd_map[node].items():
1397
        test = instance not in instanceinfo
1398
        _ErrorIf(test, self.ECLUSTERCFG, None,
1399
                 "ghost instance '%s' in temporary DRBD map", instance)
1400
          # ghost instance should not be running, but otherwise we
1401
          # don't give double warnings (both ghost instance and
1402
          # unallocated minor in use)
1403
        if test:
1404
          node_drbd[minor] = (instance, False)
1405
        else:
1406
          instance = instanceinfo[instance]
1407
          node_drbd[minor] = (instance.name, instance.admin_up)
1408

    
1409
      self._VerifyNode(node_i, file_names, local_checksums,
1410
                       nresult, master_files, node_drbd, vg_name)
1411

    
1412
      lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
1413
      if vg_name is None:
1414
        node_volume[node] = {}
1415
      elif isinstance(lvdata, basestring):
1416
        _ErrorIf(True, self.ENODELVM, node, "LVM problem on node: %s",
1417
                 utils.SafeEncode(lvdata))
1418
        node_volume[node] = {}
1419
      elif not isinstance(lvdata, dict):
1420
        _ErrorIf(True, self.ENODELVM, node, "rpc call to node failed (lvlist)")
1421
        continue
1422
      else:
1423
        node_volume[node] = lvdata
1424

    
1425
      # node_instance
1426
      idata = nresult.get(constants.NV_INSTANCELIST, None)
1427
      test = not isinstance(idata, list)
1428
      _ErrorIf(test, self.ENODEHV, node,
1429
               "rpc call to node failed (instancelist)")
1430
      if test:
1431
        continue
1432

    
1433
      node_instance[node] = idata
1434

    
1435
      # node_info
1436
      nodeinfo = nresult.get(constants.NV_HVINFO, None)
1437
      test = not isinstance(nodeinfo, dict)
1438
      _ErrorIf(test, self.ENODEHV, node, "rpc call to node failed (hvinfo)")
1439
      if test:
1440
        continue
1441

    
1442
      # Node time
1443
      ntime = nresult.get(constants.NV_TIME, None)
1444
      try:
1445
        ntime_merged = utils.MergeTime(ntime)
1446
      except (ValueError, TypeError):
1447
        _ErrorIf(test, self.ENODETIME, node, "Node returned invalid time")
1448

    
1449
      if ntime_merged < (nvinfo_starttime - constants.NODE_MAX_CLOCK_SKEW):
1450
        ntime_diff = abs(nvinfo_starttime - ntime_merged)
1451
      elif ntime_merged > (nvinfo_endtime + constants.NODE_MAX_CLOCK_SKEW):
1452
        ntime_diff = abs(ntime_merged - nvinfo_endtime)
1453
      else:
1454
        ntime_diff = None
1455

    
1456
      _ErrorIf(ntime_diff is not None, self.ENODETIME, node,
1457
               "Node time diverges by at least %0.1fs from master node time",
1458
               ntime_diff)
1459

    
1460
      if ntime_diff is not None:
1461
        continue
1462

    
1463
      try:
1464
        node_info[node] = {
1465
          "mfree": int(nodeinfo['memory_free']),
1466
          "pinst": [],
1467
          "sinst": [],
1468
          # dictionary holding all instances this node is secondary for,
1469
          # grouped by their primary node. Each key is a cluster node, and each
1470
          # value is a list of instances which have the key as primary and the
1471
          # current node as secondary.  this is handy to calculate N+1 memory
1472
          # availability if you can only failover from a primary to its
1473
          # secondary.
1474
          "sinst-by-pnode": {},
1475
        }
1476
        # FIXME: devise a free space model for file based instances as well
1477
        if vg_name is not None:
1478
          test = (constants.NV_VGLIST not in nresult or
1479
                  vg_name not in nresult[constants.NV_VGLIST])
1480
          _ErrorIf(test, self.ENODELVM, node,
1481
                   "node didn't return data for the volume group '%s'"
1482
                   " - it is either missing or broken", vg_name)
1483
          if test:
1484
            continue
1485
          node_info[node]["dfree"] = int(nresult[constants.NV_VGLIST][vg_name])
1486
      except (ValueError, KeyError):
1487
        _ErrorIf(True, self.ENODERPC, node,
1488
                 "node returned invalid nodeinfo, check lvm/hypervisor")
1489
        continue
1490

    
1491
    node_vol_should = {}
1492

    
1493
    feedback_fn("* Verifying instance status")
1494
    for instance in instancelist:
1495
      if verbose:
1496
        feedback_fn("* Verifying instance %s" % instance)
1497
      inst_config = instanceinfo[instance]
1498
      self._VerifyInstance(instance, inst_config, node_volume,
1499
                           node_instance, n_offline)
1500
      inst_nodes_offline = []
1501

    
1502
      inst_config.MapLVsByNode(node_vol_should)
1503

    
1504
      instance_cfg[instance] = inst_config
1505

    
1506
      pnode = inst_config.primary_node
1507
      _ErrorIf(pnode not in node_info and pnode not in n_offline,
1508
               self.ENODERPC, pnode, "instance %s, connection to"
1509
               " primary node failed", instance)
1510
      if pnode in node_info:
1511
        node_info[pnode]['pinst'].append(instance)
1512

    
1513
      if pnode in n_offline:
1514
        inst_nodes_offline.append(pnode)
1515

    
1516
      # If the instance is non-redundant we cannot survive losing its primary
1517
      # node, so we are not N+1 compliant. On the other hand we have no disk
1518
      # templates with more than one secondary so that situation is not well
1519
      # supported either.
1520
      # FIXME: does not support file-backed instances
1521
      if len(inst_config.secondary_nodes) == 0:
1522
        i_non_redundant.append(instance)
1523
      _ErrorIf(len(inst_config.secondary_nodes) > 1,
1524
               self.EINSTANCELAYOUT, instance,
1525
               "instance has multiple secondary nodes", code="WARNING")
1526

    
1527
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1528
        i_non_a_balanced.append(instance)
1529

    
1530
      for snode in inst_config.secondary_nodes:
1531
        _ErrorIf(snode not in node_info and snode not in n_offline,
1532
                 self.ENODERPC, snode,
1533
                 "instance %s, connection to secondary node"
1534
                 "failed", instance)
1535

    
1536
        if snode in node_info:
1537
          node_info[snode]['sinst'].append(instance)
1538
          if pnode not in node_info[snode]['sinst-by-pnode']:
1539
            node_info[snode]['sinst-by-pnode'][pnode] = []
1540
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1541

    
1542
        if snode in n_offline:
1543
          inst_nodes_offline.append(snode)
1544

    
1545
      # warn that the instance lives on offline nodes
1546
      _ErrorIf(inst_nodes_offline, self.EINSTANCEBADNODE, instance,
1547
               "instance lives on offline node(s) %s",
1548
               utils.CommaJoin(inst_nodes_offline))
1549

    
1550
    feedback_fn("* Verifying orphan volumes")
1551
    self._VerifyOrphanVolumes(node_vol_should, node_volume)
1552

    
1553
    feedback_fn("* Verifying remaining instances")
1554
    self._VerifyOrphanInstances(instancelist, node_instance)
1555

    
1556
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1557
      feedback_fn("* Verifying N+1 Memory redundancy")
1558
      self._VerifyNPlusOneMemory(node_info, instance_cfg)
1559

    
1560
    feedback_fn("* Other Notes")
1561
    if i_non_redundant:
1562
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1563
                  % len(i_non_redundant))
1564

    
1565
    if i_non_a_balanced:
1566
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1567
                  % len(i_non_a_balanced))
1568

    
1569
    if n_offline:
1570
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1571

    
1572
    if n_drained:
1573
      feedback_fn("  - NOTICE: %d drained node(s) found." % len(n_drained))
1574

    
1575
    return not self.bad
1576

    
1577
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1578
    """Analyze the post-hooks' result
1579

1580
    This method analyses the hook result, handles it, and sends some
1581
    nicely-formatted feedback back to the user.
1582

1583
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
1584
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1585
    @param hooks_results: the results of the multi-node hooks rpc call
1586
    @param feedback_fn: function used send feedback back to the caller
1587
    @param lu_result: previous Exec result
1588
    @return: the new Exec result, based on the previous result
1589
        and hook results
1590

1591
    """
1592
    # We only really run POST phase hooks, and are only interested in
1593
    # their results
1594
    if phase == constants.HOOKS_PHASE_POST:
1595
      # Used to change hooks' output to proper indentation
1596
      indent_re = re.compile('^', re.M)
1597
      feedback_fn("* Hooks Results")
1598
      assert hooks_results, "invalid result from hooks"
1599

    
1600
      for node_name in hooks_results:
1601
        res = hooks_results[node_name]
1602
        msg = res.fail_msg
1603
        test = msg and not res.offline
1604
        self._ErrorIf(test, self.ENODEHOOKS, node_name,
1605
                      "Communication failure in hooks execution: %s", msg)
1606
        if test:
1607
          # override manually lu_result here as _ErrorIf only
1608
          # overrides self.bad
1609
          lu_result = 1
1610
          continue
1611
        for script, hkr, output in res.payload:
1612
          test = hkr == constants.HKR_FAIL
1613
          self._ErrorIf(test, self.ENODEHOOKS, node_name,
1614
                        "Script %s failed, output:", script)
1615
          if test:
1616
            output = indent_re.sub('      ', output)
1617
            feedback_fn("%s" % output)
1618
            lu_result = 1
1619

    
1620
      return lu_result
1621

    
1622

    
1623
class LUVerifyDisks(NoHooksLU):
1624
  """Verifies the cluster disks status.
1625

1626
  """
1627
  _OP_REQP = []
1628
  REQ_BGL = False
1629

    
1630
  def ExpandNames(self):
1631
    self.needed_locks = {
1632
      locking.LEVEL_NODE: locking.ALL_SET,
1633
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1634
    }
1635
    self.share_locks = dict.fromkeys(locking.LEVELS, 1)
1636

    
1637
  def CheckPrereq(self):
1638
    """Check prerequisites.
1639

1640
    This has no prerequisites.
1641

1642
    """
1643
    pass
1644

    
1645
  def Exec(self, feedback_fn):
1646
    """Verify integrity of cluster disks.
1647

1648
    @rtype: tuple of three items
1649
    @return: a tuple of (dict of node-to-node_error, list of instances
1650
        which need activate-disks, dict of instance: (node, volume) for
1651
        missing volumes
1652

1653
    """
1654
    result = res_nodes, res_instances, res_missing = {}, [], {}
1655

    
1656
    vg_name = self.cfg.GetVGName()
1657
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1658
    instances = [self.cfg.GetInstanceInfo(name)
1659
                 for name in self.cfg.GetInstanceList()]
1660

    
1661
    nv_dict = {}
1662
    for inst in instances:
1663
      inst_lvs = {}
1664
      if (not inst.admin_up or
1665
          inst.disk_template not in constants.DTS_NET_MIRROR):
1666
        continue
1667
      inst.MapLVsByNode(inst_lvs)
1668
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1669
      for node, vol_list in inst_lvs.iteritems():
1670
        for vol in vol_list:
1671
          nv_dict[(node, vol)] = inst
1672

    
1673
    if not nv_dict:
1674
      return result
1675

    
1676
    node_lvs = self.rpc.call_lv_list(nodes, vg_name)
1677

    
1678
    for node in nodes:
1679
      # node_volume
1680
      node_res = node_lvs[node]
1681
      if node_res.offline:
1682
        continue
1683
      msg = node_res.fail_msg
1684
      if msg:
1685
        logging.warning("Error enumerating LVs on node %s: %s", node, msg)
1686
        res_nodes[node] = msg
1687
        continue
1688

    
1689
      lvs = node_res.payload
1690
      for lv_name, (_, _, lv_online) in lvs.items():
1691
        inst = nv_dict.pop((node, lv_name), None)
1692
        if (not lv_online and inst is not None
1693
            and inst.name not in res_instances):
1694
          res_instances.append(inst.name)
1695

    
1696
    # any leftover items in nv_dict are missing LVs, let's arrange the
1697
    # data better
1698
    for key, inst in nv_dict.iteritems():
1699
      if inst.name not in res_missing:
1700
        res_missing[inst.name] = []
1701
      res_missing[inst.name].append(key)
1702

    
1703
    return result
1704

    
1705

    
1706
class LURepairDiskSizes(NoHooksLU):
1707
  """Verifies the cluster disks sizes.
1708

1709
  """
1710
  _OP_REQP = ["instances"]
1711
  REQ_BGL = False
1712

    
1713
  def ExpandNames(self):
1714
    if not isinstance(self.op.instances, list):
1715
      raise errors.OpPrereqError("Invalid argument type 'instances'",
1716
                                 errors.ECODE_INVAL)
1717

    
1718
    if self.op.instances:
1719
      self.wanted_names = []
1720
      for name in self.op.instances:
1721
        full_name = self.cfg.ExpandInstanceName(name)
1722
        if full_name is None:
1723
          raise errors.OpPrereqError("Instance '%s' not known" % name,
1724
                                     errors.ECODE_NOENT)
1725
        self.wanted_names.append(full_name)
1726
      self.needed_locks = {
1727
        locking.LEVEL_NODE: [],
1728
        locking.LEVEL_INSTANCE: self.wanted_names,
1729
        }
1730
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
1731
    else:
1732
      self.wanted_names = None
1733
      self.needed_locks = {
1734
        locking.LEVEL_NODE: locking.ALL_SET,
1735
        locking.LEVEL_INSTANCE: locking.ALL_SET,
1736
        }
1737
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1738

    
1739
  def DeclareLocks(self, level):
1740
    if level == locking.LEVEL_NODE and self.wanted_names is not None:
1741
      self._LockInstancesNodes(primary_only=True)
1742

    
1743
  def CheckPrereq(self):
1744
    """Check prerequisites.
1745

1746
    This only checks the optional instance list against the existing names.
1747

1748
    """
1749
    if self.wanted_names is None:
1750
      self.wanted_names = self.acquired_locks[locking.LEVEL_INSTANCE]
1751

    
1752
    self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
1753
                             in self.wanted_names]
1754

    
1755
  def _EnsureChildSizes(self, disk):
1756
    """Ensure children of the disk have the needed disk size.
1757

1758
    This is valid mainly for DRBD8 and fixes an issue where the
1759
    children have smaller disk size.
1760

1761
    @param disk: an L{ganeti.objects.Disk} object
1762

1763
    """
1764
    if disk.dev_type == constants.LD_DRBD8:
1765
      assert disk.children, "Empty children for DRBD8?"
1766
      fchild = disk.children[0]
1767
      mismatch = fchild.size < disk.size
1768
      if mismatch:
1769
        self.LogInfo("Child disk has size %d, parent %d, fixing",
1770
                     fchild.size, disk.size)
1771
        fchild.size = disk.size
1772

    
1773
      # and we recurse on this child only, not on the metadev
1774
      return self._EnsureChildSizes(fchild) or mismatch
1775
    else:
1776
      return False
1777

    
1778
  def Exec(self, feedback_fn):
1779
    """Verify the size of cluster disks.
1780

1781
    """
1782
    # TODO: check child disks too
1783
    # TODO: check differences in size between primary/secondary nodes
1784
    per_node_disks = {}
1785
    for instance in self.wanted_instances:
1786
      pnode = instance.primary_node
1787
      if pnode not in per_node_disks:
1788
        per_node_disks[pnode] = []
1789
      for idx, disk in enumerate(instance.disks):
1790
        per_node_disks[pnode].append((instance, idx, disk))
1791

    
1792
    changed = []
1793
    for node, dskl in per_node_disks.items():
1794
      newl = [v[2].Copy() for v in dskl]
1795
      for dsk in newl:
1796
        self.cfg.SetDiskID(dsk, node)
1797
      result = self.rpc.call_blockdev_getsizes(node, newl)
1798
      if result.fail_msg:
1799
        self.LogWarning("Failure in blockdev_getsizes call to node"
1800
                        " %s, ignoring", node)
1801
        continue
1802
      if len(result.data) != len(dskl):
1803
        self.LogWarning("Invalid result from node %s, ignoring node results",
1804
                        node)
1805
        continue
1806
      for ((instance, idx, disk), size) in zip(dskl, result.data):
1807
        if size is None:
1808
          self.LogWarning("Disk %d of instance %s did not return size"
1809
                          " information, ignoring", idx, instance.name)
1810
          continue
1811
        if not isinstance(size, (int, long)):
1812
          self.LogWarning("Disk %d of instance %s did not return valid"
1813
                          " size information, ignoring", idx, instance.name)
1814
          continue
1815
        size = size >> 20
1816
        if size != disk.size:
1817
          self.LogInfo("Disk %d of instance %s has mismatched size,"
1818
                       " correcting: recorded %d, actual %d", idx,
1819
                       instance.name, disk.size, size)
1820
          disk.size = size
1821
          self.cfg.Update(instance, feedback_fn)
1822
          changed.append((instance.name, idx, size))
1823
        if self._EnsureChildSizes(disk):
1824
          self.cfg.Update(instance, feedback_fn)
1825
          changed.append((instance.name, idx, disk.size))
1826
    return changed
1827

    
1828

    
1829
class LURenameCluster(LogicalUnit):
1830
  """Rename the cluster.
1831

1832
  """
1833
  HPATH = "cluster-rename"
1834
  HTYPE = constants.HTYPE_CLUSTER
1835
  _OP_REQP = ["name"]
1836

    
1837
  def BuildHooksEnv(self):
1838
    """Build hooks env.
1839

1840
    """
1841
    env = {
1842
      "OP_TARGET": self.cfg.GetClusterName(),
1843
      "NEW_NAME": self.op.name,
1844
      }
1845
    mn = self.cfg.GetMasterNode()
1846
    all_nodes = self.cfg.GetNodeList()
1847
    return env, [mn], all_nodes
1848

    
1849
  def CheckPrereq(self):
1850
    """Verify that the passed name is a valid one.
1851

1852
    """
1853
    hostname = utils.GetHostInfo(self.op.name)
1854

    
1855
    new_name = hostname.name
1856
    self.ip = new_ip = hostname.ip
1857
    old_name = self.cfg.GetClusterName()
1858
    old_ip = self.cfg.GetMasterIP()
1859
    if new_name == old_name and new_ip == old_ip:
1860
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1861
                                 " cluster has changed",
1862
                                 errors.ECODE_INVAL)
1863
    if new_ip != old_ip:
1864
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1865
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1866
                                   " reachable on the network. Aborting." %
1867
                                   new_ip, errors.ECODE_NOTUNIQUE)
1868

    
1869
    self.op.name = new_name
1870

    
1871
  def Exec(self, feedback_fn):
1872
    """Rename the cluster.
1873

1874
    """
1875
    clustername = self.op.name
1876
    ip = self.ip
1877

    
1878
    # shutdown the master IP
1879
    master = self.cfg.GetMasterNode()
1880
    result = self.rpc.call_node_stop_master(master, False)
1881
    result.Raise("Could not disable the master role")
1882

    
1883
    try:
1884
      cluster = self.cfg.GetClusterInfo()
1885
      cluster.cluster_name = clustername
1886
      cluster.master_ip = ip
1887
      self.cfg.Update(cluster, feedback_fn)
1888

    
1889
      # update the known hosts file
1890
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1891
      node_list = self.cfg.GetNodeList()
1892
      try:
1893
        node_list.remove(master)
1894
      except ValueError:
1895
        pass
1896
      result = self.rpc.call_upload_file(node_list,
1897
                                         constants.SSH_KNOWN_HOSTS_FILE)
1898
      for to_node, to_result in result.iteritems():
1899
        msg = to_result.fail_msg
1900
        if msg:
1901
          msg = ("Copy of file %s to node %s failed: %s" %
1902
                 (constants.SSH_KNOWN_HOSTS_FILE, to_node, msg))
1903
          self.proc.LogWarning(msg)
1904

    
1905
    finally:
1906
      result = self.rpc.call_node_start_master(master, False, False)
1907
      msg = result.fail_msg
1908
      if msg:
1909
        self.LogWarning("Could not re-enable the master role on"
1910
                        " the master, please restart manually: %s", msg)
1911

    
1912

    
1913
def _RecursiveCheckIfLVMBased(disk):
1914
  """Check if the given disk or its children are lvm-based.
1915

1916
  @type disk: L{objects.Disk}
1917
  @param disk: the disk to check
1918
  @rtype: boolean
1919
  @return: boolean indicating whether a LD_LV dev_type was found or not
1920

1921
  """
1922
  if disk.children:
1923
    for chdisk in disk.children:
1924
      if _RecursiveCheckIfLVMBased(chdisk):
1925
        return True
1926
  return disk.dev_type == constants.LD_LV
1927

    
1928

    
1929
class LUSetClusterParams(LogicalUnit):
1930
  """Change the parameters of the cluster.
1931

1932
  """
1933
  HPATH = "cluster-modify"
1934
  HTYPE = constants.HTYPE_CLUSTER
1935
  _OP_REQP = []
1936
  REQ_BGL = False
1937

    
1938
  def CheckArguments(self):
1939
    """Check parameters
1940

1941
    """
1942
    if not hasattr(self.op, "candidate_pool_size"):
1943
      self.op.candidate_pool_size = None
1944
    if self.op.candidate_pool_size is not None:
1945
      try:
1946
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1947
      except (ValueError, TypeError), err:
1948
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1949
                                   str(err), errors.ECODE_INVAL)
1950
      if self.op.candidate_pool_size < 1:
1951
        raise errors.OpPrereqError("At least one master candidate needed",
1952
                                   errors.ECODE_INVAL)
1953

    
1954
  def ExpandNames(self):
1955
    # FIXME: in the future maybe other cluster params won't require checking on
1956
    # all nodes to be modified.
1957
    self.needed_locks = {
1958
      locking.LEVEL_NODE: locking.ALL_SET,
1959
    }
1960
    self.share_locks[locking.LEVEL_NODE] = 1
1961

    
1962
  def BuildHooksEnv(self):
1963
    """Build hooks env.
1964

1965
    """
1966
    env = {
1967
      "OP_TARGET": self.cfg.GetClusterName(),
1968
      "NEW_VG_NAME": self.op.vg_name,
1969
      }
1970
    mn = self.cfg.GetMasterNode()
1971
    return env, [mn], [mn]
1972

    
1973
  def CheckPrereq(self):
1974
    """Check prerequisites.
1975

1976
    This checks whether the given params don't conflict and
1977
    if the given volume group is valid.
1978

1979
    """
1980
    if self.op.vg_name is not None and not self.op.vg_name:
1981
      instances = self.cfg.GetAllInstancesInfo().values()
1982
      for inst in instances:
1983
        for disk in inst.disks:
1984
          if _RecursiveCheckIfLVMBased(disk):
1985
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1986
                                       " lvm-based instances exist",
1987
                                       errors.ECODE_INVAL)
1988

    
1989
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1990

    
1991
    # if vg_name not None, checks given volume group on all nodes
1992
    if self.op.vg_name:
1993
      vglist = self.rpc.call_vg_list(node_list)
1994
      for node in node_list:
1995
        msg = vglist[node].fail_msg
1996
        if msg:
1997
          # ignoring down node
1998
          self.LogWarning("Error while gathering data on node %s"
1999
                          " (ignoring node): %s", node, msg)
2000
          continue
2001
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].payload,
2002
                                              self.op.vg_name,
2003
                                              constants.MIN_VG_SIZE)
2004
        if vgstatus:
2005
          raise errors.OpPrereqError("Error on node '%s': %s" %
2006
                                     (node, vgstatus), errors.ECODE_ENVIRON)
2007

    
2008
    self.cluster = cluster = self.cfg.GetClusterInfo()
2009
    # validate params changes
2010
    if self.op.beparams:
2011
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
2012
      self.new_beparams = objects.FillDict(
2013
        cluster.beparams[constants.PP_DEFAULT], self.op.beparams)
2014

    
2015
    if self.op.nicparams:
2016
      utils.ForceDictType(self.op.nicparams, constants.NICS_PARAMETER_TYPES)
2017
      self.new_nicparams = objects.FillDict(
2018
        cluster.nicparams[constants.PP_DEFAULT], self.op.nicparams)
2019
      objects.NIC.CheckParameterSyntax(self.new_nicparams)
2020
      nic_errors = []
2021

    
2022
      # check all instances for consistency
2023
      for instance in self.cfg.GetAllInstancesInfo().values():
2024
        for nic_idx, nic in enumerate(instance.nics):
2025
          params_copy = copy.deepcopy(nic.nicparams)
2026
          params_filled = objects.FillDict(self.new_nicparams, params_copy)
2027

    
2028
          # check parameter syntax
2029
          try:
2030
            objects.NIC.CheckParameterSyntax(params_filled)
2031
          except errors.ConfigurationError, err:
2032
            nic_errors.append("Instance %s, nic/%d: %s" %
2033
                              (instance.name, nic_idx, err))
2034

    
2035
          # if we're moving instances to routed, check that they have an ip
2036
          target_mode = params_filled[constants.NIC_MODE]
2037
          if target_mode == constants.NIC_MODE_ROUTED and not nic.ip:
2038
            nic_errors.append("Instance %s, nic/%d: routed nick with no ip" %
2039
                              (instance.name, nic_idx))
2040
      if nic_errors:
2041
        raise errors.OpPrereqError("Cannot apply the change, errors:\n%s" %
2042
                                   "\n".join(nic_errors))
2043

    
2044
    # hypervisor list/parameters
2045
    self.new_hvparams = objects.FillDict(cluster.hvparams, {})
2046
    if self.op.hvparams:
2047
      if not isinstance(self.op.hvparams, dict):
2048
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input",
2049
                                   errors.ECODE_INVAL)
2050
      for hv_name, hv_dict in self.op.hvparams.items():
2051
        if hv_name not in self.new_hvparams:
2052
          self.new_hvparams[hv_name] = hv_dict
2053
        else:
2054
          self.new_hvparams[hv_name].update(hv_dict)
2055

    
2056
    if self.op.enabled_hypervisors is not None:
2057
      self.hv_list = self.op.enabled_hypervisors
2058
      if not self.hv_list:
2059
        raise errors.OpPrereqError("Enabled hypervisors list must contain at"
2060
                                   " least one member",
2061
                                   errors.ECODE_INVAL)
2062
      invalid_hvs = set(self.hv_list) - constants.HYPER_TYPES
2063
      if invalid_hvs:
2064
        raise errors.OpPrereqError("Enabled hypervisors contains invalid"
2065
                                   " entries: %s" %
2066
                                   utils.CommaJoin(invalid_hvs),
2067
                                   errors.ECODE_INVAL)
2068
    else:
2069
      self.hv_list = cluster.enabled_hypervisors
2070

    
2071
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
2072
      # either the enabled list has changed, or the parameters have, validate
2073
      for hv_name, hv_params in self.new_hvparams.items():
2074
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
2075
            (self.op.enabled_hypervisors and
2076
             hv_name in self.op.enabled_hypervisors)):
2077
          # either this is a new hypervisor, or its parameters have changed
2078
          hv_class = hypervisor.GetHypervisor(hv_name)
2079
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
2080
          hv_class.CheckParameterSyntax(hv_params)
2081
          _CheckHVParams(self, node_list, hv_name, hv_params)
2082

    
2083
  def Exec(self, feedback_fn):
2084
    """Change the parameters of the cluster.
2085

2086
    """
2087
    if self.op.vg_name is not None:
2088
      new_volume = self.op.vg_name
2089
      if not new_volume:
2090
        new_volume = None
2091
      if new_volume != self.cfg.GetVGName():
2092
        self.cfg.SetVGName(new_volume)
2093
      else:
2094
        feedback_fn("Cluster LVM configuration already in desired"
2095
                    " state, not changing")
2096
    if self.op.hvparams:
2097
      self.cluster.hvparams = self.new_hvparams
2098
    if self.op.enabled_hypervisors is not None:
2099
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
2100
    if self.op.beparams:
2101
      self.cluster.beparams[constants.PP_DEFAULT] = self.new_beparams
2102
    if self.op.nicparams:
2103
      self.cluster.nicparams[constants.PP_DEFAULT] = self.new_nicparams
2104

    
2105
    if self.op.candidate_pool_size is not None:
2106
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
2107
      # we need to update the pool size here, otherwise the save will fail
2108
      _AdjustCandidatePool(self, [])
2109

    
2110
    self.cfg.Update(self.cluster, feedback_fn)
2111

    
2112

    
2113
def _RedistributeAncillaryFiles(lu, additional_nodes=None):
2114
  """Distribute additional files which are part of the cluster configuration.
2115

2116
  ConfigWriter takes care of distributing the config and ssconf files, but
2117
  there are more files which should be distributed to all nodes. This function
2118
  makes sure those are copied.
2119

2120
  @param lu: calling logical unit
2121
  @param additional_nodes: list of nodes not in the config to distribute to
2122

2123
  """
2124
  # 1. Gather target nodes
2125
  myself = lu.cfg.GetNodeInfo(lu.cfg.GetMasterNode())
2126
  dist_nodes = lu.cfg.GetNodeList()
2127
  if additional_nodes is not None:
2128
    dist_nodes.extend(additional_nodes)
2129
  if myself.name in dist_nodes:
2130
    dist_nodes.remove(myself.name)
2131

    
2132
  # 2. Gather files to distribute
2133
  dist_files = set([constants.ETC_HOSTS,
2134
                    constants.SSH_KNOWN_HOSTS_FILE,
2135
                    constants.RAPI_CERT_FILE,
2136
                    constants.RAPI_USERS_FILE,
2137
                    constants.HMAC_CLUSTER_KEY,
2138
                   ])
2139

    
2140
  enabled_hypervisors = lu.cfg.GetClusterInfo().enabled_hypervisors
2141
  for hv_name in enabled_hypervisors:
2142
    hv_class = hypervisor.GetHypervisor(hv_name)
2143
    dist_files.update(hv_class.GetAncillaryFiles())
2144

    
2145
  # 3. Perform the files upload
2146
  for fname in dist_files:
2147
    if os.path.exists(fname):
2148
      result = lu.rpc.call_upload_file(dist_nodes, fname)
2149
      for to_node, to_result in result.items():
2150
        msg = to_result.fail_msg
2151
        if msg:
2152
          msg = ("Copy of file %s to node %s failed: %s" %
2153
                 (fname, to_node, msg))
2154
          lu.proc.LogWarning(msg)
2155

    
2156

    
2157
class LURedistributeConfig(NoHooksLU):
2158
  """Force the redistribution of cluster configuration.
2159

2160
  This is a very simple LU.
2161

2162
  """
2163
  _OP_REQP = []
2164
  REQ_BGL = False
2165

    
2166
  def ExpandNames(self):
2167
    self.needed_locks = {
2168
      locking.LEVEL_NODE: locking.ALL_SET,
2169
    }
2170
    self.share_locks[locking.LEVEL_NODE] = 1
2171

    
2172
  def CheckPrereq(self):
2173
    """Check prerequisites.
2174

2175
    """
2176

    
2177
  def Exec(self, feedback_fn):
2178
    """Redistribute the configuration.
2179

2180
    """
2181
    self.cfg.Update(self.cfg.GetClusterInfo(), feedback_fn)
2182
    _RedistributeAncillaryFiles(self)
2183

    
2184

    
2185
def _WaitForSync(lu, instance, oneshot=False):
2186
  """Sleep and poll for an instance's disk to sync.
2187

2188
  """
2189
  if not instance.disks:
2190
    return True
2191

    
2192
  if not oneshot:
2193
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
2194

    
2195
  node = instance.primary_node
2196

    
2197
  for dev in instance.disks:
2198
    lu.cfg.SetDiskID(dev, node)
2199

    
2200
  # TODO: Convert to utils.Retry
2201

    
2202
  retries = 0
2203
  degr_retries = 10 # in seconds, as we sleep 1 second each time
2204
  while True:
2205
    max_time = 0
2206
    done = True
2207
    cumul_degraded = False
2208
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
2209
    msg = rstats.fail_msg
2210
    if msg:
2211
      lu.LogWarning("Can't get any data from node %s: %s", node, msg)
2212
      retries += 1
2213
      if retries >= 10:
2214
        raise errors.RemoteError("Can't contact node %s for mirror data,"
2215
                                 " aborting." % node)
2216
      time.sleep(6)
2217
      continue
2218
    rstats = rstats.payload
2219
    retries = 0
2220
    for i, mstat in enumerate(rstats):
2221
      if mstat is None:
2222
        lu.LogWarning("Can't compute data for node %s/%s",
2223
                           node, instance.disks[i].iv_name)
2224
        continue
2225

    
2226
      cumul_degraded = (cumul_degraded or
2227
                        (mstat.is_degraded and mstat.sync_percent is None))
2228
      if mstat.sync_percent is not None:
2229
        done = False
2230
        if mstat.estimated_time is not None:
2231
          rem_time = "%d estimated seconds remaining" % mstat.estimated_time
2232
          max_time = mstat.estimated_time
2233
        else:
2234
          rem_time = "no time estimate"
2235
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
2236
                        (instance.disks[i].iv_name, mstat.sync_percent,
2237
                         rem_time))
2238

    
2239
    # if we're done but degraded, let's do a few small retries, to
2240
    # make sure we see a stable and not transient situation; therefore
2241
    # we force restart of the loop
2242
    if (done or oneshot) and cumul_degraded and degr_retries > 0:
2243
      logging.info("Degraded disks found, %d retries left", degr_retries)
2244
      degr_retries -= 1
2245
      time.sleep(1)
2246
      continue
2247

    
2248
    if done or oneshot:
2249
      break
2250

    
2251
    time.sleep(min(60, max_time))
2252

    
2253
  if done:
2254
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
2255
  return not cumul_degraded
2256

    
2257

    
2258
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
2259
  """Check that mirrors are not degraded.
2260

2261
  The ldisk parameter, if True, will change the test from the
2262
  is_degraded attribute (which represents overall non-ok status for
2263
  the device(s)) to the ldisk (representing the local storage status).
2264

2265
  """
2266
  lu.cfg.SetDiskID(dev, node)
2267

    
2268
  result = True
2269

    
2270
  if on_primary or dev.AssembleOnSecondary():
2271
    rstats = lu.rpc.call_blockdev_find(node, dev)
2272
    msg = rstats.fail_msg
2273
    if msg:
2274
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
2275
      result = False
2276
    elif not rstats.payload:
2277
      lu.LogWarning("Can't find disk on node %s", node)
2278
      result = False
2279
    else:
2280
      if ldisk:
2281
        result = result and rstats.payload.ldisk_status == constants.LDS_OKAY
2282
      else:
2283
        result = result and not rstats.payload.is_degraded
2284

    
2285
  if dev.children:
2286
    for child in dev.children:
2287
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
2288

    
2289
  return result
2290

    
2291

    
2292
class LUDiagnoseOS(NoHooksLU):
2293
  """Logical unit for OS diagnose/query.
2294

2295
  """
2296
  _OP_REQP = ["output_fields", "names"]
2297
  REQ_BGL = False
2298
  _FIELDS_STATIC = utils.FieldSet()
2299
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status", "variants")
2300
  # Fields that need calculation of global os validity
2301
  _FIELDS_NEEDVALID = frozenset(["valid", "variants"])
2302

    
2303
  def ExpandNames(self):
2304
    if self.op.names:
2305
      raise errors.OpPrereqError("Selective OS query not supported",
2306
                                 errors.ECODE_INVAL)
2307

    
2308
    _CheckOutputFields(static=self._FIELDS_STATIC,
2309
                       dynamic=self._FIELDS_DYNAMIC,
2310
                       selected=self.op.output_fields)
2311

    
2312
    # Lock all nodes, in shared mode
2313
    # Temporary removal of locks, should be reverted later
2314
    # TODO: reintroduce locks when they are lighter-weight
2315
    self.needed_locks = {}
2316
    #self.share_locks[locking.LEVEL_NODE] = 1
2317
    #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2318

    
2319
  def CheckPrereq(self):
2320
    """Check prerequisites.
2321

2322
    """
2323

    
2324
  @staticmethod
2325
  def _DiagnoseByOS(rlist):
2326
    """Remaps a per-node return list into an a per-os per-node dictionary
2327

2328
    @param rlist: a map with node names as keys and OS objects as values
2329

2330
    @rtype: dict
2331
    @return: a dictionary with osnames as keys and as value another map, with
2332
        nodes as keys and tuples of (path, status, diagnose) as values, eg::
2333

2334
          {"debian-etch": {"node1": [(/usr/lib/..., True, ""),
2335
                                     (/srv/..., False, "invalid api")],
2336
                           "node2": [(/srv/..., True, "")]}
2337
          }
2338

2339
    """
2340
    all_os = {}
2341
    # we build here the list of nodes that didn't fail the RPC (at RPC
2342
    # level), so that nodes with a non-responding node daemon don't
2343
    # make all OSes invalid
2344
    good_nodes = [node_name for node_name in rlist
2345
                  if not rlist[node_name].fail_msg]
2346
    for node_name, nr in rlist.items():
2347
      if nr.fail_msg or not nr.payload:
2348
        continue
2349
      for name, path, status, diagnose, variants in nr.payload:
2350
        if name not in all_os:
2351
          # build a list of nodes for this os containing empty lists
2352
          # for each node in node_list
2353
          all_os[name] = {}
2354
          for nname in good_nodes:
2355
            all_os[name][nname] = []
2356
        all_os[name][node_name].append((path, status, diagnose, variants))
2357
    return all_os
2358

    
2359
  def Exec(self, feedback_fn):
2360
    """Compute the list of OSes.
2361

2362
    """
2363
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
2364
    node_data = self.rpc.call_os_diagnose(valid_nodes)
2365
    pol = self._DiagnoseByOS(node_data)
2366
    output = []
2367
    calc_valid = self._FIELDS_NEEDVALID.intersection(self.op.output_fields)
2368
    calc_variants = "variants" in self.op.output_fields
2369

    
2370
    for os_name, os_data in pol.items():
2371
      row = []
2372
      if calc_valid:
2373
        valid = True
2374
        variants = None
2375
        for osl in os_data.values():
2376
          valid = valid and osl and osl[0][1]
2377
          if not valid:
2378
            variants = None
2379
            break
2380
          if calc_variants:
2381
            node_variants = osl[0][3]
2382
            if variants is None:
2383
              variants = node_variants
2384
            else:
2385
              variants = [v for v in variants if v in node_variants]
2386

    
2387
      for field in self.op.output_fields:
2388
        if field == "name":
2389
          val = os_name
2390
        elif field == "valid":
2391
          val = valid
2392
        elif field == "node_status":
2393
          # this is just a copy of the dict
2394
          val = {}
2395
          for node_name, nos_list in os_data.items():
2396
            val[node_name] = nos_list
2397
        elif field == "variants":
2398
          val =  variants
2399
        else:
2400
          raise errors.ParameterError(field)
2401
        row.append(val)
2402
      output.append(row)
2403

    
2404
    return output
2405

    
2406

    
2407
class LURemoveNode(LogicalUnit):
2408
  """Logical unit for removing a node.
2409

2410
  """
2411
  HPATH = "node-remove"
2412
  HTYPE = constants.HTYPE_NODE
2413
  _OP_REQP = ["node_name"]
2414

    
2415
  def BuildHooksEnv(self):
2416
    """Build hooks env.
2417

2418
    This doesn't run on the target node in the pre phase as a failed
2419
    node would then be impossible to remove.
2420

2421
    """
2422
    env = {
2423
      "OP_TARGET": self.op.node_name,
2424
      "NODE_NAME": self.op.node_name,
2425
      }
2426
    all_nodes = self.cfg.GetNodeList()
2427
    if self.op.node_name in all_nodes:
2428
      all_nodes.remove(self.op.node_name)
2429
    return env, all_nodes, all_nodes
2430

    
2431
  def CheckPrereq(self):
2432
    """Check prerequisites.
2433

2434
    This checks:
2435
     - the node exists in the configuration
2436
     - it does not have primary or secondary instances
2437
     - it's not the master
2438

2439
    Any errors are signaled by raising errors.OpPrereqError.
2440

2441
    """
2442
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
2443
    if node is None:
2444
      raise errors.OpPrereqError("Node '%s' is unknown." % self.op.node_name,
2445
                                 errors.ECODE_NOENT)
2446

    
2447
    instance_list = self.cfg.GetInstanceList()
2448

    
2449
    masternode = self.cfg.GetMasterNode()
2450
    if node.name == masternode:
2451
      raise errors.OpPrereqError("Node is the master node,"
2452
                                 " you need to failover first.",
2453
                                 errors.ECODE_INVAL)
2454

    
2455
    for instance_name in instance_list:
2456
      instance = self.cfg.GetInstanceInfo(instance_name)
2457
      if node.name in instance.all_nodes:
2458
        raise errors.OpPrereqError("Instance %s is still running on the node,"
2459
                                   " please remove first." % instance_name,
2460
                                   errors.ECODE_INVAL)
2461
    self.op.node_name = node.name
2462
    self.node = node
2463

    
2464
  def Exec(self, feedback_fn):
2465
    """Removes the node from the cluster.
2466

2467
    """
2468
    node = self.node
2469
    logging.info("Stopping the node daemon and removing configs from node %s",
2470
                 node.name)
2471

    
2472
    modify_ssh_setup = self.cfg.GetClusterInfo().modify_ssh_setup
2473

    
2474
    # Promote nodes to master candidate as needed
2475
    _AdjustCandidatePool(self, exceptions=[node.name])
2476
    self.context.RemoveNode(node.name)
2477

    
2478
    # Run post hooks on the node before it's removed
2479
    hm = self.proc.hmclass(self.rpc.call_hooks_runner, self)
2480
    try:
2481
      hm.RunPhase(constants.HOOKS_PHASE_POST, [node.name])
2482
    except:
2483
      # pylint: disable-msg=W0702
2484
      self.LogWarning("Errors occurred running hooks on %s" % node.name)
2485

    
2486
    result = self.rpc.call_node_leave_cluster(node.name, modify_ssh_setup)
2487
    msg = result.fail_msg
2488
    if msg:
2489
      self.LogWarning("Errors encountered on the remote node while leaving"
2490
                      " the cluster: %s", msg)
2491

    
2492

    
2493
class LUQueryNodes(NoHooksLU):
2494
  """Logical unit for querying nodes.
2495

2496
  """
2497
  # pylint: disable-msg=W0142
2498
  _OP_REQP = ["output_fields", "names", "use_locking"]
2499
  REQ_BGL = False
2500

    
2501
  _SIMPLE_FIELDS = ["name", "serial_no", "ctime", "mtime", "uuid",
2502
                    "master_candidate", "offline", "drained"]
2503

    
2504
  _FIELDS_DYNAMIC = utils.FieldSet(
2505
    "dtotal", "dfree",
2506
    "mtotal", "mnode", "mfree",
2507
    "bootid",
2508
    "ctotal", "cnodes", "csockets",
2509
    )
2510

    
2511
  _FIELDS_STATIC = utils.FieldSet(*[
2512
    "pinst_cnt", "sinst_cnt",
2513
    "pinst_list", "sinst_list",
2514
    "pip", "sip", "tags",
2515
    "master",
2516
    "role"] + _SIMPLE_FIELDS
2517
    )
2518

    
2519
  def ExpandNames(self):
2520
    _CheckOutputFields(static=self._FIELDS_STATIC,
2521
                       dynamic=self._FIELDS_DYNAMIC,
2522
                       selected=self.op.output_fields)
2523

    
2524
    self.needed_locks = {}
2525
    self.share_locks[locking.LEVEL_NODE] = 1
2526

    
2527
    if self.op.names:
2528
      self.wanted = _GetWantedNodes(self, self.op.names)
2529
    else:
2530
      self.wanted = locking.ALL_SET
2531

    
2532
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
2533
    self.do_locking = self.do_node_query and self.op.use_locking
2534
    if self.do_locking:
2535
      # if we don't request only static fields, we need to lock the nodes
2536
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
2537

    
2538
  def CheckPrereq(self):
2539
    """Check prerequisites.
2540

2541
    """
2542
    # The validation of the node list is done in the _GetWantedNodes,
2543
    # if non empty, and if empty, there's no validation to do
2544
    pass
2545

    
2546
  def Exec(self, feedback_fn):
2547
    """Computes the list of nodes and their attributes.
2548

2549
    """
2550
    all_info = self.cfg.GetAllNodesInfo()
2551
    if self.do_locking:
2552
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
2553
    elif self.wanted != locking.ALL_SET:
2554
      nodenames = self.wanted
2555
      missing = set(nodenames).difference(all_info.keys())
2556
      if missing:
2557
        raise errors.OpExecError(
2558
          "Some nodes were removed before retrieving their data: %s" % missing)
2559
    else:
2560
      nodenames = all_info.keys()
2561

    
2562
    nodenames = utils.NiceSort(nodenames)
2563
    nodelist = [all_info[name] for name in nodenames]
2564

    
2565
    # begin data gathering
2566

    
2567
    if self.do_node_query:
2568
      live_data = {}
2569
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
2570
                                          self.cfg.GetHypervisorType())
2571
      for name in nodenames:
2572
        nodeinfo = node_data[name]
2573
        if not nodeinfo.fail_msg and nodeinfo.payload:
2574
          nodeinfo = nodeinfo.payload
2575
          fn = utils.TryConvert
2576
          live_data[name] = {
2577
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
2578
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
2579
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
2580
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
2581
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
2582
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
2583
            "bootid": nodeinfo.get('bootid', None),
2584
            "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
2585
            "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
2586
            }
2587
        else:
2588
          live_data[name] = {}
2589
    else:
2590
      live_data = dict.fromkeys(nodenames, {})
2591

    
2592
    node_to_primary = dict([(name, set()) for name in nodenames])
2593
    node_to_secondary = dict([(name, set()) for name in nodenames])
2594

    
2595
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
2596
                             "sinst_cnt", "sinst_list"))
2597
    if inst_fields & frozenset(self.op.output_fields):
2598
      inst_data = self.cfg.GetAllInstancesInfo()
2599

    
2600
      for inst in inst_data.values():
2601
        if inst.primary_node in node_to_primary:
2602
          node_to_primary[inst.primary_node].add(inst.name)
2603
        for secnode in inst.secondary_nodes:
2604
          if secnode in node_to_secondary:
2605
            node_to_secondary[secnode].add(inst.name)
2606

    
2607
    master_node = self.cfg.GetMasterNode()
2608

    
2609
    # end data gathering
2610

    
2611
    output = []
2612
    for node in nodelist:
2613
      node_output = []
2614
      for field in self.op.output_fields:
2615
        if field in self._SIMPLE_FIELDS:
2616
          val = getattr(node, field)
2617
        elif field == "pinst_list":
2618
          val = list(node_to_primary[node.name])
2619
        elif field == "sinst_list":
2620
          val = list(node_to_secondary[node.name])
2621
        elif field == "pinst_cnt":
2622
          val = len(node_to_primary[node.name])
2623
        elif field == "sinst_cnt":
2624
          val = len(node_to_secondary[node.name])
2625
        elif field == "pip":
2626
          val = node.primary_ip
2627
        elif field == "sip":
2628
          val = node.secondary_ip
2629
        elif field == "tags":
2630
          val = list(node.GetTags())
2631
        elif field == "master":
2632
          val = node.name == master_node
2633
        elif self._FIELDS_DYNAMIC.Matches(field):
2634
          val = live_data[node.name].get(field, None)
2635
        elif field == "role":
2636
          if node.name == master_node:
2637
            val = "M"
2638
          elif node.master_candidate:
2639
            val = "C"
2640
          elif node.drained:
2641
            val = "D"
2642
          elif node.offline:
2643
            val = "O"
2644
          else:
2645
            val = "R"
2646
        else:
2647
          raise errors.ParameterError(field)
2648
        node_output.append(val)
2649
      output.append(node_output)
2650

    
2651
    return output
2652

    
2653

    
2654
class LUQueryNodeVolumes(NoHooksLU):
2655
  """Logical unit for getting volumes on node(s).
2656

2657
  """
2658
  _OP_REQP = ["nodes", "output_fields"]
2659
  REQ_BGL = False
2660
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2661
  _FIELDS_STATIC = utils.FieldSet("node")
2662

    
2663
  def ExpandNames(self):
2664
    _CheckOutputFields(static=self._FIELDS_STATIC,
2665
                       dynamic=self._FIELDS_DYNAMIC,
2666
                       selected=self.op.output_fields)
2667

    
2668
    self.needed_locks = {}
2669
    self.share_locks[locking.LEVEL_NODE] = 1
2670
    if not self.op.nodes:
2671
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2672
    else:
2673
      self.needed_locks[locking.LEVEL_NODE] = \
2674
        _GetWantedNodes(self, self.op.nodes)
2675

    
2676
  def CheckPrereq(self):
2677
    """Check prerequisites.
2678

2679
    This checks that the fields required are valid output fields.
2680

2681
    """
2682
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2683

    
2684
  def Exec(self, feedback_fn):
2685
    """Computes the list of nodes and their attributes.
2686

2687
    """
2688
    nodenames = self.nodes
2689
    volumes = self.rpc.call_node_volumes(nodenames)
2690

    
2691
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
2692
             in self.cfg.GetInstanceList()]
2693

    
2694
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2695

    
2696
    output = []
2697
    for node in nodenames:
2698
      nresult = volumes[node]
2699
      if nresult.offline:
2700
        continue
2701
      msg = nresult.fail_msg
2702
      if msg:
2703
        self.LogWarning("Can't compute volume data on node %s: %s", node, msg)
2704
        continue
2705

    
2706
      node_vols = nresult.payload[:]
2707
      node_vols.sort(key=lambda vol: vol['dev'])
2708

    
2709
      for vol in node_vols:
2710
        node_output = []
2711
        for field in self.op.output_fields:
2712
          if field == "node":
2713
            val = node
2714
          elif field == "phys":
2715
            val = vol['dev']
2716
          elif field == "vg":
2717
            val = vol['vg']
2718
          elif field == "name":
2719
            val = vol['name']
2720
          elif field == "size":
2721
            val = int(float(vol['size']))
2722
          elif field == "instance":
2723
            for inst in ilist:
2724
              if node not in lv_by_node[inst]:
2725
                continue
2726
              if vol['name'] in lv_by_node[inst][node]:
2727
                val = inst.name
2728
                break
2729
            else:
2730
              val = '-'
2731
          else:
2732
            raise errors.ParameterError(field)
2733
          node_output.append(str(val))
2734

    
2735
        output.append(node_output)
2736

    
2737
    return output
2738

    
2739

    
2740
class LUQueryNodeStorage(NoHooksLU):
2741
  """Logical unit for getting information on storage units on node(s).
2742

2743
  """
2744
  _OP_REQP = ["nodes", "storage_type", "output_fields"]
2745
  REQ_BGL = False
2746
  _FIELDS_STATIC = utils.FieldSet(constants.SF_NODE)
2747

    
2748
  def ExpandNames(self):
2749
    storage_type = self.op.storage_type
2750

    
2751
    if storage_type not in constants.VALID_STORAGE_TYPES:
2752
      raise errors.OpPrereqError("Unknown storage type: %s" % storage_type,
2753
                                 errors.ECODE_INVAL)
2754

    
2755
    _CheckOutputFields(static=self._FIELDS_STATIC,
2756
                       dynamic=utils.FieldSet(*constants.VALID_STORAGE_FIELDS),
2757
                       selected=self.op.output_fields)
2758

    
2759
    self.needed_locks = {}
2760
    self.share_locks[locking.LEVEL_NODE] = 1
2761

    
2762
    if self.op.nodes:
2763
      self.needed_locks[locking.LEVEL_NODE] = \
2764
        _GetWantedNodes(self, self.op.nodes)
2765
    else:
2766
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2767

    
2768
  def CheckPrereq(self):
2769
    """Check prerequisites.
2770

2771
    This checks that the fields required are valid output fields.
2772

2773
    """
2774
    self.op.name = getattr(self.op, "name", None)
2775

    
2776
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2777

    
2778
  def Exec(self, feedback_fn):
2779
    """Computes the list of nodes and their attributes.
2780

2781
    """
2782
    # Always get name to sort by
2783
    if constants.SF_NAME in self.op.output_fields:
2784
      fields = self.op.output_fields[:]
2785
    else:
2786
      fields = [constants.SF_NAME] + self.op.output_fields
2787

    
2788
    # Never ask for node or type as it's only known to the LU
2789
    for extra in [constants.SF_NODE, constants.SF_TYPE]:
2790
      while extra in fields:
2791
        fields.remove(extra)
2792

    
2793
    field_idx = dict([(name, idx) for (idx, name) in enumerate(fields)])
2794
    name_idx = field_idx[constants.SF_NAME]
2795

    
2796
    st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
2797
    data = self.rpc.call_storage_list(self.nodes,
2798
                                      self.op.storage_type, st_args,
2799
                                      self.op.name, fields)
2800

    
2801
    result = []
2802

    
2803
    for node in utils.NiceSort(self.nodes):
2804
      nresult = data[node]
2805
      if nresult.offline:
2806
        continue
2807

    
2808
      msg = nresult.fail_msg
2809
      if msg:
2810
        self.LogWarning("Can't get storage data from node %s: %s", node, msg)
2811
        continue
2812

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

    
2815
      for name in utils.NiceSort(rows.keys()):
2816
        row = rows[name]
2817

    
2818
        out = []
2819

    
2820
        for field in self.op.output_fields:
2821
          if field == constants.SF_NODE:
2822
            val = node
2823
          elif field == constants.SF_TYPE:
2824
            val = self.op.storage_type
2825
          elif field in field_idx:
2826
            val = row[field_idx[field]]
2827
          else:
2828
            raise errors.ParameterError(field)
2829

    
2830
          out.append(val)
2831

    
2832
        result.append(out)
2833

    
2834
    return result
2835

    
2836

    
2837
class LUModifyNodeStorage(NoHooksLU):
2838
  """Logical unit for modifying a storage volume on a node.
2839

2840
  """
2841
  _OP_REQP = ["node_name", "storage_type", "name", "changes"]
2842
  REQ_BGL = False
2843

    
2844
  def CheckArguments(self):
2845
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2846
    if node_name is None:
2847
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name,
2848
                                 errors.ECODE_NOENT)
2849

    
2850
    self.op.node_name = node_name
2851

    
2852
    storage_type = self.op.storage_type
2853
    if storage_type not in constants.VALID_STORAGE_TYPES:
2854
      raise errors.OpPrereqError("Unknown storage type: %s" % storage_type,
2855
                                 errors.ECODE_INVAL)
2856

    
2857
  def ExpandNames(self):
2858
    self.needed_locks = {
2859
      locking.LEVEL_NODE: self.op.node_name,
2860
      }
2861

    
2862
  def CheckPrereq(self):
2863
    """Check prerequisites.
2864

2865
    """
2866
    storage_type = self.op.storage_type
2867

    
2868
    try:
2869
      modifiable = constants.MODIFIABLE_STORAGE_FIELDS[storage_type]
2870
    except KeyError:
2871
      raise errors.OpPrereqError("Storage units of type '%s' can not be"
2872
                                 " modified" % storage_type,
2873
                                 errors.ECODE_INVAL)
2874

    
2875
    diff = set(self.op.changes.keys()) - modifiable
2876
    if diff:
2877
      raise errors.OpPrereqError("The following fields can not be modified for"
2878
                                 " storage units of type '%s': %r" %
2879
                                 (storage_type, list(diff)),
2880
                                 errors.ECODE_INVAL)
2881

    
2882
  def Exec(self, feedback_fn):
2883
    """Computes the list of nodes and their attributes.
2884

2885
    """
2886
    st_args = _GetStorageTypeArgs(self.cfg, self.op.storage_type)
2887
    result = self.rpc.call_storage_modify(self.op.node_name,
2888
                                          self.op.storage_type, st_args,
2889
                                          self.op.name, self.op.changes)
2890
    result.Raise("Failed to modify storage unit '%s' on %s" %
2891
                 (self.op.name, self.op.node_name))
2892

    
2893

    
2894
class LUAddNode(LogicalUnit):
2895
  """Logical unit for adding node to the cluster.
2896

2897
  """
2898
  HPATH = "node-add"
2899
  HTYPE = constants.HTYPE_NODE
2900
  _OP_REQP = ["node_name"]
2901

    
2902
  def BuildHooksEnv(self):
2903
    """Build hooks env.
2904

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

2907
    """
2908
    env = {
2909
      "OP_TARGET": self.op.node_name,
2910
      "NODE_NAME": self.op.node_name,
2911
      "NODE_PIP": self.op.primary_ip,
2912
      "NODE_SIP": self.op.secondary_ip,
2913
      }
2914
    nodes_0 = self.cfg.GetNodeList()
2915
    nodes_1 = nodes_0 + [self.op.node_name, ]
2916
    return env, nodes_0, nodes_1
2917

    
2918
  def CheckPrereq(self):
2919
    """Check prerequisites.
2920

2921
    This checks:
2922
     - the new node is not already in the config
2923
     - it is resolvable
2924
     - its parameters (single/dual homed) matches the cluster
2925

2926
    Any errors are signaled by raising errors.OpPrereqError.
2927

2928
    """
2929
    node_name = self.op.node_name
2930
    cfg = self.cfg
2931

    
2932
    dns_data = utils.GetHostInfo(node_name)
2933

    
2934
    node = dns_data.name
2935
    primary_ip = self.op.primary_ip = dns_data.ip
2936
    secondary_ip = getattr(self.op, "secondary_ip", None)
2937
    if secondary_ip is None:
2938
      secondary_ip = primary_ip
2939
    if not utils.IsValidIP(secondary_ip):
2940
      raise errors.OpPrereqError("Invalid secondary IP given",
2941
                                 errors.ECODE_INVAL)
2942
    self.op.secondary_ip = secondary_ip
2943

    
2944
    node_list = cfg.GetNodeList()
2945
    if not self.op.readd and node in node_list:
2946
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2947
                                 node, errors.ECODE_EXISTS)
2948
    elif self.op.readd and node not in node_list:
2949
      raise errors.OpPrereqError("Node %s is not in the configuration" % node,
2950
                                 errors.ECODE_NOENT)
2951

    
2952
    for existing_node_name in node_list:
2953
      existing_node = cfg.GetNodeInfo(existing_node_name)
2954

    
2955
      if self.op.readd and node == existing_node_name:
2956
        if (existing_node.primary_ip != primary_ip or
2957
            existing_node.secondary_ip != secondary_ip):
2958
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2959
                                     " address configuration as before",
2960
                                     errors.ECODE_INVAL)
2961
        continue
2962

    
2963
      if (existing_node.primary_ip == primary_ip or
2964
          existing_node.secondary_ip == primary_ip or
2965
          existing_node.primary_ip == secondary_ip or
2966
          existing_node.secondary_ip == secondary_ip):
2967
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2968
                                   " existing node %s" % existing_node.name,
2969
                                   errors.ECODE_NOTUNIQUE)
2970

    
2971
    # check that the type of the node (single versus dual homed) is the
2972
    # same as for the master
2973
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2974
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2975
    newbie_singlehomed = secondary_ip == primary_ip
2976
    if master_singlehomed != newbie_singlehomed:
2977
      if master_singlehomed:
2978
        raise errors.OpPrereqError("The master has no private ip but the"
2979
                                   " new node has one",
2980
                                   errors.ECODE_INVAL)
2981
      else:
2982
        raise errors.OpPrereqError("The master has a private ip but the"
2983
                                   " new node doesn't have one",
2984
                                   errors.ECODE_INVAL)
2985

    
2986
    # checks reachability
2987
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2988
      raise errors.OpPrereqError("Node not reachable by ping",
2989
                                 errors.ECODE_ENVIRON)
2990

    
2991
    if not newbie_singlehomed:
2992
      # check reachability from my secondary ip to newbie's secondary ip
2993
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2994
                           source=myself.secondary_ip):
2995
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2996
                                   " based ping to noded port",
2997
                                   errors.ECODE_ENVIRON)
2998

    
2999
    if self.op.readd:
3000
      exceptions = [node]
3001
    else:
3002
      exceptions = []
3003

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

    
3006
    if self.op.readd:
3007
      self.new_node = self.cfg.GetNodeInfo(node)
3008
      assert self.new_node is not None, "Can't retrieve locked node %s" % node
3009
    else:
3010
      self.new_node = objects.Node(name=node,
3011
                                   primary_ip=primary_ip,
3012
                                   secondary_ip=secondary_ip,
3013
                                   master_candidate=self.master_candidate,
3014
                                   offline=False, drained=False)
3015

    
3016
  def Exec(self, feedback_fn):
3017
    """Adds the new node to the cluster.
3018

3019
    """
3020
    new_node = self.new_node
3021
    node = new_node.name
3022

    
3023
    # for re-adds, reset the offline/drained/master-candidate flags;
3024
    # we need to reset here, otherwise offline would prevent RPC calls
3025
    # later in the procedure; this also means that if the re-add
3026
    # fails, we are left with a non-offlined, broken node
3027
    if self.op.readd:
3028
      new_node.drained = new_node.offline = False # pylint: disable-msg=W0201
3029
      self.LogInfo("Readding a node, the offline/drained flags were reset")
3030
      # if we demote the node, we do cleanup later in the procedure
3031
      new_node.master_candidate = self.master_candidate
3032

    
3033
    # notify the user about any possible mc promotion
3034
    if new_node.master_candidate:
3035
      self.LogInfo("Node will be a master candidate")
3036

    
3037
    # check connectivity
3038
    result = self.rpc.call_version([node])[node]
3039
    result.Raise("Can't get version information from node %s" % node)
3040
    if constants.PROTOCOL_VERSION == result.payload:
3041
      logging.info("Communication to node %s fine, sw version %s match",
3042
                   node, result.payload)
3043
    else:
3044
      raise errors.OpExecError("Version mismatch master version %s,"
3045
                               " node version %s" %
3046
                               (constants.PROTOCOL_VERSION, result.payload))
3047

    
3048
    # setup ssh on node
3049
    if self.cfg.GetClusterInfo().modify_ssh_setup:
3050
      logging.info("Copy ssh key to node %s", node)
3051
      priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
3052
      keyarray = []
3053
      keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
3054
                  constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
3055
                  priv_key, pub_key]
3056

    
3057
      for i in keyfiles:
3058
        keyarray.append(utils.ReadFile(i))
3059

    
3060
      result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
3061
                                      keyarray[2], keyarray[3], keyarray[4],
3062
                                      keyarray[5])
3063
      result.Raise("Cannot transfer ssh keys to the new node")
3064

    
3065
    # Add node to our /etc/hosts, and add key to known_hosts
3066
    if self.cfg.GetClusterInfo().modify_etc_hosts:
3067
      utils.AddHostToEtcHosts(new_node.name)
3068

    
3069
    if new_node.secondary_ip != new_node.primary_ip:
3070
      result = self.rpc.call_node_has_ip_address(new_node.name,
3071
                                                 new_node.secondary_ip)
3072
      result.Raise("Failure checking secondary ip on node %s" % new_node.name,
3073
                   prereq=True, ecode=errors.ECODE_ENVIRON)
3074
      if not result.payload:
3075
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
3076
                                 " you gave (%s). Please fix and re-run this"
3077
                                 " command." % new_node.secondary_ip)
3078

    
3079
    node_verify_list = [self.cfg.GetMasterNode()]
3080
    node_verify_param = {
3081
      constants.NV_NODELIST: [node],
3082
      # TODO: do a node-net-test as well?
3083
    }
3084

    
3085
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
3086
                                       self.cfg.GetClusterName())
3087
    for verifier in node_verify_list:
3088
      result[verifier].Raise("Cannot communicate with node %s" % verifier)
3089
      nl_payload = result[verifier].payload[constants.NV_NODELIST]
3090
      if nl_payload:
3091
        for failed in nl_payload:
3092
          feedback_fn("ssh/hostname verification failed"
3093
                      " (checking from %s): %s" %
3094
                      (verifier, nl_payload[failed]))
3095
        raise errors.OpExecError("ssh/hostname verification failed.")
3096

    
3097
    if self.op.readd:
3098
      _RedistributeAncillaryFiles(self)
3099
      self.context.ReaddNode(new_node)
3100
      # make sure we redistribute the config
3101
      self.cfg.Update(new_node, feedback_fn)
3102
      # and make sure the new node will not have old files around
3103
      if not new_node.master_candidate:
3104
        result = self.rpc.call_node_demote_from_mc(new_node.name)
3105
        msg = result.fail_msg
3106
        if msg:
3107
          self.LogWarning("Node failed to demote itself from master"
3108
                          " candidate status: %s" % msg)
3109
    else:
3110
      _RedistributeAncillaryFiles(self, additional_nodes=[node])
3111
      self.context.AddNode(new_node, self.proc.GetECId())
3112

    
3113

    
3114
class LUSetNodeParams(LogicalUnit):
3115
  """Modifies the parameters of a node.
3116

3117
  """
3118
  HPATH = "node-modify"
3119
  HTYPE = constants.HTYPE_NODE
3120
  _OP_REQP = ["node_name"]
3121
  REQ_BGL = False
3122

    
3123
  def CheckArguments(self):
3124
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
3125
    if node_name is None:
3126
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name,
3127
                                 errors.ECODE_INVAL)
3128
    self.op.node_name = node_name
3129
    _CheckBooleanOpField(self.op, 'master_candidate')
3130
    _CheckBooleanOpField(self.op, 'offline')
3131
    _CheckBooleanOpField(self.op, 'drained')
3132
    all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
3133
    if all_mods.count(None) == 3:
3134
      raise errors.OpPrereqError("Please pass at least one modification",
3135
                                 errors.ECODE_INVAL)
3136
    if all_mods.count(True) > 1:
3137
      raise errors.OpPrereqError("Can't set the node into more than one"
3138
                                 " state at the same time",
3139
                                 errors.ECODE_INVAL)
3140

    
3141
  def ExpandNames(self):
3142
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
3143

    
3144
  def BuildHooksEnv(self):
3145
    """Build hooks env.
3146

3147
    This runs on the master node.
3148

3149
    """
3150
    env = {
3151
      "OP_TARGET": self.op.node_name,
3152
      "MASTER_CANDIDATE": str(self.op.master_candidate),
3153
      "OFFLINE": str(self.op.offline),
3154
      "DRAINED": str(self.op.drained),
3155
      }
3156
    nl = [self.cfg.GetMasterNode(),
3157
          self.op.node_name]
3158
    return env, nl, nl
3159

    
3160
  def CheckPrereq(self):
3161
    """Check prerequisites.
3162

3163
    This only checks the instance list against the existing names.
3164

3165
    """
3166
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
3167

    
3168
    if (self.op.master_candidate is not None or
3169
        self.op.drained is not None or
3170
        self.op.offline is not None):
3171
      # we can't change the master's node flags
3172
      if self.op.node_name == self.cfg.GetMasterNode():
3173
        raise errors.OpPrereqError("The master role can be changed"
3174
                                   " only via masterfailover",
3175
                                   errors.ECODE_INVAL)
3176

    
3177
    # Boolean value that tells us whether we're offlining or draining the node
3178
    offline_or_drain = self.op.offline == True or self.op.drained == True
3179
    deoffline_or_drain = self.op.offline == False or self.op.drained == False
3180

    
3181
    if (node.master_candidate and
3182
        (self.op.master_candidate == False or offline_or_drain)):
3183
      cp_size = self.cfg.GetClusterInfo().candidate_pool_size
3184
      mc_now, mc_should, mc_max = self.cfg.GetMasterCandidateStats()
3185
      if mc_now <= cp_size:
3186
        msg = ("Not enough master candidates (desired"
3187
               " %d, new value will be %d)" % (cp_size, mc_now-1))
3188
        # Only allow forcing the operation if it's an offline/drain operation,
3189
        # and we could not possibly promote more nodes.
3190
        # FIXME: this can still lead to issues if in any way another node which
3191
        # could be promoted appears in the meantime.
3192
        if self.op.force and offline_or_drain and mc_should == mc_max:
3193
          self.LogWarning(msg)
3194
        else:
3195
          raise errors.OpPrereqError(msg, errors.ECODE_INVAL)
3196

    
3197
    if (self.op.master_candidate == True and
3198
        ((node.offline and not self.op.offline == False) or
3199
         (node.drained and not self.op.drained == False))):
3200
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
3201
                                 " to master_candidate" % node.name,
3202
                                 errors.ECODE_INVAL)
3203

    
3204
    # If we're being deofflined/drained, we'll MC ourself if needed
3205
    if (deoffline_or_drain and not offline_or_drain and not
3206
        self.op.master_candidate == True):
3207
      self.op.master_candidate = _DecideSelfPromotion(self)
3208
      if self.op.master_candidate:
3209
        self.LogInfo("Autopromoting node to master candidate")
3210

    
3211
    return
3212

    
3213
  def Exec(self, feedback_fn):
3214
    """Modifies a node.
3215

3216
    """
3217
    node = self.node
3218

    
3219
    result = []
3220
    changed_mc = False
3221

    
3222
    if self.op.offline is not None:
3223
      node.offline = self.op.offline
3224
      result.append(("offline", str(self.op.offline)))
3225
      if self.op.offline == True:
3226
        if node.master_candidate:
3227
          node.master_candidate = False
3228
          changed_mc = True
3229
          result.append(("master_candidate", "auto-demotion due to offline"))
3230
        if node.drained:
3231
          node.drained = False
3232
          result.append(("drained", "clear drained status due to offline"))
3233

    
3234
    if self.op.master_candidate is not None:
3235
      node.master_candidate = self.op.master_candidate
3236
      changed_mc = True
3237
      result.append(("master_candidate", str(self.op.master_candidate)))
3238
      if self.op.master_candidate == False:
3239
        rrc = self.rpc.call_node_demote_from_mc(node.name)
3240
        msg = rrc.fail_msg
3241
        if msg:
3242
          self.LogWarning("Node failed to demote itself: %s" % msg)
3243

    
3244
    if self.op.drained is not None:
3245
      node.drained = self.op.drained
3246
      result.append(("drained", str(self.op.drained)))
3247
      if self.op.drained == True:
3248
        if node.master_candidate:
3249
          node.master_candidate = False
3250
          changed_mc = True
3251
          result.append(("master_candidate", "auto-demotion due to drain"))
3252
          rrc = self.rpc.call_node_demote_from_mc(node.name)
3253
          msg = rrc.fail_msg
3254
          if msg:
3255
            self.LogWarning("Node failed to demote itself: %s" % msg)
3256
        if node.offline:
3257
          node.offline = False
3258
          result.append(("offline", "clear offline status due to drain"))
3259

    
3260
    # this will trigger configuration file update, if needed
3261
    self.cfg.Update(node, feedback_fn)
3262
    # this will trigger job queue propagation or cleanup
3263
    if changed_mc:
3264
      self.context.ReaddNode(node)
3265

    
3266
    return result
3267

    
3268

    
3269
class LUPowercycleNode(NoHooksLU):
3270
  """Powercycles a node.
3271

3272
  """
3273
  _OP_REQP = ["node_name", "force"]
3274
  REQ_BGL = False
3275

    
3276
  def CheckArguments(self):
3277
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
3278
    if node_name is None:
3279
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name,
3280
                                 errors.ECODE_NOENT)
3281
    self.op.node_name = node_name
3282
    if node_name == self.cfg.GetMasterNode() and not self.op.force:
3283
      raise errors.OpPrereqError("The node is the master and the force"
3284
                                 " parameter was not set",
3285
                                 errors.ECODE_INVAL)
3286

    
3287
  def ExpandNames(self):
3288
    """Locking for PowercycleNode.
3289

3290
    This is a last-resort option and shouldn't block on other
3291
    jobs. Therefore, we grab no locks.
3292

3293
    """
3294
    self.needed_locks = {}
3295

    
3296
  def CheckPrereq(self):
3297
    """Check prerequisites.
3298

3299
    This LU has no prereqs.
3300

3301
    """
3302
    pass
3303

    
3304
  def Exec(self, feedback_fn):
3305
    """Reboots a node.
3306

3307
    """
3308
    result = self.rpc.call_node_powercycle(self.op.node_name,
3309
                                           self.cfg.GetHypervisorType())
3310
    result.Raise("Failed to schedule the reboot")
3311
    return result.payload
3312

    
3313

    
3314
class LUQueryClusterInfo(NoHooksLU):
3315
  """Query cluster configuration.
3316

3317
  """
3318
  _OP_REQP = []
3319
  REQ_BGL = False
3320

    
3321
  def ExpandNames(self):
3322
    self.needed_locks = {}
3323

    
3324
  def CheckPrereq(self):
3325
    """No prerequsites needed for this LU.
3326

3327
    """
3328
    pass
3329

    
3330
  def Exec(self, feedback_fn):
3331
    """Return cluster config.
3332

3333
    """
3334
    cluster = self.cfg.GetClusterInfo()
3335
    result = {
3336
      "software_version": constants.RELEASE_VERSION,
3337
      "protocol_version": constants.PROTOCOL_VERSION,
3338
      "config_version": constants.CONFIG_VERSION,
3339
      "os_api_version": max(constants.OS_API_VERSIONS),
3340
      "export_version": constants.EXPORT_VERSION,
3341
      "architecture": (platform.architecture()[0], platform.machine()),
3342
      "name": cluster.cluster_name,
3343
      "master": cluster.master_node,
3344
      "default_hypervisor": cluster.enabled_hypervisors[0],
3345
      "enabled_hypervisors": cluster.enabled_hypervisors,
3346
      "hvparams": dict([(hypervisor_name, cluster.hvparams[hypervisor_name])
3347
                        for hypervisor_name in cluster.enabled_hypervisors]),
3348
      "beparams": cluster.beparams,
3349
      "nicparams": cluster.nicparams,
3350
      "candidate_pool_size": cluster.candidate_pool_size,
3351
      "master_netdev": cluster.master_netdev,
3352
      "volume_group_name": cluster.volume_group_name,
3353
      "file_storage_dir": cluster.file_storage_dir,
3354
      "ctime": cluster.ctime,
3355
      "mtime": cluster.mtime,
3356
      "uuid": cluster.uuid,
3357
      "tags": list(cluster.GetTags()),
3358
      }
3359

    
3360
    return result
3361

    
3362

    
3363
class LUQueryConfigValues(NoHooksLU):
3364
  """Return configuration values.
3365

3366
  """
3367
  _OP_REQP = []
3368
  REQ_BGL = False
3369
  _FIELDS_DYNAMIC = utils.FieldSet()
3370
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag",
3371
                                  "watcher_pause")
3372

    
3373
  def ExpandNames(self):
3374
    self.needed_locks = {}
3375

    
3376
    _CheckOutputFields(static=self._FIELDS_STATIC,
3377
                       dynamic=self._FIELDS_DYNAMIC,
3378
                       selected=self.op.output_fields)
3379

    
3380
  def CheckPrereq(self):
3381
    """No prerequisites.
3382

3383
    """
3384
    pass
3385

    
3386
  def Exec(self, feedback_fn):
3387
    """Dump a representation of the cluster config to the standard output.
3388

3389
    """
3390
    values = []
3391
    for field in self.op.output_fields:
3392
      if field == "cluster_name":
3393
        entry = self.cfg.GetClusterName()
3394
      elif field == "master_node":
3395
        entry = self.cfg.GetMasterNode()
3396
      elif field == "drain_flag":
3397
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
3398
      elif field == "watcher_pause":
3399
        return utils.ReadWatcherPauseFile(constants.WATCHER_PAUSEFILE)
3400
      else:
3401
        raise errors.ParameterError(field)
3402
      values.append(entry)
3403
    return values
3404

    
3405

    
3406
class LUActivateInstanceDisks(NoHooksLU):
3407
  """Bring up an instance's disks.
3408

3409
  """
3410
  _OP_REQP = ["instance_name"]
3411
  REQ_BGL = False
3412

    
3413
  def ExpandNames(self):
3414
    self._ExpandAndLockInstance()
3415
    self.needed_locks[locking.LEVEL_NODE] = []
3416
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3417

    
3418
  def DeclareLocks(self, level):
3419
    if level == locking.LEVEL_NODE:
3420
      self._LockInstancesNodes()
3421

    
3422
  def CheckPrereq(self):
3423
    """Check prerequisites.
3424

3425
    This checks that the instance is in the cluster.
3426

3427
    """
3428
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3429
    assert self.instance is not None, \
3430
      "Cannot retrieve locked instance %s" % self.op.instance_name
3431
    _CheckNodeOnline(self, self.instance.primary_node)
3432
    if not hasattr(self.op, "ignore_size"):
3433
      self.op.ignore_size = False
3434

    
3435
  def Exec(self, feedback_fn):
3436
    """Activate the disks.
3437

3438
    """
3439
    disks_ok, disks_info = \
3440
              _AssembleInstanceDisks(self, self.instance,
3441
                                     ignore_size=self.op.ignore_size)
3442
    if not disks_ok:
3443
      raise errors.OpExecError("Cannot activate block devices")
3444

    
3445
    return disks_info
3446

    
3447

    
3448
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False,
3449
                           ignore_size=False):
3450
  """Prepare the block devices for an instance.
3451

3452
  This sets up the block devices on all nodes.
3453

3454
  @type lu: L{LogicalUnit}
3455
  @param lu: the logical unit on whose behalf we execute
3456
  @type instance: L{objects.Instance}
3457
  @param instance: the instance for whose disks we assemble
3458
  @type ignore_secondaries: boolean
3459
  @param ignore_secondaries: if true, errors on secondary nodes
3460
      won't result in an error return from the function
3461
  @type ignore_size: boolean
3462
  @param ignore_size: if true, the current known size of the disk
3463
      will not be used during the disk activation, useful for cases
3464
      when the size is wrong
3465
  @return: False if the operation failed, otherwise a list of
3466
      (host, instance_visible_name, node_visible_name)
3467
      with the mapping from node devices to instance devices
3468

3469
  """
3470
  device_info = []
3471
  disks_ok = True
3472
  iname = instance.name
3473
  # With the two passes mechanism we try to reduce the window of
3474
  # opportunity for the race condition of switching DRBD to primary
3475
  # before handshaking occured, but we do not eliminate it
3476

    
3477
  # The proper fix would be to wait (with some limits) until the
3478
  # connection has been made and drbd transitions from WFConnection
3479
  # into any other network-connected state (Connected, SyncTarget,
3480
  # SyncSource, etc.)
3481

    
3482
  # 1st pass, assemble on all nodes in secondary mode
3483
  for inst_disk in instance.disks:
3484
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3485
      if ignore_size:
3486
        node_disk = node_disk.Copy()
3487
        node_disk.UnsetSize()
3488
      lu.cfg.SetDiskID(node_disk, node)
3489
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
3490
      msg = result.fail_msg
3491
      if msg:
3492
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
3493
                           " (is_primary=False, pass=1): %s",
3494
                           inst_disk.iv_name, node, msg)
3495
        if not ignore_secondaries:
3496
          disks_ok = False
3497

    
3498
  # FIXME: race condition on drbd migration to primary
3499

    
3500
  # 2nd pass, do only the primary node
3501
  for inst_disk in instance.disks:
3502
    dev_path = None
3503

    
3504
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
3505
      if node != instance.primary_node:
3506
        continue
3507
      if ignore_size:
3508
        node_disk = node_disk.Copy()
3509
        node_disk.UnsetSize()
3510
      lu.cfg.SetDiskID(node_disk, node)
3511
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
3512
      msg = result.fail_msg
3513
      if msg:
3514
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
3515
                           " (is_primary=True, pass=2): %s",
3516
                           inst_disk.iv_name, node, msg)
3517
        disks_ok = False
3518
      else:
3519
        dev_path = result.payload
3520

    
3521
    device_info.append((instance.primary_node, inst_disk.iv_name, dev_path))
3522

    
3523
  # leave the disks configured for the primary node
3524
  # this is a workaround that would be fixed better by
3525
  # improving the logical/physical id handling
3526
  for disk in instance.disks:
3527
    lu.cfg.SetDiskID(disk, instance.primary_node)
3528

    
3529
  return disks_ok, device_info
3530

    
3531

    
3532
def _StartInstanceDisks(lu, instance, force):
3533
  """Start the disks of an instance.
3534

3535
  """
3536
  disks_ok, _ = _AssembleInstanceDisks(lu, instance,
3537
                                           ignore_secondaries=force)
3538
  if not disks_ok:
3539
    _ShutdownInstanceDisks(lu, instance)
3540
    if force is not None and not force:
3541
      lu.proc.LogWarning("", hint="If the message above refers to a"
3542
                         " secondary node,"
3543
                         " you can retry the operation using '--force'.")
3544
    raise errors.OpExecError("Disk consistency error")
3545

    
3546

    
3547
class LUDeactivateInstanceDisks(NoHooksLU):
3548
  """Shutdown an instance's disks.
3549

3550
  """
3551
  _OP_REQP = ["instance_name"]
3552
  REQ_BGL = False
3553

    
3554
  def ExpandNames(self):
3555
    self._ExpandAndLockInstance()
3556
    self.needed_locks[locking.LEVEL_NODE] = []
3557
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3558

    
3559
  def DeclareLocks(self, level):
3560
    if level == locking.LEVEL_NODE:
3561
      self._LockInstancesNodes()
3562

    
3563
  def CheckPrereq(self):
3564
    """Check prerequisites.
3565

3566
    This checks that the instance is in the cluster.
3567

3568
    """
3569
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3570
    assert self.instance is not None, \
3571
      "Cannot retrieve locked instance %s" % self.op.instance_name
3572

    
3573
  def Exec(self, feedback_fn):
3574
    """Deactivate the disks
3575

3576
    """
3577
    instance = self.instance
3578
    _SafeShutdownInstanceDisks(self, instance)
3579

    
3580

    
3581
def _SafeShutdownInstanceDisks(lu, instance):
3582
  """Shutdown block devices of an instance.
3583

3584
  This function checks if an instance is running, before calling
3585
  _ShutdownInstanceDisks.
3586

3587
  """
3588
  pnode = instance.primary_node
3589
  ins_l = lu.rpc.call_instance_list([pnode], [instance.hypervisor])[pnode]
3590
  ins_l.Raise("Can't contact node %s" % pnode)
3591

    
3592
  if instance.name in ins_l.payload:
3593
    raise errors.OpExecError("Instance is running, can't shutdown"
3594
                             " block devices.")
3595

    
3596
  _ShutdownInstanceDisks(lu, instance)
3597

    
3598

    
3599
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
3600
  """Shutdown block devices of an instance.
3601

3602
  This does the shutdown on all nodes of the instance.
3603

3604
  If the ignore_primary is false, errors on the primary node are
3605
  ignored.
3606

3607
  """
3608
  all_result = True
3609
  for disk in instance.disks:
3610
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
3611
      lu.cfg.SetDiskID(top_disk, node)
3612
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
3613
      msg = result.fail_msg
3614
      if msg:
3615
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
3616
                      disk.iv_name, node, msg)
3617
        if not ignore_primary or node != instance.primary_node:
3618
          all_result = False
3619
  return all_result
3620

    
3621

    
3622
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
3623
  """Checks if a node has enough free memory.
3624

3625
  This function check if a given node has the needed amount of free
3626
  memory. In case the node has less memory or we cannot get the
3627
  information from the node, this function raise an OpPrereqError
3628
  exception.
3629

3630
  @type lu: C{LogicalUnit}
3631
  @param lu: a logical unit from which we get configuration data
3632
  @type node: C{str}
3633
  @param node: the node to check
3634
  @type reason: C{str}
3635
  @param reason: string to use in the error message
3636
  @type requested: C{int}
3637
  @param requested: the amount of memory in MiB to check for
3638
  @type hypervisor_name: C{str}
3639
  @param hypervisor_name: the hypervisor to ask for memory stats
3640
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
3641
      we cannot check the node
3642

3643
  """
3644
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
3645
  nodeinfo[node].Raise("Can't get data from node %s" % node,
3646
                       prereq=True, ecode=errors.ECODE_ENVIRON)
3647
  free_mem = nodeinfo[node].payload.get('memory_free', None)
3648
  if not isinstance(free_mem, int):
3649
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
3650
                               " was '%s'" % (node, free_mem),
3651
                               errors.ECODE_ENVIRON)
3652
  if requested > free_mem:
3653
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
3654
                               " needed %s MiB, available %s MiB" %
3655
                               (node, reason, requested, free_mem),
3656
                               errors.ECODE_NORES)
3657

    
3658

    
3659
class LUStartupInstance(LogicalUnit):
3660
  """Starts an instance.
3661

3662
  """
3663
  HPATH = "instance-start"
3664
  HTYPE = constants.HTYPE_INSTANCE
3665
  _OP_REQP = ["instance_name", "force"]
3666
  REQ_BGL = False
3667

    
3668
  def ExpandNames(self):
3669
    self._ExpandAndLockInstance()
3670

    
3671
  def BuildHooksEnv(self):
3672
    """Build hooks env.
3673

3674
    This runs on master, primary and secondary nodes of the instance.
3675

3676
    """
3677
    env = {
3678
      "FORCE": self.op.force,
3679
      }
3680
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3681
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3682
    return env, nl, nl
3683

    
3684
  def CheckPrereq(self):
3685
    """Check prerequisites.
3686

3687
    This checks that the instance is in the cluster.
3688

3689
    """
3690
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3691
    assert self.instance is not None, \
3692
      "Cannot retrieve locked instance %s" % self.op.instance_name
3693

    
3694
    # extra beparams
3695
    self.beparams = getattr(self.op, "beparams", {})
3696
    if self.beparams:
3697
      if not isinstance(self.beparams, dict):
3698
        raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
3699
                                   " dict" % (type(self.beparams), ),
3700
                                   errors.ECODE_INVAL)
3701
      # fill the beparams dict
3702
      utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
3703
      self.op.beparams = self.beparams
3704

    
3705
    # extra hvparams
3706
    self.hvparams = getattr(self.op, "hvparams", {})
3707
    if self.hvparams:
3708
      if not isinstance(self.hvparams, dict):
3709
        raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
3710
                                   " dict" % (type(self.hvparams), ),
3711
                                   errors.ECODE_INVAL)
3712

    
3713
      # check hypervisor parameter syntax (locally)
3714
      cluster = self.cfg.GetClusterInfo()
3715
      utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
3716
      filled_hvp = objects.FillDict(cluster.hvparams[instance.hypervisor],
3717
                                    instance.hvparams)
3718
      filled_hvp.update(self.hvparams)
3719
      hv_type = hypervisor.GetHypervisor(instance.hypervisor)
3720
      hv_type.CheckParameterSyntax(filled_hvp)
3721
      _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
3722
      self.op.hvparams = self.hvparams
3723

    
3724
    _CheckNodeOnline(self, instance.primary_node)
3725

    
3726
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3727
    # check bridges existence
3728
    _CheckInstanceBridgesExist(self, instance)
3729

    
3730
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3731
                                              instance.name,
3732
                                              instance.hypervisor)
3733
    remote_info.Raise("Error checking node %s" % instance.primary_node,
3734
                      prereq=True, ecode=errors.ECODE_ENVIRON)
3735
    if not remote_info.payload: # not running already
3736
      _CheckNodeFreeMemory(self, instance.primary_node,
3737
                           "starting instance %s" % instance.name,
3738
                           bep[constants.BE_MEMORY], instance.hypervisor)
3739

    
3740
  def Exec(self, feedback_fn):
3741
    """Start the instance.
3742

3743
    """
3744
    instance = self.instance
3745
    force = self.op.force
3746

    
3747
    self.cfg.MarkInstanceUp(instance.name)
3748

    
3749
    node_current = instance.primary_node
3750

    
3751
    _StartInstanceDisks(self, instance, force)
3752

    
3753
    result = self.rpc.call_instance_start(node_current, instance,
3754
                                          self.hvparams, self.beparams)
3755
    msg = result.fail_msg
3756
    if msg:
3757
      _ShutdownInstanceDisks(self, instance)
3758
      raise errors.OpExecError("Could not start instance: %s" % msg)
3759

    
3760

    
3761
class LURebootInstance(LogicalUnit):
3762
  """Reboot an instance.
3763

3764
  """
3765
  HPATH = "instance-reboot"
3766
  HTYPE = constants.HTYPE_INSTANCE
3767
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
3768
  REQ_BGL = False
3769

    
3770
  def CheckArguments(self):
3771
    """Check the arguments.
3772

3773
    """
3774
    self.shutdown_timeout = getattr(self.op, "shutdown_timeout",
3775
                                    constants.DEFAULT_SHUTDOWN_TIMEOUT)
3776

    
3777
  def ExpandNames(self):
3778
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
3779
                                   constants.INSTANCE_REBOOT_HARD,
3780
                                   constants.INSTANCE_REBOOT_FULL]:
3781
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
3782
                                  (constants.INSTANCE_REBOOT_SOFT,
3783
                                   constants.INSTANCE_REBOOT_HARD,
3784
                                   constants.INSTANCE_REBOOT_FULL))
3785
    self._ExpandAndLockInstance()
3786

    
3787
  def BuildHooksEnv(self):
3788
    """Build hooks env.
3789

3790
    This runs on master, primary and secondary nodes of the instance.
3791

3792
    """
3793
    env = {
3794
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
3795
      "REBOOT_TYPE": self.op.reboot_type,
3796
      "SHUTDOWN_TIMEOUT": self.shutdown_timeout,
3797
      }
3798
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3799
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3800
    return env, nl, nl
3801

    
3802
  def CheckPrereq(self):
3803
    """Check prerequisites.
3804

3805
    This checks that the instance is in the cluster.
3806

3807
    """
3808
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3809
    assert self.instance is not None, \
3810
      "Cannot retrieve locked instance %s" % self.op.instance_name
3811

    
3812
    _CheckNodeOnline(self, instance.primary_node)
3813

    
3814
    # check bridges existence
3815
    _CheckInstanceBridgesExist(self, instance)
3816

    
3817
  def Exec(self, feedback_fn):
3818
    """Reboot the instance.
3819

3820
    """
3821
    instance = self.instance
3822
    ignore_secondaries = self.op.ignore_secondaries
3823
    reboot_type = self.op.reboot_type
3824

    
3825
    node_current = instance.primary_node
3826

    
3827
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
3828
                       constants.INSTANCE_REBOOT_HARD]:
3829
      for disk in instance.disks:
3830
        self.cfg.SetDiskID(disk, node_current)
3831
      result = self.rpc.call_instance_reboot(node_current, instance,
3832
                                             reboot_type,
3833
                                             self.shutdown_timeout)
3834
      result.Raise("Could not reboot instance")
3835
    else:
3836
      result = self.rpc.call_instance_shutdown(node_current, instance,
3837
                                               self.shutdown_timeout)
3838
      result.Raise("Could not shutdown instance for full reboot")
3839
      _ShutdownInstanceDisks(self, instance)
3840
      _StartInstanceDisks(self, instance, ignore_secondaries)
3841
      result = self.rpc.call_instance_start(node_current, instance, None, None)
3842
      msg = result.fail_msg
3843
      if msg:
3844
        _ShutdownInstanceDisks(self, instance)
3845
        raise errors.OpExecError("Could not start instance for"
3846
                                 " full reboot: %s" % msg)
3847

    
3848
    self.cfg.MarkInstanceUp(instance.name)
3849

    
3850

    
3851
class LUShutdownInstance(LogicalUnit):
3852
  """Shutdown an instance.
3853

3854
  """
3855
  HPATH = "instance-stop"
3856
  HTYPE = constants.HTYPE_INSTANCE
3857
  _OP_REQP = ["instance_name"]
3858
  REQ_BGL = False
3859

    
3860
  def CheckArguments(self):
3861
    """Check the arguments.
3862

3863
    """
3864
    self.timeout = getattr(self.op, "timeout",
3865
                           constants.DEFAULT_SHUTDOWN_TIMEOUT)
3866

    
3867
  def ExpandNames(self):
3868
    self._ExpandAndLockInstance()
3869

    
3870
  def BuildHooksEnv(self):
3871
    """Build hooks env.
3872

3873
    This runs on master, primary and secondary nodes of the instance.
3874

3875
    """
3876
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3877
    env["TIMEOUT"] = self.timeout
3878
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3879
    return env, nl, nl
3880

    
3881
  def CheckPrereq(self):
3882
    """Check prerequisites.
3883

3884
    This checks that the instance is in the cluster.
3885

3886
    """
3887
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3888
    assert self.instance is not None, \
3889
      "Cannot retrieve locked instance %s" % self.op.instance_name
3890
    _CheckNodeOnline(self, self.instance.primary_node)
3891

    
3892
  def Exec(self, feedback_fn):
3893
    """Shutdown the instance.
3894

3895
    """
3896
    instance = self.instance
3897
    node_current = instance.primary_node
3898
    timeout = self.timeout
3899
    self.cfg.MarkInstanceDown(instance.name)
3900
    result = self.rpc.call_instance_shutdown(node_current, instance, timeout)
3901
    msg = result.fail_msg
3902
    if msg:
3903
      self.proc.LogWarning("Could not shutdown instance: %s" % msg)
3904

    
3905
    _ShutdownInstanceDisks(self, instance)
3906

    
3907

    
3908
class LUReinstallInstance(LogicalUnit):
3909
  """Reinstall an instance.
3910

3911
  """
3912
  HPATH = "instance-reinstall"
3913
  HTYPE = constants.HTYPE_INSTANCE
3914
  _OP_REQP = ["instance_name"]
3915
  REQ_BGL = False
3916

    
3917
  def ExpandNames(self):
3918
    self._ExpandAndLockInstance()
3919

    
3920
  def BuildHooksEnv(self):
3921
    """Build hooks env.
3922

3923
    This runs on master, primary and secondary nodes of the instance.
3924

3925
    """
3926
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3927
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3928
    return env, nl, nl
3929

    
3930
  def CheckPrereq(self):
3931
    """Check prerequisites.
3932

3933
    This checks that the instance is in the cluster and is not running.
3934

3935
    """
3936
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3937
    assert instance is not None, \
3938
      "Cannot retrieve locked instance %s" % self.op.instance_name
3939
    _CheckNodeOnline(self, instance.primary_node)
3940

    
3941
    if instance.disk_template == constants.DT_DISKLESS:
3942
      raise errors.OpPrereqError("Instance '%s' has no disks" %
3943
                                 self.op.instance_name,
3944
                                 errors.ECODE_INVAL)
3945
    if instance.admin_up:
3946
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3947
                                 self.op.instance_name,
3948
                                 errors.ECODE_STATE)
3949
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3950
                                              instance.name,
3951
                                              instance.hypervisor)
3952
    remote_info.Raise("Error checking node %s" % instance.primary_node,
3953
                      prereq=True, ecode=errors.ECODE_ENVIRON)
3954
    if remote_info.payload:
3955
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3956
                                 (self.op.instance_name,
3957
                                  instance.primary_node),
3958
                                 errors.ECODE_STATE)
3959

    
3960
    self.op.os_type = getattr(self.op, "os_type", None)
3961
    self.op.force_variant = getattr(self.op, "force_variant", False)
3962
    if self.op.os_type is not None:
3963
      # OS verification
3964
      pnode = self.cfg.GetNodeInfo(
3965
        self.cfg.ExpandNodeName(instance.primary_node))
3966
      if pnode is None:
3967
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
3968
                                   self.op.pnode, errors.ECODE_NOENT)
3969
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
3970
      result.Raise("OS '%s' not in supported OS list for primary node %s" %
3971
                   (self.op.os_type, pnode.name),
3972
                   prereq=True, ecode=errors.ECODE_INVAL)
3973
      if not self.op.force_variant:
3974
        _CheckOSVariant(result.payload, self.op.os_type)
3975

    
3976
    self.instance = instance
3977

    
3978
  def Exec(self, feedback_fn):
3979
    """Reinstall the instance.
3980

3981
    """
3982
    inst = self.instance
3983

    
3984
    if self.op.os_type is not None:
3985
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
3986
      inst.os = self.op.os_type
3987
      self.cfg.Update(inst, feedback_fn)
3988

    
3989
    _StartInstanceDisks(self, inst, None)
3990
    try:
3991
      feedback_fn("Running the instance OS create scripts...")
3992
      result = self.rpc.call_instance_os_add(inst.primary_node, inst, True)
3993
      result.Raise("Could not install OS for instance %s on node %s" %
3994
                   (inst.name, inst.primary_node))
3995
    finally:
3996
      _ShutdownInstanceDisks(self, inst)
3997

    
3998

    
3999
class LURecreateInstanceDisks(LogicalUnit):
4000
  """Recreate an instance's missing disks.
4001

4002
  """
4003
  HPATH = "instance-recreate-disks"
4004
  HTYPE = constants.HTYPE_INSTANCE
4005
  _OP_REQP = ["instance_name", "disks"]
4006
  REQ_BGL = False
4007

    
4008
  def CheckArguments(self):
4009
    """Check the arguments.
4010

4011